yovision/Sense/server/app/sense/provisioning/service_test.go

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