|
@@ -0,0 +1,637 @@
|
|
|
|
|
+package faultRecordQueue
|
|
|
|
|
+
|
|
|
|
|
+import (
|
|
|
|
|
+ "context"
|
|
|
|
|
+ "encoding/csv"
|
|
|
|
|
+ "encoding/json"
|
|
|
|
|
+ "fmt"
|
|
|
|
|
+ "os"
|
|
|
|
|
+ "strings"
|
|
|
|
|
+ "sync"
|
|
|
|
|
+ "testing"
|
|
|
|
|
+ "time"
|
|
|
|
|
+
|
|
|
|
|
+ "fwh-toolbox-server/domain"
|
|
|
|
|
+ proto "fwh-toolbox-server/function/FaultRecordProtocol"
|
|
|
|
|
+ "fwh-toolbox-server/repositories/mqtt/fwh_mqtt"
|
|
|
|
|
+
|
|
|
|
|
+ "go.mongodb.org/mongo-driver/bson"
|
|
|
|
|
+ "go.mongodb.org/mongo-driver/bson/primitive"
|
|
|
|
|
+)
|
|
|
|
|
+
|
|
|
|
|
+const (
|
|
|
|
|
+ sampleLog = "../../../../docs/fault-record/CloudLog_2026-10-14_104644.csv"
|
|
|
|
|
+ sampleArchive = "../../../../docs/fault-record/2026-10-14_05_46_20.tar.bz2"
|
|
|
|
|
+ sampleGateway = "AFCI_TEST_0713"
|
|
|
|
|
+ sampleBatch = "2026-10-14_05_46_20"
|
|
|
|
|
+)
|
|
|
|
|
+
|
|
|
|
|
+/************** fakes **************/
|
|
|
|
|
+
|
|
|
|
|
+type memStore struct {
|
|
|
|
|
+ mu sync.Mutex
|
|
|
|
|
+ ops map[primitive.ObjectID]*domain.FaultRecordOperation
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (s *memStore) Get(_ context.Context, id primitive.ObjectID) (*domain.FaultRecordOperation, error) {
|
|
|
|
|
+ s.mu.Lock()
|
|
|
|
|
+ defer s.mu.Unlock()
|
|
|
|
|
+ cp := *s.ops[id]
|
|
|
|
|
+ cp.StageTimes = map[string]time.Time{}
|
|
|
|
|
+ for k, v := range s.ops[id].StageTimes {
|
|
|
|
|
+ cp.StageTimes[k] = v
|
|
|
|
|
+ }
|
|
|
|
|
+ return &cp, nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (s *memStore) UpdateStatus(_ context.Context, id primitive.ObjectID, st domain.FaultRecordOpStatus, extra bson.M) (bool, error) {
|
|
|
|
|
+ s.mu.Lock()
|
|
|
|
|
+ defer s.mu.Unlock()
|
|
|
|
|
+ op := s.ops[id]
|
|
|
|
|
+ if op.Status.Terminal() {
|
|
|
|
|
+ return false, nil
|
|
|
|
|
+ }
|
|
|
|
|
+ op.Status = st
|
|
|
|
|
+ op.StageTimes[string(st)] = time.Now()
|
|
|
|
|
+ applyFields(op, extra)
|
|
|
|
|
+ return true, nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (s *memStore) SetFields(_ context.Context, id primitive.ObjectID, f bson.M) error {
|
|
|
|
|
+ s.mu.Lock()
|
|
|
|
|
+ defer s.mu.Unlock()
|
|
|
|
|
+ applyFields(s.ops[id], f)
|
|
|
|
|
+ return nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (s *memStore) AppendMessage(_ context.Context, id primitive.ObjectID, m domain.FaultRecordOpMessage) error {
|
|
|
|
|
+ s.mu.Lock()
|
|
|
|
|
+ defer s.mu.Unlock()
|
|
|
|
|
+ s.ops[id].Messages = append(s.ops[id].Messages, m)
|
|
|
|
|
+ return nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func applyFields(op *domain.FaultRecordOperation, f bson.M) {
|
|
|
|
|
+ for k, v := range f {
|
|
|
|
|
+ var idx int
|
|
|
|
|
+ if n, _ := fmt.Sscanf(k, "messages.%d", &idx); n == 1 {
|
|
|
|
|
+ op.Messages[idx] = v.(domain.FaultRecordOpMessage)
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ switch k {
|
|
|
|
|
+ case "error":
|
|
|
|
|
+ op.Error = v.(string)
|
|
|
|
|
+ case "reply":
|
|
|
|
|
+ r := v.(proto.Reply)
|
|
|
|
|
+ op.Reply = &r
|
|
|
|
|
+ case "fileName":
|
|
|
|
|
+ op.FileName = v.(string)
|
|
|
|
|
+ case "faultRecordId":
|
|
|
|
|
+ op.FaultRecordID = v.(int64)
|
|
|
|
|
+ case "uploadEquipNo":
|
|
|
|
|
+ op.UploadEquipNo = v.(string)
|
|
|
|
|
+ case "minioObject":
|
|
|
|
|
+ op.MinioObject = v.(string)
|
|
|
|
|
+ case "fileSize":
|
|
|
|
|
+ op.FileSize = int64(v.(int))
|
|
|
|
|
+ case "exitSent":
|
|
|
|
|
+ op.ExitSent = v.(bool)
|
|
|
|
|
+ case "timeoutStage":
|
|
|
|
|
+ op.TimeoutStage = v.(domain.FaultRecordOpStatus)
|
|
|
|
|
+ case "cancelRequested":
|
|
|
|
|
+ op.CancelReq = v.(bool)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// fakeBus 收到 377 后按脚本回放消息
|
|
|
|
|
+type fakeBus struct {
|
|
|
|
|
+ mu sync.Mutex
|
|
|
|
|
+ handlers map[string][]fwh_mqtt.Handler
|
|
|
|
|
+ published []fwh_mqtt.FwhMqttMsg
|
|
|
|
|
+ script func(b *fakeBus, msg fwh_mqtt.FwhMqttMsg)
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (b *fakeBus) Subscribe(_ context.Context, topic string, h fwh_mqtt.Handler) (func(), error) {
|
|
|
|
|
+ b.mu.Lock()
|
|
|
|
|
+ defer b.mu.Unlock()
|
|
|
|
|
+ if b.handlers == nil {
|
|
|
|
|
+ b.handlers = map[string][]fwh_mqtt.Handler{}
|
|
|
|
|
+ }
|
|
|
|
|
+ b.handlers[topic] = append(b.handlers[topic], h)
|
|
|
|
|
+ return func() {}, nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (b *fakeBus) Publish(_ context.Context, topic string, msg fwh_mqtt.FwhMqttMsg) error {
|
|
|
|
|
+ b.mu.Lock()
|
|
|
|
|
+ b.published = append(b.published, msg)
|
|
|
|
|
+ script := b.script
|
|
|
|
|
+ b.mu.Unlock()
|
|
|
|
|
+ if script != nil {
|
|
|
|
|
+ go script(b, msg)
|
|
|
|
|
+ }
|
|
|
|
|
+ return nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (b *fakeBus) push(raw string) {
|
|
|
|
|
+ cmdType, _, _, _ := proto.SplitEnvelope([]byte(raw))
|
|
|
|
|
+ // 与实际一致:378/379 走 c2s/mng,380 走 s2c/mng;607/610 走 s2c/debug,608/609 走 c2s/debug
|
|
|
|
|
+ topic := fwh_mqtt.C2STopic("mng/" + sampleGateway)
|
|
|
|
|
+ switch cmdType {
|
|
|
|
|
+ case proto.CmdTypeReplyAck:
|
|
|
|
|
+ topic = fwh_mqtt.S2CTopic("mng/" + sampleGateway)
|
|
|
|
|
+ case proto.CmdTypeUploadCall, proto.CmdTypeUploadDoneA:
|
|
|
|
|
+ topic = fwh_mqtt.S2CTopic("debug/" + sampleGateway)
|
|
|
|
|
+ case proto.CmdTypeUploadAck, proto.CmdTypeUploadDone:
|
|
|
|
|
+ topic = fwh_mqtt.C2STopic("debug/" + sampleGateway)
|
|
|
|
|
+ }
|
|
|
|
|
+ b.mu.Lock()
|
|
|
|
|
+ hs := append([]fwh_mqtt.Handler(nil), b.handlers[topic]...)
|
|
|
|
|
+ b.mu.Unlock()
|
|
|
|
|
+ for _, h := range hs {
|
|
|
|
|
+ h(fwh_mqtt.Message{Topic: topic, Raw: []byte(raw), ReceivedAt: time.Now()})
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+type memFiles struct {
|
|
|
|
|
+ mu sync.Mutex
|
|
|
|
|
+ data map[string][]byte
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (f *memFiles) Put(_ context.Context, key string, d []byte) error {
|
|
|
|
|
+ f.mu.Lock()
|
|
|
|
|
+ defer f.mu.Unlock()
|
|
|
|
|
+ f.data[key] = d
|
|
|
|
|
+ return nil
|
|
|
|
|
+}
|
|
|
|
|
+func (f *memFiles) Get(_ context.Context, key string) ([]byte, error) {
|
|
|
|
|
+ f.mu.Lock()
|
|
|
|
|
+ defer f.mu.Unlock()
|
|
|
|
|
+ d, ok := f.data[key]
|
|
|
|
|
+ if !ok {
|
|
|
|
|
+ return nil, fmt.Errorf("not found")
|
|
|
|
|
+ }
|
|
|
|
|
+ return d, nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+type fakeLock struct{ locked bool }
|
|
|
|
|
+
|
|
|
|
|
+func (l fakeLock) Acquire(context.Context, string, string, time.Duration) (func(), bool, error) {
|
|
|
|
|
+ return func() {}, !l.locked, nil
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/************** helpers **************/
|
|
|
|
|
+
|
|
|
|
|
+// sampleRound 读取样本日志第二轮(05:45~05:46)按时间正序的报文
|
|
|
|
|
+func sampleRound(t *testing.T) []string {
|
|
|
|
|
+ t.Helper()
|
|
|
|
|
+ f, err := os.Open(sampleLog)
|
|
|
|
|
+ if err != nil {
|
|
|
|
|
+ t.Skipf("sample log not found: %v", err)
|
|
|
|
|
+ }
|
|
|
|
|
+ defer f.Close()
|
|
|
|
|
+ rows, err := csv.NewReader(f).ReadAll()
|
|
|
|
|
+ if err != nil {
|
|
|
|
|
+ t.Fatal(err)
|
|
|
|
|
+ }
|
|
|
|
|
+ var list []string
|
|
|
|
|
+ for i := len(rows) - 1; i >= 1; i-- { // CSV 为倒序
|
|
|
|
|
+ if strings.HasPrefix(rows[i][8], "2026-10-14 05:4") && rows[i][8] >= "2026-10-14 05:45" {
|
|
|
|
|
+ list = append(list, rows[i][7])
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return list
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// pickCmd 从样本中取出与 cmd 相关的报文(378/379/380 按 dataArea.cmd 过滤),不含 377
|
|
|
|
|
+func pickCmd(msgs []string, cmd int) []string {
|
|
|
|
|
+ var out []string
|
|
|
|
|
+ for _, raw := range msgs {
|
|
|
|
|
+ cmdType, _, da, _ := proto.SplitEnvelope([]byte(raw))
|
|
|
|
|
+ if cmdType == proto.CmdTypeRequest {
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ if cmdType >= proto.CmdTypeAck && cmdType <= proto.CmdTypeReplyAck {
|
|
|
|
|
+ var v struct {
|
|
|
|
|
+ Cmd int `json:"cmd"`
|
|
|
|
|
+ }
|
|
|
|
|
+ json.Unmarshal(da, &v)
|
|
|
|
|
+ if v.Cmd != cmd {
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ } else if cmd != proto.CmdGetRecord {
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ out = append(out, raw)
|
|
|
|
|
+ }
|
|
|
|
|
+ return out
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+type harness struct {
|
|
|
|
|
+ d deps
|
|
|
|
|
+ store *memStore
|
|
|
|
|
+ bus *fakeBus
|
|
|
|
|
+ files *memFiles
|
|
|
|
|
+ op *domain.FaultRecordOperation
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func newHarness(t *testing.T, kind domain.FaultRecordOpKind, req domain.FaultRecordOpRequest) *harness {
|
|
|
|
|
+ t.Helper()
|
|
|
|
|
+ op := &domain.FaultRecordOperation{
|
|
|
|
|
+ ID: primitive.NewObjectID(), Env: "test", GatewaySn: sampleGateway,
|
|
|
|
|
+ DeviceType: proto.DeviceTypeAGate, DeviceSn: sampleGateway,
|
|
|
|
|
+ Kind: kind, Cmd: kind.Cmd(), Request: req,
|
|
|
|
|
+ Status: domain.FaultRecordStatusPending, StageTimes: map[string]time.Time{"pending": time.Now()},
|
|
|
|
|
+ }
|
|
|
|
|
+ h := &harness{
|
|
|
|
|
+ store: &memStore{ops: map[primitive.ObjectID]*domain.FaultRecordOperation{op.ID: op}},
|
|
|
|
|
+ bus: &fakeBus{},
|
|
|
|
|
+ files: &memFiles{data: map[string][]byte{}},
|
|
|
|
|
+ op: op,
|
|
|
|
|
+ }
|
|
|
|
|
+ archive, _ := os.ReadFile(sampleArchive)
|
|
|
|
|
+ to := defaultTimeouts
|
|
|
|
|
+ to.downloadInterval = 10 * time.Millisecond
|
|
|
|
|
+ to.cancelPoll = 10 * time.Millisecond
|
|
|
|
|
+ to.drain = 50 * time.Millisecond
|
|
|
|
|
+ to.afterUploadAck = 10 * time.Millisecond
|
|
|
|
|
+ h.d = deps{
|
|
|
|
|
+ bus: func(string) (mqttBus, error) { return h.bus, nil },
|
|
|
|
|
+ store: h.store,
|
|
|
|
|
+ files: h.files,
|
|
|
|
|
+ download: func(env, equipNo, fileName string) ([]byte, error) {
|
|
|
|
|
+ if equipNo != sampleGateway || fileName != sampleBatch+".tar.bz2" || archive == nil {
|
|
|
|
|
+ return nil, fmt.Errorf("not found")
|
|
|
|
|
+ }
|
|
|
|
|
+ return archive, nil
|
|
|
|
|
+ },
|
|
|
|
|
+ lock: fakeLock{},
|
|
|
|
|
+ downloadURL: func(env, equipNo, fileName string) string {
|
|
|
|
|
+ return "https://" + env + ".example/hes-upload/v1/file/download/logger/" + equipNo + "/" + fileName
|
|
|
|
|
+ },
|
|
|
|
|
+ timeouts: to,
|
|
|
|
|
+ }
|
|
|
|
|
+ return h
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func (h *harness) run(t *testing.T) *domain.FaultRecordOperation {
|
|
|
|
|
+ t.Helper()
|
|
|
|
|
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
|
|
|
+ defer cancel()
|
|
|
|
|
+ if err := h.d.run(ctx, h.op.ID); err != nil {
|
|
|
|
|
+ t.Fatalf("run: %v", err)
|
|
|
|
|
+ }
|
|
|
|
|
+ op, _ := h.store.Get(ctx, h.op.ID)
|
|
|
|
|
+ return op
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// replay 收到 377 后依次推送消息
|
|
|
|
|
+func replay(msgs []string) func(*fakeBus, fwh_mqtt.FwhMqttMsg) {
|
|
|
|
|
+ return func(b *fakeBus, m fwh_mqtt.FwhMqttMsg) {
|
|
|
|
|
+ if m.CmdType != proto.CmdTypeRequest {
|
|
|
|
|
+ return
|
|
|
|
|
+ }
|
|
|
|
|
+ for _, raw := range msgs {
|
|
|
|
|
+ b.push(raw)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/************** tests **************/
|
|
|
|
|
+
|
|
|
|
|
+func TestRecordFetchFullFlow(t *testing.T) {
|
|
|
|
|
+ msgs := pickCmd(sampleRound(t), proto.CmdGetRecord)
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.bus.script = replay(msgs)
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone {
|
|
|
|
|
+ t.Fatalf("status = %s, error = %s", op.Status, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+ if op.FileName != sampleBatch+".tar.bz2" || op.FaultRecordID != 5226 || op.MinioObject == "" {
|
|
|
|
|
+ t.Errorf("record fields: %+v", op)
|
|
|
|
|
+ }
|
|
|
|
|
+ for _, s := range []string{"sent", "accepted", "ready", "uploading", "uploaded", "downloaded", "done"} {
|
|
|
|
|
+ if _, ok := op.StageTimes[s]; !ok {
|
|
|
|
|
+ t.Errorf("missing stage %s", s)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ // 发送的 377 + 样本全部报文(含 609 之后的 610)+ 1 条 HTTP 下载记录
|
|
|
|
|
+ var hasHTTP bool
|
|
|
|
|
+ for _, m := range op.Messages {
|
|
|
|
|
+ if m.Direction == "http" && strings.Contains(m.Raw, `"result":"ok"`) {
|
|
|
|
|
+ hasHTTP = true
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ // 顺序:…609 → 610 → HTTP 下载(收到 610 后才下载)
|
|
|
|
|
+ first, last := op.Messages[0], op.Messages[len(op.Messages)-1]
|
|
|
|
|
+ prev := op.Messages[len(op.Messages)-2]
|
|
|
|
|
+ if first.CmdType != proto.CmdTypeRequest || last.Direction != "http" || prev.CmdType != proto.CmdTypeUploadDoneA || !hasHTTP || len(op.Messages) != len(msgs)+2 {
|
|
|
|
|
+ t.Errorf("messages = %d (first %d, prev %d, last %s, http %v)", len(op.Messages), first.CmdType, prev.CmdType, last.Direction, hasHTTP)
|
|
|
|
|
+ }
|
|
|
|
|
+ b, _ := json.Marshal(h.bus.published[0].DataArea)
|
|
|
|
|
+ if string(b) != `{"cmd":2,"data":{"list":[{"type":"aGate","info":[{"sn":"AFCI_TEST_0713","index":[7]}]}]}}` {
|
|
|
|
|
+ t.Errorf("request = %s", b)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestDeviceInfoFlow(t *testing.T) {
|
|
|
|
|
+ msgs := pickCmd(sampleRound(t), proto.CmdRealtime)
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpDeviceInfo, domain.FaultRecordOpRequest{CodeList: []int{3}})
|
|
|
|
|
+ h.bus.script = replay(msgs)
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone || op.Reply == nil || len(op.Reply.List) != 1 {
|
|
|
|
|
+ t.Fatalf("status = %s reply = %+v", op.Status, op.Reply)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestBusyReply(t *testing.T) {
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpParamRead, domain.FaultRecordOpRequest{IdList: []int{5}})
|
|
|
|
|
+ h.bus.script = replay([]string{
|
|
|
|
|
+ `{"cmdType":378,"equipNo":"AFCI_TEST_0713","dataArea":{"cmd":3,"result":0}}`,
|
|
|
|
|
+ `{"cmdType":379,"equipNo":"AFCI_TEST_0713","dataArea":{"cmd":3,"result":"busy"}}`,
|
|
|
|
|
+ })
|
|
|
|
|
+ if op := h.run(t); op.Status != domain.FaultRecordStatusBusy {
|
|
|
|
|
+ t.Fatalf("status = %s", op.Status)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestIgnoresOtherCmdAndOtherGateway(t *testing.T) {
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpParamRead, domain.FaultRecordOpRequest{IdList: []int{5}})
|
|
|
|
|
+ h.bus.script = replay([]string{
|
|
|
|
|
+ `{"cmdType":378,"equipNo":"AFCI_TEST_0713","dataArea":{"cmd":3,"result":0}}`,
|
|
|
|
|
+ `{"cmdType":379,"equipNo":"AFCI_TEST_0713","dataArea":{"cmd":0,"result":"ok"}}`, // 其他 cmd
|
|
|
|
|
+ `{"cmdType":379,"equipNo":"OTHER","dataArea":{"cmd":3,"result":"busy"}}`, // 其他网关
|
|
|
|
|
+ `{"cmdType":379,"equipNo":"AFCI_TEST_0713","dataArea":{"cmd":3,"result":"ok","list":[{"type":"aGate","info":[{"sn":"AFCI_TEST_0713","idInfo":[{"id":5,"idData":"1194","resultOne":"ok"}]}]}]}}`,
|
|
|
|
|
+ })
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone || op.Reply.List[0].Info[0].IdInfo[0].IdData != "1194" {
|
|
|
|
|
+ t.Fatalf("status = %s reply = %+v", op.Status, op.Reply)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestAckTimeout(t *testing.T) {
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpParamRead, domain.FaultRecordOpRequest{IdList: []int{5}})
|
|
|
|
|
+ h.d.timeouts.ack = 50 * time.Millisecond
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusTimeout || op.TimeoutStage != domain.FaultRecordStatusAccepted {
|
|
|
|
|
+ t.Fatalf("status = %s stage = %s", op.Status, op.TimeoutStage)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestCancelSendsExit(t *testing.T) {
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.bus.script = func(b *fakeBus, m fwh_mqtt.FwhMqttMsg) {
|
|
|
|
|
+ if m.CmdType != proto.CmdTypeRequest {
|
|
|
|
|
+ return
|
|
|
|
|
+ }
|
|
|
|
|
+ if da, _ := json.Marshal(m.DataArea); strings.Contains(string(da), `"cmd":2`) {
|
|
|
|
|
+ b.push(`{"cmdType":378,"equipNo":"AFCI_TEST_0713","dataArea":{"cmd":2,"result":0}}`)
|
|
|
|
|
+ time.Sleep(30 * time.Millisecond)
|
|
|
|
|
+ h.store.SetFields(context.Background(), h.op.ID, bson.M{"cancelRequested": true})
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusCancelled || !op.ExitSent {
|
|
|
|
|
+ t.Fatalf("status = %s exitSent = %v", op.Status, op.ExitSent)
|
|
|
|
|
+ }
|
|
|
|
|
+ last, _ := json.Marshal(h.bus.published[len(h.bus.published)-1].DataArea)
|
|
|
|
|
+ if string(last) != `{"cmd":4,"data":{}}` {
|
|
|
|
|
+ t.Errorf("exit request = %s", last)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestRecoverSentIsInterrupted(t *testing.T) {
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.op.Status = domain.FaultRecordStatusAccepted
|
|
|
|
|
+ if op := h.run(t); op.Status != domain.FaultRecordStatusInterrupted || len(h.bus.published) != 0 {
|
|
|
|
|
+ t.Fatalf("status = %s published = %d", op.Status, len(h.bus.published))
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestRecoverReadyDownloadsDirectly(t *testing.T) {
|
|
|
|
|
+ if _, err := os.Stat(sampleArchive); err != nil {
|
|
|
|
|
+ t.Skip("sample archive not found")
|
|
|
|
|
+ }
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.op.Status = domain.FaultRecordStatusReady
|
|
|
|
|
+ h.op.FileName = sampleBatch + ".tar.bz2"
|
|
|
|
|
+ h.op.StageTimes["ready"] = time.Now()
|
|
|
|
|
+ if op := h.run(t); op.Status != domain.FaultRecordStatusDone || len(h.bus.published) != 0 {
|
|
|
|
|
+ t.Fatalf("status = %s error = %s", op.Status, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+func TestGatewayLocked(t *testing.T) {
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpParamRead, domain.FaultRecordOpRequest{IdList: []int{5}})
|
|
|
|
|
+ h.d.lock = fakeLock{locked: true}
|
|
|
|
|
+ if op := h.run(t); op.Status != domain.FaultRecordStatusGatewayLocked {
|
|
|
|
|
+ t.Fatalf("status = %s", op.Status)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 订阅不到平台下发方向(无 607/380/610)时:靠 608 取得平台记录 ID;没有 610 不会提前下载,超时时按规则直接下载一次
|
|
|
|
|
+func TestRecordFetchWithoutS2C(t *testing.T) {
|
|
|
|
|
+ var c2sOnly []string
|
|
|
|
|
+ for _, raw := range pickCmd(sampleRound(t), proto.CmdGetRecord) {
|
|
|
|
|
+ cmdType, _, _, _ := proto.SplitEnvelope([]byte(raw))
|
|
|
|
|
+ if cmdType != proto.CmdTypeUploadCall && cmdType != proto.CmdTypeReplyAck && cmdType != proto.CmdTypeUploadDoneA {
|
|
|
|
|
+ c2sOnly = append(c2sOnly, raw)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.d.timeouts.upload = 300 * time.Millisecond
|
|
|
|
|
+ h.bus.script = replay(c2sOnly)
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone || op.FaultRecordID != 5226 {
|
|
|
|
|
+ t.Fatalf("status = %s faultRecordId = %d error = %s", op.Status, op.FaultRecordID, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+ for _, s := range []string{"uploading", "uploaded", "downloaded"} {
|
|
|
|
|
+ if _, ok := op.StageTimes[s]; !ok {
|
|
|
|
|
+ t.Errorf("missing stage %s", s)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ // 下载发生在超时时,而不是收到 609 时
|
|
|
|
|
+ if op.StageTimes["downloaded"].Sub(op.StageTimes["uploaded"]) < 200*time.Millisecond {
|
|
|
|
|
+ t.Error("should not download before 610 or timeout")
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 一次获取多条记录:请求包含全部序号;压缩包中缺少的序号在结果中提示
|
|
|
|
|
+func TestRecordFetchMultiple(t *testing.T) {
|
|
|
|
|
+ msgs := pickCmd(sampleRound(t), proto.CmdGetRecord)
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndexes: []int{7, 9}})
|
|
|
|
|
+ h.bus.script = replay(msgs)
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ b, _ := json.Marshal(h.bus.published[0].DataArea)
|
|
|
|
|
+ if !strings.Contains(string(b), `"index":[7,9]`) {
|
|
|
|
|
+ t.Errorf("request = %s", b)
|
|
|
|
|
+ }
|
|
|
|
|
+ // 样本压缩包只有记录 7
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone || !strings.Contains(op.Error, "9") {
|
|
|
|
|
+ t.Fatalf("status = %s error = %q", op.Status, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 收到 610 后等待 afterUploadAck 再下载
|
|
|
|
|
+func TestDownloadAfter610Delay(t *testing.T) {
|
|
|
|
|
+ msgs := pickCmd(sampleRound(t), proto.CmdGetRecord)
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.d.timeouts.afterUploadAck = 150 * time.Millisecond
|
|
|
|
|
+ h.bus.script = replay(msgs)
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone {
|
|
|
|
|
+ t.Fatalf("status = %s error = %s", op.Status, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+ var at610 time.Time
|
|
|
|
|
+ for _, m := range op.Messages {
|
|
|
|
|
+ if m.CmdType == proto.CmdTypeUploadDoneA {
|
|
|
|
|
+ at610 = m.At
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if at610.IsZero() || op.StageTimes["downloaded"].Sub(at610) < 150*time.Millisecond {
|
|
|
|
|
+ t.Errorf("download should start ≥150ms after 610 (610 at %v, downloaded at %v)", at610, op.StageTimes["downloaded"])
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 收到 610 后开始下载;连续相同的失败只保留一条并累计次数,成功后另记一条
|
|
|
|
|
+func TestDownloadRetryDedupesFailures(t *testing.T) {
|
|
|
|
|
+ msgs := pickCmd(sampleRound(t), proto.CmdGetRecord)
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.bus.script = replay(msgs)
|
|
|
|
|
+ real := h.d.download
|
|
|
|
|
+ var mu sync.Mutex
|
|
|
|
|
+ calls := 0
|
|
|
|
|
+ h.d.download = func(env, equipNo, fileName string) ([]byte, error) {
|
|
|
|
|
+ mu.Lock()
|
|
|
|
|
+ calls++
|
|
|
|
|
+ n := calls
|
|
|
|
|
+ mu.Unlock()
|
|
|
|
|
+ if n <= 3 {
|
|
|
|
|
+ return nil, fmt.Errorf("404 Not Found")
|
|
|
|
|
+ }
|
|
|
|
|
+ return real(env, equipNo, fileName)
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone {
|
|
|
|
|
+ t.Fatalf("status = %s error = %s", op.Status, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+ var httpMsgs []domain.FaultRecordOpMessage
|
|
|
|
|
+ for _, m := range op.Messages {
|
|
|
|
|
+ if m.Direction == "http" {
|
|
|
|
|
+ httpMsgs = append(httpMsgs, m)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if len(httpMsgs) != 2 || !strings.Contains(httpMsgs[0].Raw, `"url":"https://test.example/hes-upload/v1/file/download/logger/AFCI_TEST_0713/2026-10-14_05_46_20.tar.bz2"`) || !strings.Contains(httpMsgs[0].Raw, `"count":3`) || !strings.Contains(httpMsgs[1].Raw, `"result":"ok"`) {
|
|
|
|
|
+ t.Fatalf("http records = %+v", httpMsgs)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 一直失败时重试到总超时,记为下载失败并注明尝试次数
|
|
|
|
|
+func TestDownloadRetryUntilTimeout(t *testing.T) {
|
|
|
|
|
+ msgs := pickCmd(sampleRound(t), proto.CmdGetRecord)
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.d.timeouts.upload = 200 * time.Millisecond
|
|
|
|
|
+ h.d.timeouts.downloadInterval = 30 * time.Millisecond
|
|
|
|
|
+ h.bus.script = replay(msgs)
|
|
|
|
|
+ h.d.download = func(string, string, string) ([]byte, error) { return nil, fmt.Errorf("404 Not Found") }
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDownloadFailed || !strings.Contains(op.Error, "已尝试") {
|
|
|
|
|
+ t.Fatalf("status = %s error = %s", op.Status, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+ n := 0
|
|
|
|
|
+ for _, m := range op.Messages {
|
|
|
|
|
+ if m.Direction == "http" {
|
|
|
|
|
+ n++
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if n != 1 {
|
|
|
|
|
+ t.Errorf("same failures should collapse into 1 record, got %d", n)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 新固件:609.fileName 不带 .tar.bz2、608 不带 id;下载地址仍须带后缀,平台记录 ID 取自 607
|
|
|
|
|
+func TestNewFirmware609WithoutSuffix(t *testing.T) {
|
|
|
|
|
+ var msgs []string
|
|
|
|
|
+ for _, raw := range pickCmd(sampleRound(t), proto.CmdGetRecord) {
|
|
|
|
|
+ raw = strings.Replace(raw, `"fileName":"2026-10-14_05_46_20.tar.bz2"`, `"fileName":"2026-10-14_05_46_20"`, 1)
|
|
|
|
|
+ raw = strings.Replace(raw, `"result":0,"filename":"2026-10-14_05_46_20","result":5226`, `"result":0,"filename":"2026-10-14_05_46_20"`, 1)
|
|
|
|
|
+ msgs = append(msgs, raw)
|
|
|
|
|
+ }
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ h.bus.script = replay(msgs)
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone || op.FileName != sampleBatch+".tar.bz2" || op.FaultRecordID != 5226 {
|
|
|
|
|
+ t.Fatalf("status = %s file = %s frid = %d error = %s", op.Status, op.FileName, op.FaultRecordID, op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 手动重试下载:状态为 uploaded 且有 DownloadRetryAt 时,从重试时刻重新计算总超时并直接下载
|
|
|
|
|
+func TestManualDownloadRetry(t *testing.T) {
|
|
|
|
|
+ if _, err := os.Stat(sampleArchive); err != nil {
|
|
|
|
|
+ t.Skip("sample archive not found")
|
|
|
|
|
+ }
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ now := time.Now()
|
|
|
|
|
+ h.op.Status = domain.FaultRecordStatusUploaded
|
|
|
|
|
+ h.op.FileName = sampleBatch + ".tar.bz2"
|
|
|
|
|
+ h.op.StageTimes["ready"] = now.Add(-time.Hour) // 原总超时早已过去
|
|
|
|
|
+ h.op.StageTimes["uploaded"] = now.Add(-time.Hour)
|
|
|
|
|
+ h.op.DownloadRetryAt = &now
|
|
|
|
|
+ calls := 0
|
|
|
|
|
+ real := h.d.download
|
|
|
|
|
+ h.d.download = func(env, equipNo, fileName string) ([]byte, error) {
|
|
|
|
|
+ calls++
|
|
|
|
|
+ if calls == 1 {
|
|
|
|
|
+ return nil, fmt.Errorf("404 Not Found")
|
|
|
|
|
+ }
|
|
|
|
|
+ return real(env, equipNo, fileName)
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDone || calls != 2 || len(h.bus.published) != 0 {
|
|
|
|
|
+ t.Fatalf("status = %s calls = %d published = %d error = %s", op.Status, calls, len(h.bus.published), op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// 手动重试只尝试 3 次;已保存的文件名不带后缀(旧记录)时,下载链接仍按规则补齐 .tar.bz2
|
|
|
|
|
+func TestManualRetryThreeAttemptsAndSuffix(t *testing.T) {
|
|
|
|
|
+ h := newHarness(t, domain.FaultRecordOpRecordFetch, domain.FaultRecordOpRequest{RecordIndex: 7})
|
|
|
|
|
+ now := time.Now()
|
|
|
|
|
+ h.op.Status = domain.FaultRecordStatusUploaded
|
|
|
|
|
+ h.op.FileName = sampleBatch // 旧记录:不带后缀
|
|
|
|
|
+ h.op.StageTimes["ready"] = now
|
|
|
|
|
+ h.op.StageTimes["uploaded"] = now
|
|
|
|
|
+ h.op.DownloadRetryAt = &now
|
|
|
|
|
+ var names []string
|
|
|
|
|
+ h.d.download = func(env, equipNo, fileName string) ([]byte, error) {
|
|
|
|
|
+ names = append(names, fileName)
|
|
|
|
|
+ return nil, fmt.Errorf("404 Not Found")
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ op := h.run(t)
|
|
|
|
|
+ if op.Status != domain.FaultRecordStatusDownloadFailed || len(names) != 3 || !strings.Contains(op.Error, "已尝试 3 次") {
|
|
|
|
|
+ t.Fatalf("status = %s attempts = %d error = %s", op.Status, len(names), op.Error)
|
|
|
|
|
+ }
|
|
|
|
|
+ for _, n := range names {
|
|
|
|
|
+ if n != sampleBatch+".tar.bz2" {
|
|
|
|
|
+ t.Errorf("download file name = %s", n)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ var url string
|
|
|
|
|
+ for _, m := range op.Messages {
|
|
|
|
|
+ if m.Direction == "http" {
|
|
|
|
|
+ url = m.Raw
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ if !strings.Contains(url, sampleBatch+".tar.bz2") || !strings.Contains(url, `"count":3`) {
|
|
|
|
|
+ t.Errorf("http record = %s", url)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|