iotlcm/upstream-chan/internal/cmd/http_test.go

package cmd

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	_ "github.com/gogf/gf/contrib/drivers/mysql/v2"
	"github.com/gogf/gf/v2/database/gdb"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/ghttp"
	"github.com/gogf/gf/v2/os/gcfg"
	"github.com/gogf/gf/v2/os/gtime"
	"gopkg.in/yaml.v3"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/logic/upstreamapi"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/logic/upstreamimpl"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/model/entity"
	"git.5iot.com/iotlcm-public/iotlcm-upstream-chan/internal/service"
)

// TestAPNHTTPHarness 仅供跨模块测试子进程使用，不启动真实运营商或默认业务连接。
func TestAPNHTTPHarness(t *testing.T) {
	if os.Getenv("APN_HTTP_CHAIN_HELPER") != "1" {
		t.Skip("未启用本机HTTP模拟链")
	}
	body, err := os.ReadFile(filepath.Join("..", "..", "..", "iotlcm-sms", "deploy", "config.yaml"))
	if err != nil {
		t.Fatal("读取本机认证配置失败")
	}
	var source struct {
		DSN string `yaml:"sceneDbDsn"`
	}
	if yaml.Unmarshal(body, &source) != nil {
		t.Fatal("解析本机认证配置失败")
	}
	dsn, err := mysql.ParseDSN(source.DSN)
	if err != nil || dsn.Net != "tcp" || (dsn.Addr != "127.0.0.1:3306" && dsn.Addr != "localhost:3306") {
		t.Fatal("不是已授权本机连接")
	}
	node := gdb.ConfigNode{Type: "mysql", Host: "127.0.0.1", Port: "3306", User: dsn.User, Pass: dsn.Passwd, Name: "iotdb_apn_chan_test_20261015"}
	db, err := gdb.New(node)
	if err != nil {
		t.Fatal("创建专用测试连接失败")
	}
	identity, err := db.GetOne(context.Background(), "SELECT @@port AS port,@@server_uuid AS uuid,@@datadir AS datadir,DATABASE() AS dbname")
	if err != nil || identity["port"].String() != "3306" || identity["uuid"].String() != "4fa545af-ab36-11f1-be02-040e3c9de361" || identity["dbname"].String() != node.Name || strings.TrimRight(strings.ReplaceAll(identity["datadir"].String(), "\\", "/"), "/") != "D:/MySQL/MySQL Server 8.0/Data" {
		t.Fatal("专用测试实例身份核验失败")
	}
	if _, err := db.GetAll(context.Background(), "SELECT revision FROM t_apn_operation LIMIT 0"); err != nil {
		t.Fatal("先执行指定测试库迁移回归")
	}
	adapter, _ := gcfg.NewAdapterContent(`{}`)
	g.Cfg().SetAdapter(adapter)
	if err := gdb.SetConfig(gdb.Config{"default": gdb.ConfigGroup{node}}); err != nil {
		t.Fatal("设置专用测试连接失败")
	}
	var mu sync.Mutex
	writes, queries := map[string]int{}, map[string]int{}
	submits, unexpected := 0, 0
	jobs := map[string][]string{}
	operator := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		mu.Lock()
		defer mu.Unlock()
		if strings.HasSuffix(r.URL.Path, "/sim-communication-function/batch") {
			numbers := strings.Split(r.URL.Query().Get("msisdns"), "_")
			first := numbers[0]
			submits++
			for _, number := range numbers {
				writes[number]++
			}
			if strings.HasPrefix(first, "lost-") {
				conn, _, e := w.(http.Hijacker).Hijack()
				if e != nil {
					t.Error(e)
					return
				}
				_ = conn.Close()
				return
			}
			job := "fake-job-" + first
			jobs[job] = numbers
			_ = json.NewEncoder(w).Encode(g.Map{"status": "0", "result": []g.Map{{"jobId": job}}})
			return
		}
		if strings.HasSuffix(r.URL.Path, "/sim-batch-result") {
			job := r.URL.Query().Get("jobId")
			queries[job]++
			if queries[job] == 1 {
				_, _ = io.WriteString(w, `{"status":"14008","message":"尚未可查"}`)
				return
			}
			if queries[job] == 2 {
				_, _ = io.WriteString(w, `{"status":"0","result":[{"jobStatus":"1"}]}`)
				return
			}
			if strings.Contains(job, "failed-") {
				_, _ = io.WriteString(w, `{"status":"0","message":"明确失败","result":[{"jobStatus":"4"}]}`)
				return
			}
			cards := []g.Map{}
			for _, number := range jobs[job] {
				cards = append(cards, g.Map{"resultId": number, "status": "0", "message": "明确成功"})
			}
			_ = json.NewEncoder(w).Encode(g.Map{"status": "0", "result": []g.Map{{"jobStatus": "2", "resultList": cards}}})
			return
		}
		unexpected++
		t.Errorf("模拟运营商收到非预期请求：%s", r.URL.Path)
		http.Error(w, "非预期请求", 500)
	}))
	defer operator.Close()
	channel := upstreamimpl.NewSimYD().NewTemp(context.Background(), &model.SimChannelConfigItemCache{
		Template: &entity.TSimChannelTemplate{TempConfig: `{"api":{"baseUrl":"` + operator.URL + `/v5"},"http":{"timeout":"2s"}}`},
		Channel:  &entity.TSimChannels{ChanKey: "fake-channel", AppKey: "fake", AccessToken: "fake-token", AccessTokenExpire: gtime.New(time.Now().Add(time.Hour))},
	})
	register := func() {
		api := upstreamapi.NewUpstreamAPI()
		api.RegisterChan("fake-channel", "1", &model.SimCardChannelCache{Instance: channel})
		service.RegisterUpstreamAPI(api)
	}
	register()
	server := ghttp.GetServer(t.Name())
	server.SetAddr("127.0.0.1:0")
	bindHTTPRoutes(server)
	if err := server.Start(); err != nil {
		t.Fatal(err)
	}
	defer server.Shutdown()
	stop := make(chan struct{})
	var stopOnce sync.Once
	dropped := map[string]bool{}
	front := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.URL.Path {
		case "/__test/stats":
			mu.Lock()
			defer mu.Unlock()
			_ = json.NewEncoder(w).Encode(g.Map{"writes": writes, "queries": queries, "submits": submits, "unexpected": unexpected})
			return
		case "/__test/stop":
			stopOnce.Do(func() { close(stop) })
			return
		case "/__test/restart":
			register()
			return
		}
		raw, _ := io.ReadAll(r.Body)
		var input model.SimCardOperateChangeApnInput
		if json.Unmarshal(raw, &input) != nil || len(input.SimKeys) == 0 || input.SimKeys[0].ChanKey != "fake-channel" {
			http.Error(w, "只接受假卡", 400)
			return
		}
		req, _ := http.NewRequestWithContext(r.Context(), r.Method, fmt.Sprintf("http://127.0.0.1:%d%s", server.GetListenedPort(), r.URL.Path), strings.NewReader(string(raw)))
		req.Header.Set("Content-Type", "application/json")
		res, e := http.DefaultClient.Do(req)
		if e != nil {
			http.Error(w, e.Error(), 500)
			return
		}
		out, _ := io.ReadAll(res.Body)
		_ = res.Body.Close()
		mu.Lock()
		lose := input.ReqData.SubmitOnly && !dropped[input.ReqData.OperationID] && (strings.Contains(input.ReqData.OperationID, "drop-response") || strings.Contains(input.ReqData.OperationID, "top-error"))
		if lose {
			dropped[input.ReqData.OperationID] = true
		}
		mu.Unlock()
		if lose {
			if strings.Contains(input.ReqData.OperationID, "top-error") {
				_, _ = io.WriteString(w, `{"code":9999,"message":"模拟受理后网关结果丢失"}`)
				return
			}
			conn, _, e := w.(http.Hijacker).Hijack()
			if e != nil {
				t.Error(e)
				return
			}
			_ = conn.Close()
			return
		}
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(res.StatusCode)
		_, _ = w.Write(out)
	}))
	defer front.Close()
	fmt.Println("APN_CHAIN_READY=" + front.URL)
	select {
	case <-stop:
	case <-time.After(3 * time.Minute):
		t.Error("HTTP模拟链未正常结束")
	}
}

type apnBindingTestAPI struct {
	service.IUpstreamAPI
	captured chan model.SimCardOperateChangeApnInput
}

func (api *apnBindingTestAPI) ChangeSimCardOperateApn(_ context.Context, input *model.SimCardOperateChangeApnInput) (model.SimCardOperateChangeApnOutput, error) {
	api.captured <- *input
	return model.SimCardOperateChangeApnOutput{}, nil
}

func (api *apnBindingTestAPI) QuerySimApnCapability(_ context.Context, input *model.SimCardOperateChangeApnInput) (model.SimApnCapabilityOutput, error) {
	api.captured <- *input
	return model.SimApnCapabilityOutput{}, nil
}

func TestAPNProductionHTTPBindingPreservesAsyncFields(t *testing.T) {
	old := service.UpstreamAPI()
	api := &apnBindingTestAPI{IUpstreamAPI: old, captured: make(chan model.SimCardOperateChangeApnInput, 1)}
	service.RegisterUpstreamAPI(api)
	t.Cleanup(func() { service.RegisterUpstreamAPI(old) })
	server := ghttp.GetServer(t.Name())
	server.SetAddr("127.0.0.1:0")
	bindHTTPRoutes(server)
	if err := server.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = server.Shutdown() })
	for _, tc := range []struct {
		path string
		data model.SimCardOperateChangeApnInputItem
	}{
		{"/sim/change/sim-apn", model.SimCardOperateChangeApnInputItem{OperationID: "frozen-operation", OperType: "1", ApnName: "CMIOT", SubmitOnly: true}},
		{"/sim/change/sim-apn", model.SimCardOperateChangeApnInputItem{OperationID: "frozen-operation", OperType: "1", ApnName: "CMIOT", QueryJobID: "saved-job"}},
		{"/sim/change/sim-apn", model.SimCardOperateChangeApnInputItem{OperationID: "frozen-operation", OperType: "1", ApnName: "CMIOT", RecoverOnly: true}},
		{"/sim/query/sim-apn-capability", model.SimCardOperateChangeApnInputItem{OperationID: "frozen-operation", OperType: "1", ApnName: "CMIOT"}},
	} {
		input := model.SimCardOperateChangeApnInput{SimKeys: []*model.SimKey{{ChanKey: "fake", ICCID: "fake-card", MSISDN: "frozen-number", IMSI: "frozen-imsi"}}, ReqData: tc.data}
		client := g.Client().ContentJson()
		res, err := client.Post(context.Background(), fmt.Sprintf("http://127.0.0.1:%d%s", server.GetListenedPort(), tc.path), input)
		if err != nil {
			t.Fatal(err)
		}
		body := res.ReadAll()
		_ = res.Close()
		var envelope struct {
			Code int `json:"code"`
		}
		if json.Unmarshal(body, &envelope) != nil || envelope.Code != 0 {
			t.Fatalf("生产绑定失败：%s", body)
		}
		select {
		case got := <-api.captured:
			if got.ReqData != tc.data || len(got.SimKeys) != 1 || *got.SimKeys[0] != *input.SimKeys[0] {
				t.Fatalf("异步字段或快照丢失：%+v", got)
			}
		default:
			t.Fatalf("未调用生产controller：%s", body)
		}
	}
}

func TestInternalChannelTestWithoutTokenReachesRequestValidation(t *testing.T) {
	oldAdapter := g.Cfg().GetAdapter()
	adapter, err := gcfg.NewAdapterContent(`{"internal":{}}`)
	if err != nil {
		t.Fatal(err)
	}
	g.Cfg().SetAdapter(adapter)
	t.Cleanup(func() { g.Cfg().SetAdapter(oldAdapter) })

	server := ghttp.GetServer(t.Name())
	server.SetAddr("127.0.0.1:0")
	bindHTTPRoutes(server)
	if err := server.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = server.Shutdown() })
	response, err := http.Post(fmt.Sprintf("http://127.0.0.1:%d/internal/channel/test", server.GetListenedPort()), "application/json", strings.NewReader(`{}`))
	if err != nil {
		t.Fatal(err)
	}
	defer response.Body.Close()
	if response.StatusCode == http.StatusForbidden {
		t.Fatalf("request was rejected before existing parameter validation: status=%d", response.StatusCode)
	}
}

func TestBindHTTPRoutesRegistersChannelControlRoutes(t *testing.T) {
	server := ghttp.GetServer("chan-channel-control-route-test")
	server.SetAddr("127.0.0.1:0")
	bindHTTPRoutes(server)
	if err := server.Start(); err != nil {
		t.Fatalf("start HTTP server: %v", err)
	}
	t.Cleanup(func() {
		if err := server.Shutdown(); err != nil {
			t.Errorf("shutdown HTTP server: %v", err)
		}
	})

	want := map[string]bool{
		"/internal/channel/test":        false,
		"/sim/query/sim-apn-capability": false,
		"/internal/channel/refresh":     false,
		"/internal/channel/remove":      false,
	}
	for _, route := range server.GetRoutes() {
		if _, ok := want[route.Route]; ok && strings.EqualFold(route.Method, "POST") {
			want[route.Route] = true
		}
	}
	for route, found := range want {
		if !found {
			t.Fatalf("POST %s route was not registered: %+v", route, server.GetRoutes())
		}
	}
}