kern/v4/d2/lti/kbus_http/kbus_http_test.go

package kbus_http

import (
	"encoding/json"
	"io"
	"net/http"
	"os"
	"strings"
	"testing"
	"time"

	"gitp78su.ipnodns.ru/svi/kern/v4/d0"
	. "gitp78su.ipnodns.ru/svi/kern/v4/d1"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/kern_ctx"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/kserv_http"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/lti/bus_ent/topic"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/lti/bus_mock/mock_hand_serve"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/lti/bus_mock/mock_hand_sub_local"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/lti/dot"
)

type tester struct {
	t        *testing.T
	handSub  *mock_hand_sub_local.MockHandlerSub
	handServ *mock_hand_serve.MockHandlerServe
	owner    *d0.LOwner
}

const (
	test_uuid = "test_uuid"
)

var (
	qNameSub  = topic.LetTopic("topic_sub")
	qNameServ = topic.LetTopic("topic_serv")
)

func TestKernelBusHttp(t *testing.T) {
	owner := d0.LetOwner(1, "test_kernel_bus_http")
	sf := &tester{
		owner: owner,
		t:     t,
		handSub: mock_hand_sub_local.NewMockHandlerSub(qNameSub,
			d0.LetTxt("http://localhost:18200/bus/pub")),
		handServ: mock_hand_serve.NewMockHandlerServe(qNameServ, "name_serv"),
	}
	ctx := kern_ctx.GetKernCtx()
	ctx.Set(d0.LetTxt("monolitName"), "test_monolit", "comment")
	sf.get()
	sf.req()
	sf.sub()
	sf.pub()
	sf.unsub()
}

// Запрос на отписку.
func (sf *tester) unsub() {
	sf.t.Log("unsub")
	sf.unsubBad1()
	sf.unsubBad2()
	sf.unsubGood1()
	sf.unsubBad3()
	sf.unsubGood2()
}

func (sf *tester) unsubGood2() {
	sf.t.Log("unsubGood2")
	res := Bus_.Subscribe(sf.handSub)
	if res.IsErr() {
		sf.t.Fatalf("unsubGood1(): err=%v", res.Err())
	}
	req := &dot.UnsubReq{
		HandlerName_: sf.handSub.Name().Get(),
		Uuid_:        d0.LetTxt(test_uuid),
	}
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/unsub", body)
	if err != nil {
		sf.t.Fatalf("unsubGood2(): err=%v", err)
	}
	hReq.Header.Add("Content-Type", "application/json")

	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("unsubGood2(): after request, err=%v", err)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	if _resp.StatusCode != http.StatusOK {
		sf.t.Fatalf("unsubGood2(): statusCode(%v)!=200", _resp.StatusCode)
	}
	binBody, _ := io.ReadAll(_resp.Body)
	resp := &dot.UnsubResp{}
	err = json.Unmarshal(binBody, resp)
	if err != nil {
		sf.t.Fatalf("unsubGood2(): err=%v", err)
	}
	if resp.Status_.Get() != "ok" {
		sf.t.Fatalf("unsubGood2(): resp(%v)!='ok'", resp.Status_)
	}
}

// Кривой запрос.
func (sf *tester) unsubBad3() {
	sf.t.Log("unsubBad3")
	req := "tra-la-la"
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/unsub", body)
	hReq.Header.Add("Content-Type", "application/json")
	if err != nil {
		sf.t.Fatalf("unsubBad3(): err=%v", err)
	}
	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("unsubBad3(): after request, err=%v", err)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	if _resp.StatusCode == http.StatusOK {
		sf.t.Fatalf("unsubBad3(): statusCode(%v)==200", _resp.StatusCode)
	}
}

func (sf *tester) unsubGood1() {
	sf.t.Log("unsubGood1")
	res := Bus_.Subscribe(sf.handSub)
	if res.IsErr() {
		sf.t.Fatalf("unsubGood1(): err=%v", res.Err())
	}
	count := 0
	for count < 100 {
		time.Sleep(time.Millisecond * 1)
		count++
	}
	req := &dot.UnsubReq{
		HandlerName_: sf.handSub.Name().Get(),
		Uuid_:        d0.LetTxt(test_uuid),
	}
	resp := Bus_.processUnsubRequest(req)
	if resp.Status_.Get() != "ok" {
		sf.t.Fatalf("unsubGood1(): status(%v)!='ok'", resp.Status_)
	}
}

// Все поля на месте, нет подписчика.
func (sf *tester) unsubBad2() {
	sf.t.Log("unsubBad2")
	Bus_.Unsubscribe(sf.handSub)
	count := 0
	for count < 100 {
		time.Sleep(time.Millisecond * 1)
		count++
	}
	req := &dot.UnsubReq{
		HandlerName_: sf.handSub.Name().Get(),
		Uuid_:        d0.LetTxt(test_uuid),
	}
	resp := Bus_.processUnsubRequest(req)
	if resp.Status_.Get() == "ok" {
		sf.t.Fatalf("unsubBad2(): status(%v)=='ok'", resp.Status_)
	}
}

// Нет полей для процесса отписки.
func (sf *tester) unsubBad1() {
	sf.t.Log("unsubBad1")
	defer func() {
		if _panic := recover(); _panic == nil {
			sf.t.Fatalf("unsubBad1(): panic==nil")
		}
	}()
	req := &dot.UnsubReq{}
	_ = Bus_.processUnsubRequest(req)
}

// Запрос на публикацию.
func (sf *tester) pub() {
	sf.t.Log("pub")
	sf.pubBad1()
	sf.pubGood1()
	sf.pubBad2()
	sf.pubBad3()
	sf.pubGood2()
}

func (sf *tester) pubGood2() {
	sf.t.Log("pubGood2")
	req := &dot.PublishReq{
		Topic_:  qNameSub.Get(),
		Uuid_:   d0.LetTxt(test_uuid),
		BinMsg_: []byte("http_pub"),
	}
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/pub", body)
	hReq.Header.Add("Content-Type", "application/json")
	if err != nil {
		sf.t.Fatalf("pubGood2(): err=%v", err)
	}
	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("pubGood2(): after request, err=%v", err)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	if _resp.StatusCode != http.StatusOK {
		sf.t.Fatalf("pubGood2(): statusCode(%v)!=200", _resp.StatusCode)
	}
	binBody, _ := io.ReadAll(_resp.Body)
	resp := &dot.PublishResp{}
	err = json.Unmarshal(binBody, resp)
	if err != nil {
		sf.t.Fatalf("pubGood2(): err=%v", err)
	}
	if resp.Status_.Get() != "ok" {
		sf.t.Fatalf("pubGood2(): resp(%v)!='ok'", resp.Status_)
	}
}

// Кривой запрос.
func (sf *tester) pubBad3() {
	sf.t.Log("pubBad3")
	req := "tra-la-la"
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/pub", body)
	hReq.Header.Add("Content-Type", "application/json")
	if err != nil {
		sf.t.Fatalf("pubBad3(): err=%v", err)
	}
	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("pubBad3(): after request, err=%v", err)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	if _resp.StatusCode == http.StatusOK {
		sf.t.Fatalf("pubBad3(): statusCode(%v)==200", _resp.StatusCode)
	}
}

// Что-то случилось внутри шины.
func (sf *tester) pubBad2() {
	sf.t.Log("pubBad2")
	Bus_.IsWork_.Reset()
	defer Bus_.IsWork_.Set()

	req := &dot.PublishReq{
		Topic_:  qNameSub.Get(),
		Uuid_:   d0.LetTxt(test_uuid),
		BinMsg_: []byte("test_pub"),
	}
	resp := Bus_.processPublish(req)
	if resp.Status_.Get() == "ok" {
		sf.t.Fatalf("pubBad2(): status(%v)=='ok'", resp.Status_)
	}
}

// Все поля на месте.
func (sf *tester) pubGood1() {
	sf.t.Log("pubGood1")
	res := Bus_.Subscribe(sf.handSub)
	if res.IsErr() {
		sf.t.Fatalf("pubGood1(): err=%v", res.Err())
	}
	req := &dot.PublishReq{
		Topic_:  qNameSub.Get(),
		Uuid_:   d0.LetTxt(test_uuid),
		BinMsg_: []byte("test_pub"),
	}
	_ = Bus_.processPublish(req)
	for {
		time.Sleep(time.Millisecond * 1)
		msg := sf.handSub.Msg()
		if msg.Get() != "" {
			break
		}
	}
	msg := sf.handSub.Msg()
	if msg.Get() != "test_pub" {
		sf.t.Fatalf("pubGood1(): msg(%v)!='test_pub'", msg)
	}
}

// Нет полей для процесса публикации.
func (sf *tester) pubBad1() {
	sf.t.Log("pubBad1")
	defer func() {
		if _panic := recover(); _panic == nil {
			sf.t.Fatalf("pubBad1(): panic==nil")
		}
	}()
	req := &dot.PublishReq{}
	_ = Bus_.processPublish(req)
}

// Запрос подписки на топик.
func (sf *tester) sub() {
	sf.t.Log("sub")
	sf.subBad1()
	sf.subBad2()
	sf.subGood1()
	sf.subBad3()
	sf.subGood2()
}

// Полный процесс подписки.
func (sf *tester) subGood2() {
	sf.t.Log("subGood2")
	req := &dot.SubscribeReq{
		Topic_:   qNameServ.Get(),
		Uuid_:    d0.LetTxt(test_uuid),
		WebHook_: d0.LetTxt("http://localhost:18200/bus/pub/"),
	}
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/sub", body)
	hReq.Header.Add("Content-Type", "application/json")
	if err != nil {
		sf.t.Fatalf("subBad1(): err=%v", err)
	}
	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("subBad1(): after request, err=%v", err)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	if _resp.StatusCode != http.StatusOK {
		sf.t.Fatalf("subBad1(): statusCode(%v)!=200", _resp.StatusCode)
	}
	binBody, _ := io.ReadAll(_resp.Body)
	resp := &dot.SubscribeResp{}
	err = json.Unmarshal(binBody, resp)
	if err != nil {
		sf.t.Fatalf("subBad1(): err=%v", err)
	}
	if resp.Status_.Get() != "ok" {
		sf.t.Fatalf("subBad1(): resp(%v)!='ok'", resp.Status_)
	}
}

// Отключена базовая шина.
func (sf *tester) subBad3() {
	sf.t.Log("subBad3")
	req := &dot.SubscribeReq{
		Topic_:   qNameServ.Get(),
		Uuid_:    d0.LetTxt(test_uuid),
		WebHook_: d0.LetTxt("http://localhost:18200/bus/pub/"),
	}
	defer func() {
		if _panic := recover(); _panic != nil {
			sf.t.Fatalf("subBad3(): panic!=nil")
		}
	}()
	// _bus := kernel_bus_base.GetKernelBusBase()
	Bus_.IsWork_.Reset()
	defer Bus_.IsWork_.Set()
	resp := Bus_.processSubscribe(req)
	if resp.Status_.Get() == "ok" {
		sf.t.Fatalf("subBad3(): resp==ok")
	}
}

// Проверка полей запроса в процессе подписки.
func (sf *tester) subGood1() {
	sf.t.Log("subGood1")
	req := &dot.SubscribeReq{
		Topic_:   qNameServ.Get(),
		Uuid_:    d0.LetTxt(test_uuid),
		WebHook_: d0.LetTxt("http://localhost:18200/bus/"),
	}
	defer func() {
		if _panic := recover(); _panic != nil {
			sf.t.Fatalf("subGood1(): panic!=nil")
		}
	}()
	_ = Bus_.processSubscribe(req)
}

var (
	qNameEmpty = topic.LetTopic("")
)

// Проверка кривых полей запроса в процессе подписки.
func (sf *tester) subBad2() {
	sf.t.Log("subBad2")
	req := &dot.SubscribeReq{
		Topic_: qNameEmpty.Get(),
	}
	defer func() {
		if _panic := recover(); _panic == nil {
			sf.t.Fatalf("subBad2(): panic==nil")
		}
	}()
	_ = Bus_.processSubscribe(req)
}

// Кривой запрос.
func (sf *tester) subBad1() {
	sf.t.Log("subBad1")
	req := "tra-ta-ta"
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/sub", body)
	hReq.Header.Add("Content-Type", "application/json")
	if err != nil {
		sf.t.Fatalf("subBad1(): err=%v", err)
	}
	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("subBad1(): after request, err=%v", err)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	if _resp.StatusCode != http.StatusBadRequest {
		sf.t.Fatalf("subBad1(): statusCode(%v)!=400", _resp.StatusCode)
	}
}

// Входящий запрос.
func (sf *tester) req() {
	sf.t.Log("req")
	sf.reqBad1()
	sf.reqBad2()
	sf.reqBad3()
	sf.reqGood1()
	sf.reqBad4()
}

// Что-то с обработчиком.
func (sf *tester) reqBad4() {
	sf.t.Log("reqBad4")
	sf.handServ.IsBad_.Set()
	defer sf.handServ.IsBad_.Reset()
	req := &dot.ServeReq{
		Topic_:  sf.handServ.Topic().Get(),
		Uuid_:   d0.LetTxt(test_uuid),
		BinReq_: []byte("test_msg"),
	}
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/request", body)
	hReq.Header.Add("Content-Type", "application/json")
	if err != nil {
		sf.t.Fatalf("reqBad4(): err=%v", err)
	}
	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("reqBad4(): after request, err=%v", err)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	if _resp.StatusCode != http.StatusOK {
		sf.t.Fatalf("reqBad4(): statusCode(%v)!=200", _resp.StatusCode)
	}
	binBody, _ := io.ReadAll(_resp.Body)
	resp := &dot.ServeResp{}
	err = json.Unmarshal(binBody, resp)
	if err != nil {
		sf.t.Fatalf("reqBad4(): err=%v", err)
	}
	if resp.Status_.Get() == "ok" {
		sf.t.Fatalf("reqBad4(): status(%v)=='ok'", resp.Status_)
	}
}

func (sf *tester) reqGood1() {
	sf.t.Log("reqGood1")
	req := &dot.ServeReq{
		Topic_:  sf.handServ.Topic().Get(),
		Uuid_:   d0.LetTxt(test_uuid),
		BinReq_: []byte("test_msg"),
	}
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/request", body)
	hReq.Header.Add("Content-Type", "application/json")
	if err != nil {
		sf.t.Fatalf("reqGood1(): err=%v", err)
	}
	_resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("reqGood1(): after request, err=%v", err)
	}
	if _resp.StatusCode != http.StatusOK {
		sf.t.Fatalf("reqGood1(): statusCode(%v)!=200", _resp.StatusCode)
	}
	defer func() {
		_ = _resp.Body.Close()
	}()
	binBody, _ := io.ReadAll(_resp.Body)
	resp := &dot.ServeResp{}
	err = json.Unmarshal(binBody, resp)
	if err != nil {
		sf.t.Fatalf("reqGood1(): err=%v", err)
	}
	if d0.Txt(resp.BinResp_) != "test_msg" {
		sf.t.Fatalf("reqGood1(): resp(%v)!='test_msg'", resp.BinResp_)
	}
}

var (
	qNameBad = topic.LetTopic("bad_topic")
)

// Нет такого топика для запросов.
func (sf *tester) reqBad3() {
	sf.t.Log("reqBad3")
	req := &dot.ServeReq{
		Topic_:  qNameBad.Get(),
		Uuid_:   d0.LetTxt(test_uuid),
		BinReq_: []byte("test_msg"),
	}
	binReq, _ := json.MarshalIndent(req, "", "  ")
	body := strings.NewReader(string(binReq))
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/request", body)
	if err != nil {
		sf.t.Fatalf("reqBad3(): err=%v", err)
	}
	resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("reqBad3(): after request, err=%v", err)
	}
	defer func() {
		_ = resp.Body.Close()
	}()
	if resp.StatusCode != http.StatusBadRequest {
		sf.t.Fatalf("reqBad3(): statusCode(%v)!=400", resp.StatusCode)
	}
}

// Нет тела запроса.
func (sf *tester) reqBad2() {
	sf.t.Log("reqBad1")
	body := strings.NewReader("test_msg")
	kServHttp := kserv_http.GetKernelServHttp()
	fibApp := kServHttp.Fiber()
	hReq, err := http.NewRequest(http.MethodPost, "/bus/request", body)
	if err != nil {
		sf.t.Fatalf("reqBad1(): err=%v", err)
	}
	resp, err := fibApp.Test(hReq)
	if err != nil {
		sf.t.Fatalf("reqBad1(): after request, err=%v", err)
	}
	defer func() {
		_ = resp.Body.Close()
	}()
	if resp.StatusCode != http.StatusBadRequest {
		sf.t.Fatalf("reqBad1(): statusCode(%v)!=400", resp.StatusCode)
	}
}

// Отсутствуют поля в запросе.
func (sf *tester) reqBad1() {
	sf.t.Log("reqBad1")
	defer func() {
		if _panic := recover(); _panic == nil {
			sf.t.Fatalf("reqBad1(): panic==nil")
		}
	}()
	Bus_.processSendRequest(nil)
}

// Получает локальную шину.
func (sf *tester) get() {
	sf.t.Log("get")
	_ = MakeEnv(sf.owner)
	_ = os.Unsetenv("LOCAL_HTTP_URL")
	_ = os.Setenv("LOCAL_HTTP_URL", "http://localhost:18312/")
	_ = GetKernelBusHttp()
	if Bus_ == nil {
		sf.t.Fatalf("get(): bus==nil")
	}
	_ = GetKernelBusHttp()
	Bus_.RegisterServe(sf.handServ).Hassert("get()")
}