FranklinWH-Toolbox/service/task/queue/faultRecordQueue/executor_test.go

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)
	}
}