package provisioning import ( "context" "encoding/json" "errors" "fmt" "strings" "sync" "sync/atomic" "testing" coreService "github.com/go-admin-team/go-admin-core/sdk/service" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgconn" "gorm.io/driver/sqlite" "gorm.io/gorm" deviceModels "git.ilapage.cn/ila/yovision/Sense/server/app/sense/device/models" ) func provisioningService(t *testing.T, quota int) *Service { t.Helper() db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{TranslateError: true}) if err != nil { t.Fatal(err) } sqlDB, err := db.DB() if err != nil { t.Fatal(err) } sqlDB.SetMaxOpenConns(1) if err = db.AutoMigrate(&deviceModels.Device{}, &Batch{}, &Item{}); err != nil { t.Fatal(err) } return &Service{Service: coreService.Service{Orm: db}, Quota: quota} } func TestCreateBatchValidatesQuotaAndIsIdempotent(t *testing.T) { service := provisioningService(t, 2) if err := service.Orm.Create(&deviceModels.Device{ID: "existing", Name: "已接入摄像机", Modality: "video", Status: "active", AdapterStatus: "ready", Version: 1}).Error; err != nil { t.Fatal(err) } request := CreateBatchRequest{ IdempotencyKey: "import-001", CreateBy: 7, Rows: []ImportRow{ {LineNumber: 1, Name: "东门摄像机", Location: "东门", Address: "http://192.0.2.10/onvif"}, {LineNumber: 2, Name: "重复地址", Location: "东门", Address: "http://192.0.2.10/onvif"}, {LineNumber: 3, Name: "西门摄像机", Location: "西门", Address: "https://192.0.2.11/onvif"}, }, } created, err := service.CreateBatch(request) if err != nil { t.Fatal(err) } if created.QuotaLimit != 2 || created.ExistingCount != 1 || created.ReadyCount != 1 || len(created.Items) != 3 { t.Fatalf("unexpected batch: %#v", created) } if created.Items[0].Status != ItemReady || created.Items[1].Status != ItemInvalid || created.Items[2].Status != ItemQuotaExceeded { t.Fatalf("unexpected item statuses: %#v", created.Items) } request.Rows = []ImportRow{{LineNumber: 1, Name: "不应覆盖", Address: "http://192.0.2.99/onvif"}} repeated, err := service.CreateBatch(request) if err != nil { t.Fatal(err) } if repeated.ID != created.ID || repeated.TotalCount != 3 || repeated.Items[0].Name != "东门摄像机" { t.Fatalf("idempotent replay changed the batch: %#v", repeated) } } func TestExecuteSupportsPartialSuccessAndFailedOnlyRetry(t *testing.T) { service := provisioningService(t, 16) created, err := service.CreateBatch(CreateBatchRequest{ IdempotencyKey: "execute-001", Rows: []ImportRow{ {LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"}, {LineNumber: 2, Name: "西门摄像机", Address: "http://192.0.2.11/onvif"}, }, }) if err != nil { t.Fatal(err) } calls := map[int]int{} service.Activator = func(_ context.Context, _ *Service, item *Item, input CredentialInput, _ int) (string, string, string, error) { calls[item.LineNumber]++ if input.ONVIFPassword != "temporary-secret" { t.Fatalf("activator did not receive the transient credential") } if item.LineNumber == 2 && calls[item.LineNumber] == 1 { return "device-2", "", "", errors.New("synthetic network failure containing temporary-secret") } return "device-" + string(rune('0'+item.LineNumber)), "ready", "验证通过", nil } credentials := make([]CredentialInput, 0, len(created.Items)) for _, item := range created.Items { credentials = append(credentials, CredentialInput{ItemID: item.ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true}) } partial, err := service.Execute(context.Background(), created.ID, "", false, ExecuteRequest{Credentials: credentials, UpdateBy: 8}) if err != nil { t.Fatal(err) } if partial.Status != BatchPartial || partial.SuccessCount != 1 || partial.FailureCount != 1 { t.Fatalf("expected partial success, got %#v", partial) } if partial.Items[1].Detail == "" || strings.Contains(partial.Items[1].Detail, "temporary-secret") { t.Fatalf("unsafe failure detail: %q", partial.Items[1].Detail) } retried, err := service.Execute(context.Background(), created.ID, "", true, ExecuteRequest{Credentials: []CredentialInput{{ItemID: created.Items[1].ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true}}, UpdateBy: 8}) if err != nil { t.Fatal(err) } if retried.Status != BatchSucceeded || retried.SuccessCount != 2 || calls[1] != 1 || calls[2] != 2 { t.Fatalf("failed-only retry was not idempotent: response=%#v calls=%#v", retried, calls) } encoded, err := json.Marshal(retried) if err != nil { t.Fatal(err) } if strings.Contains(string(encoded), "temporary-secret") || strings.Contains(string(encoded), "installer") { t.Fatalf("response contains credentials: %s", encoded) } var stored []Item if err = service.Orm.Find(&stored).Error; err != nil { t.Fatal(err) } storedJSON, _ := json.Marshal(stored) if strings.Contains(string(storedJSON), "temporary-secret") || strings.Contains(string(storedJSON), "installer") { t.Fatalf("provisioning rows contain credentials: %s", storedJSON) } } func TestExecuteRequiresCredentialsWithoutCallingActivator(t *testing.T) { service := provisioningService(t, 16) created, err := service.CreateBatch(CreateBatchRequest{IdempotencyKey: "missing-credentials", Rows: []ImportRow{{LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"}}}) if err != nil { t.Fatal(err) } called := false service.Activator = func(context.Context, *Service, *Item, CredentialInput, int) (string, string, string, error) { called = true return "", "", "", nil } result, err := service.Execute(context.Background(), created.ID, "", false, ExecuteRequest{}) if err != nil { t.Fatal(err) } if called || result.Status != BatchFailed || result.Items[0].FailureCode != "credentials_required" { t.Fatalf("missing credentials were not handled safely: %#v called=%v", result, called) } } func TestConcurrentCreateUsesOneIdempotentBatch(t *testing.T) { service := provisioningService(t, 16) sqlDB, err := service.Orm.DB() if err != nil { t.Fatal(err) } sqlDB.SetMaxOpenConns(1) request := CreateBatchRequest{IdempotencyKey: "concurrent-import", Rows: []ImportRow{{LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"}}} const workers = 8 ids := make(chan string, workers) errs := make(chan error, workers) var wait sync.WaitGroup for index := 0; index < workers; index++ { wait.Add(1) go func() { defer wait.Done() batch, createErr := service.CreateBatch(request) if createErr != nil { errs <- createErr return } ids <- batch.ID }() } wait.Wait() close(ids) close(errs) for createErr := range errs { t.Fatalf("concurrent create failed: %v", createErr) } var first string for id := range ids { if first == "" { first = id } else if id != first { t.Fatalf("idempotent creates returned different batches: %q and %q", first, id) } } var count int64 if err = service.Orm.Model(&Batch{}).Count(&count).Error; err != nil || count != 1 { t.Fatalf("batch count=%d err=%v", count, err) } } func TestSixteenDeviceBaselineSmoke(t *testing.T) { service := provisioningService(t, 16) rows := make([]ImportRow, 0, 16) for index := 1; index <= 16; index++ { rows = append(rows, ImportRow{LineNumber: index, Name: fmt.Sprintf("摄像机-%02d", index), Address: fmt.Sprintf("http://192.0.2.%d/onvif", index)}) } created, err := service.CreateBatch(CreateBatchRequest{IdempotencyKey: "sixteen-device-smoke", Rows: rows}) if err != nil { t.Fatal(err) } if created.ReadyCount != 16 { t.Fatalf("ready count=%d", created.ReadyCount) } service.Activator = func(_ context.Context, _ *Service, item *Item, _ CredentialInput, _ int) (string, string, string, error) { return "device-" + item.ID, "ready", "验证通过", nil } credentials := make([]CredentialInput, 0, 16) for _, item := range created.Items { credentials = append(credentials, CredentialInput{ItemID: item.ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true}) } completed, err := service.Execute(context.Background(), created.ID, "", false, ExecuteRequest{Credentials: credentials}) if err != nil { t.Fatal(err) } if completed.Status != BatchSucceeded || completed.SuccessCount != 16 || completed.FailureCount != 0 { t.Fatalf("unexpected 16-device result: %#v", completed) } } func TestConcurrentExecuteClaimsAnItemOnce(t *testing.T) { service := provisioningService(t, 16) created, err := service.CreateBatch(CreateBatchRequest{IdempotencyKey: "concurrent-execute", Rows: []ImportRow{{LineNumber: 1, Name: "东门摄像机", Address: "http://192.0.2.10/onvif"}}}) if err != nil { t.Fatal(err) } var calls atomic.Int32 service.Activator = func(_ context.Context, _ *Service, item *Item, _ CredentialInput, _ int) (string, string, string, error) { calls.Add(1) return "device-" + item.ID, "ready", "验证通过", nil } request := ExecuteRequest{Credentials: []CredentialInput{{ItemID: created.Items[0].ID, ONVIFUsername: "installer", ONVIFPassword: "temporary-secret", RTSPSameAsONVIF: true}}} var wait sync.WaitGroup errs := make(chan error, 2) for index := 0; index < 2; index++ { wait.Add(1) go func() { defer wait.Done() _, executeErr := service.Execute(context.Background(), created.ID, "", false, request) errs <- executeErr }() } wait.Wait() close(errs) for executeErr := range errs { if executeErr != nil { t.Fatal(executeErr) } } if calls.Load() != 1 { t.Fatalf("activator calls=%d", calls.Load()) } } func TestPostgresDuplicateKeyDetection(t *testing.T) { if !isDuplicateKey(&pgconn.PgError{Code: "23505"}) || isDuplicateKey(&pgconn.PgError{Code: "23503"}) { t.Fatal("PostgreSQL duplicate-key detection is incorrect") } }