MiFaZhan
304 lines
11 KiB
Go
304 lines
11 KiB
Go
package lifecycle
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/gogf/gf/v2/encoding/gjson"
|
|
"github.com/gogf/gf/v2/os/gtime"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/batch"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/entity"
|
|
)
|
|
|
|
func TestRestartResumeBeforeScheduledTimeIsRescheduled(t *testing.T) {
|
|
now := gtime.New("2026-10-10 13:46:42")
|
|
repo := &fakeRestartRepository{
|
|
main: restartMain(consts.RestartPhaseWaitingResume, gjson.New(map[string]any{
|
|
"restart": map[string]any{"resumeAt": "2026-10-10 13:46:44"},
|
|
})),
|
|
}
|
|
publisher := &recordingRestartPublisher{}
|
|
executor := NewRestartExecutorWithPublisher(repo, &fakeUpstream{}, publisher.PublishDelayed).(*restartExecutor)
|
|
executor.now = func() *gtime.Time { return now }
|
|
|
|
ok, err := executor.CanExecuteProcessing(context.Background(), repo.main)
|
|
if ok || err != nil {
|
|
t.Fatalf("CanExecuteProcessing() = %t, %v; want false, nil", ok, err)
|
|
}
|
|
if publisher.calls != 1 || publisher.delay != 2*time.Second {
|
|
t.Fatalf("publisher calls=%d delay=%s; want one call with 2s delay", publisher.calls, publisher.delay)
|
|
}
|
|
if publisher.message != (batch.Message{BatchNo: repo.main.BatchNo, BatchType: repo.main.BatchType}) {
|
|
t.Fatalf("rescheduled message = %+v", publisher.message)
|
|
}
|
|
if repo.claimed || repo.finalized {
|
|
t.Fatal("early message must not claim or finalize the batch")
|
|
}
|
|
}
|
|
|
|
func TestRestartResumeRescheduleFailureIsReturned(t *testing.T) {
|
|
now := gtime.New("2026-10-10 13:46:42")
|
|
main := restartMain(consts.RestartPhaseWaitingResume, gjson.New(map[string]any{
|
|
"restart": map[string]any{"resumeAt": "2026-10-10 13:46:44"},
|
|
}))
|
|
publishErr := errors.New("publish failed")
|
|
publisher := func(context.Context, string, string, time.Duration, any) error { return publishErr }
|
|
executor := NewRestartExecutorWithPublisher(&fakeRestartRepository{main: main}, &fakeUpstream{}, publisher).(*restartExecutor)
|
|
executor.now = func() *gtime.Time { return now }
|
|
ok, err := executor.CanExecuteProcessing(context.Background(), main)
|
|
if ok || !errors.Is(err, publishErr) || batch.IsRetryable(err) {
|
|
t.Fatalf("CanExecuteProcessing() = %t, %v; want publish error without execution retry flag", ok, err)
|
|
}
|
|
}
|
|
|
|
func TestRestartStopValidationFailureSkipsUpstream(t *testing.T) {
|
|
task := restartStopTask()
|
|
task.Details = append(task.Details, &Detail{
|
|
ID: 2,
|
|
ICCID: "iccid-2",
|
|
Authorized: false,
|
|
Card: &Card{ID: 12, ChanKey: "channel", BossCardStatus: consts.SimBossCardStatusActivated},
|
|
})
|
|
repo := &fakeRestartRepository{
|
|
main: restartMain(consts.RestartPhaseCreated, nil),
|
|
stopTask: task,
|
|
}
|
|
upstream := &fakeUpstream{}
|
|
executor := NewRestartExecutorWithPublisher(repo, upstream, fakeRestartPublisher(t)).(*restartExecutor)
|
|
|
|
if err := executor.Execute(context.Background(), "RST-1"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if upstream.calls != 0 {
|
|
t.Fatalf("upstream calls = %d, want 0", upstream.calls)
|
|
}
|
|
if repo.validationStage != restartStageStop || !repo.finalized {
|
|
t.Fatalf("validation/finalize state = stage:%q finalized:%t", repo.validationStage, repo.finalized)
|
|
}
|
|
if len(repo.saved) != 0 || repo.waited {
|
|
t.Fatalf("unexpected stop result or wait: saved=%d waited=%t", len(repo.saved), repo.waited)
|
|
}
|
|
}
|
|
|
|
func TestRestartStopSuccessSchedulesDelayedResume(t *testing.T) {
|
|
now := gtime.NewFromStr("2026-07-20 10:00:00")
|
|
task := restartStopTask()
|
|
task.Batch.BatchInfo = gjson.New(map[string]any{
|
|
"resumeDelaySeconds": 1,
|
|
"restart": map[string]any{"phase": consts.RestartPhaseCreated},
|
|
})
|
|
repo := &fakeRestartRepository{
|
|
main: restartMain(consts.RestartPhaseCreated, gjson.New(map[string]any{"resumeDelaySeconds": 1})),
|
|
stopTask: task,
|
|
}
|
|
upstream := &fakeUpstream{response: []model.SimBaseRes[model.SimCardStatusChangeOutputItem]{
|
|
{Code: 0, Message: "ok", ResData: model.SimCardStatusChangeOutputItem{OrderNum: "stop-order"}},
|
|
}}
|
|
publisher := &recordingRestartPublisher{}
|
|
executor := NewRestartExecutorWithPublisher(repo, upstream, publisher.PublishDelayed).(*restartExecutor)
|
|
executor.now = func() *gtime.Time { return now }
|
|
|
|
if err := executor.Execute(context.Background(), "RST-2"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if upstream.calls != 1 || upstream.input.ReqData.OperType != consts.SimCardStatusOperTypeStop {
|
|
t.Fatalf("upstream stop input = %+v calls=%d", upstream.input, upstream.calls)
|
|
}
|
|
if len(repo.saved) != 1 || repo.saved[0].stage != restartStageStop || !repo.saved[0].outcome.Success {
|
|
t.Fatalf("saved stop results = %+v", repo.saved)
|
|
}
|
|
if !repo.waited || repo.resumeAt.String() != now.Add(time.Second).String() {
|
|
t.Fatalf("resume wait = waited:%t resumeAt:%v", repo.waited, repo.resumeAt)
|
|
}
|
|
message, ok := publisher.message.(batch.Message)
|
|
if !ok || publisher.calls != 1 || publisher.delay != time.Second || message.BatchType != consts.BatchTypeSimCardRestart {
|
|
t.Fatalf("publisher = %+v", publisher)
|
|
}
|
|
if repo.finalized {
|
|
t.Fatal("batch must not finalize before resume when stop succeeded")
|
|
}
|
|
}
|
|
|
|
func TestRestartResumeClaimLostSkipsDuplicateUpstream(t *testing.T) {
|
|
now := gtime.NewFromStr("2026-07-20 10:05:00")
|
|
repo := &fakeRestartRepository{
|
|
main: restartMain(consts.RestartPhaseWaitingResume, gjson.New(map[string]any{"restart": map[string]any{"resumeAt": now.String()}})),
|
|
claimResult: false,
|
|
resumeTask: restartResumeTask(),
|
|
}
|
|
upstream := &fakeUpstream{}
|
|
executor := NewRestartExecutorWithPublisher(repo, upstream, fakeRestartPublisher(t)).(*restartExecutor)
|
|
executor.now = func() *gtime.Time { return now }
|
|
|
|
if err := executor.Execute(context.Background(), "RST-3"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !repo.claimed {
|
|
t.Fatal("resume CAS was not attempted")
|
|
}
|
|
if upstream.calls != 0 || repo.loadedResume || repo.finalized {
|
|
t.Fatalf("duplicate was not skipped: upstream=%d loadedResume=%t finalized=%t", upstream.calls, repo.loadedResume, repo.finalized)
|
|
}
|
|
}
|
|
|
|
func TestRestartResumeExecutesAfterClaim(t *testing.T) {
|
|
now := gtime.NewFromStr("2026-07-20 10:05:00")
|
|
repo := &fakeRestartRepository{
|
|
main: restartMain(consts.RestartPhaseWaitingResume, gjson.New(map[string]any{"restart": map[string]any{"resumeAt": now.String()}})),
|
|
claimResult: true,
|
|
resumeTask: restartResumeTask(),
|
|
}
|
|
upstream := &fakeUpstream{response: []model.SimBaseRes[model.SimCardStatusChangeOutputItem]{
|
|
{Code: 0, Message: "ok", ResData: model.SimCardStatusChangeOutputItem{OrderNum: "resume-order"}},
|
|
}}
|
|
executor := NewRestartExecutorWithPublisher(repo, upstream, fakeRestartPublisher(t)).(*restartExecutor)
|
|
executor.now = func() *gtime.Time { return now }
|
|
|
|
if ok, err := executor.CanExecuteProcessing(context.Background(), repo.main); err != nil || !ok {
|
|
t.Fatalf("CanExecuteProcessing() = %t, %v; want true, nil", ok, err)
|
|
}
|
|
if err := executor.Execute(context.Background(), "RST-4"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if upstream.calls != 1 || upstream.input.ReqData.OperType != consts.SimCardStatusOperTypeStart {
|
|
t.Fatalf("upstream resume input = %+v calls=%d", upstream.input, upstream.calls)
|
|
}
|
|
if len(repo.saved) != 1 || repo.saved[0].stage != restartStageResume || !repo.saved[0].outcome.Success {
|
|
t.Fatalf("saved resume results = %+v", repo.saved)
|
|
}
|
|
if !repo.finalized {
|
|
t.Fatal("resume phase did not finalize the RST batch")
|
|
}
|
|
}
|
|
|
|
func restartMain(phase string, info *gjson.Json) *entity.BatchMain {
|
|
data := map[string]any{"restart": map[string]any{"phase": phase}}
|
|
if info != nil {
|
|
for key, value := range info.Map() {
|
|
if key == "restart" {
|
|
restart := data["restart"].(map[string]any)
|
|
for restartKey, restartValue := range gjson.New(value).Map() {
|
|
restart[restartKey] = restartValue
|
|
}
|
|
continue
|
|
}
|
|
data[key] = value
|
|
}
|
|
}
|
|
return &entity.BatchMain{
|
|
BatchNo: "RST",
|
|
BatchType: consts.BatchTypeSimCardRestart,
|
|
BatchAsyncStatus: consts.BatchAsyncStatusProcessing,
|
|
BatchInfo: gjson.New(data),
|
|
}
|
|
}
|
|
|
|
func restartStopTask() *Task {
|
|
return &Task{
|
|
Batch: Batch{
|
|
BatchNo: "RST",
|
|
BatchType: consts.BatchTypeSimCardRestart,
|
|
BatchInfo: gjson.New(map[string]any{"restart": map[string]any{"phase": consts.RestartPhaseCreated}}),
|
|
},
|
|
Operation: Operation{OperType: consts.SimCardStatusOperTypeStop, ExpectedStatus: consts.SimBossCardStatusActivated, TargetStatus: consts.SimBossCardStatusStopped},
|
|
Details: []*Detail{
|
|
{ID: 1, ICCID: "iccid-1", Authorized: true, Card: &Card{ID: 11, ChanKey: "channel", BossCardStatus: consts.SimBossCardStatusActivated}},
|
|
},
|
|
}
|
|
}
|
|
|
|
func restartResumeTask() *Task {
|
|
return &Task{
|
|
Batch: Batch{BatchNo: "RST", BatchType: consts.BatchTypeSimCardRestart},
|
|
Operation: Operation{OperType: consts.SimCardStatusOperTypeStart, ExpectedStatus: consts.SimBossCardStatusStopped, TargetStatus: consts.SimBossCardStatusActivated},
|
|
Details: []*Detail{
|
|
{ID: 1, ICCID: "iccid-1", Authorized: true, Card: &Card{ID: 11, ChanKey: "channel", BossCardStatus: consts.SimBossCardStatusStopped}},
|
|
},
|
|
}
|
|
}
|
|
|
|
type fakeRestartRepository struct {
|
|
main *entity.BatchMain
|
|
stopTask *Task
|
|
resumeTask *Task
|
|
validationStage string
|
|
saved []restartSavedResult
|
|
waited bool
|
|
resumeAt *gtime.Time
|
|
claimed bool
|
|
claimResult bool
|
|
loadedResume bool
|
|
finalized bool
|
|
}
|
|
|
|
type restartSavedResult struct {
|
|
stage string
|
|
outcome Outcome
|
|
}
|
|
|
|
func (r *fakeRestartRepository) FindRestartMain(context.Context, string) (*entity.BatchMain, error) {
|
|
return r.main, nil
|
|
}
|
|
|
|
func (r *fakeRestartRepository) LoadRestartStop(context.Context, string) (*Task, error) {
|
|
return r.stopTask, nil
|
|
}
|
|
|
|
func (r *fakeRestartRepository) LoadRestartResume(context.Context, string) (*Task, error) {
|
|
r.loadedResume = true
|
|
return r.resumeTask, nil
|
|
}
|
|
|
|
func (r *fakeRestartRepository) SaveRestartValidationFailure(_ context.Context, _ *Task, stage string) error {
|
|
r.validationStage = stage
|
|
return nil
|
|
}
|
|
|
|
func (r *fakeRestartRepository) SaveRestartResult(_ context.Context, _ *Task, detail *Detail, outcome Outcome, stage string) (Outcome, error) {
|
|
detail.ApplyOutcome(outcome)
|
|
r.saved = append(r.saved, restartSavedResult{stage: stage, outcome: outcome})
|
|
return outcome, nil
|
|
}
|
|
|
|
func (r *fakeRestartRepository) WaitRestartResume(_ context.Context, _ *Task, resumeAt *gtime.Time) error {
|
|
r.waited = true
|
|
r.resumeAt = resumeAt
|
|
return nil
|
|
}
|
|
|
|
func (r *fakeRestartRepository) ClaimRestartResume(context.Context, string, *gtime.Time) (bool, error) {
|
|
r.claimed = true
|
|
return r.claimResult, nil
|
|
}
|
|
|
|
func (r *fakeRestartRepository) FinalizeRestart(context.Context, string) error {
|
|
r.finalized = true
|
|
return nil
|
|
}
|
|
|
|
type recordingRestartPublisher struct {
|
|
calls int
|
|
delay time.Duration
|
|
message any
|
|
}
|
|
|
|
func (p *recordingRestartPublisher) PublishDelayed(_ context.Context, _, _ string, delay time.Duration, data any) error {
|
|
p.calls++
|
|
p.delay = delay
|
|
p.message = data
|
|
return nil
|
|
}
|
|
|
|
func fakeRestartPublisher(t *testing.T) RestartPublisher {
|
|
t.Helper()
|
|
return func(context.Context, string, string, time.Duration, any) error {
|
|
t.Fatal("publisher must not be called")
|
|
return nil
|
|
}
|
|
}
|