story/internal/device/command_integration_test.go

package device

import (
	"context"
	"os"
	"testing"

	"data-terminal-cloud/internal/deviceauth"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

func TestRepositoryPullRedeliverAcknowledgeCompleteAndExpireCommands(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	repository := NewRepository(pool)
	identity := deviceauth.Identity{
		ID: "10000000-0000-0000-0000-000000000501", PublicID: "10000000-0000-0000-0000-000000000511",
		TenantID: "10000000-0000-0000-0000-000000000001", Status: "registered",
	}
	successID, successPublicID := uuid.NewString(), uuid.NewString()
	failureID, failurePublicID := uuid.NewString(), uuid.NewString()
	expiredID, expiredPublicID := uuid.NewString(), uuid.NewString()
	keyPrefix := "device-command-core-" + uuid.NewString()
	defer func() {
		_, _ = pool.Exec(ctx, `DELETE FROM device_command WHERE id=ANY($1::uuid[])`, []string{successID, failureID, expiredID})
	}()
	insert := func(id, publicID, suffix, commandType, expires string) {
		t.Helper()
		if _, err := pool.Exec(ctx, `INSERT INTO device_command(id,public_id,tenant_id,device_id,command_type,payload,status,requested_by,idempotency_key,request_digest,expires_at)
			VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5,$6::jsonb,'QUEUED','10000000-0000-0000-0000-000000000101'::uuid,$7,repeat('d',64),`+expires+`)`,
			id, publicID, identity.TenantID, identity.ID, commandType, `{"fixture":"`+suffix+`"}`, keyPrefix+"-"+suffix); err != nil {
			t.Fatal(err)
		}
	}
	insert(successID, successPublicID, "success", "apply_config", "now()+interval '1 hour'")
	insert(failureID, failurePublicID, "failure", "diagnostics", "now()+interval '1 hour'")
	insert(expiredID, expiredPublicID, "expired", "reboot", "now()-interval '1 second'")

	items, err := repository.PullCommands(ctx, identity, 0)
	if err != nil || len(items) != 2 {
		t.Fatalf("first pull items=%+v err=%v", items, err)
	}
	for _, item := range items {
		if item.Status != "DELIVERED" || item.DeliveryAttempts != 1 {
			t.Fatalf("first delivery item=%+v", item)
		}
	}
	items, err = repository.PullCommands(ctx, identity, 99)
	if err != nil || len(items) != 2 {
		t.Fatalf("redelivery items=%+v err=%v", items, err)
	}
	for _, item := range items {
		if item.Status != "DELIVERED" || item.DeliveryAttempts != 2 {
			t.Fatalf("redelivery item=%+v", item)
		}
	}
	var expiredStatus string
	if err := pool.QueryRow(ctx, `SELECT status FROM device_command WHERE id=$1::uuid`, expiredID).Scan(&expiredStatus); err != nil || expiredStatus != "EXPIRED" {
		t.Fatalf("expired status=%s err=%v", expiredStatus, err)
	}

	acknowledged, err := repository.AcknowledgeCommand(ctx, identity, successPublicID, CommandAcknowledgement{Status: "ACKNOWLEDGED", ResultSummary: map[string]any{"result_code": "RECEIVED"}})
	if err != nil || acknowledged.Status != "ACKNOWLEDGED" || acknowledged.DeliveryAttempts != 2 {
		t.Fatalf("acknowledged=%+v err=%v", acknowledged, err)
	}
	repeatedAck, err := repository.AcknowledgeCommand(ctx, identity, successPublicID, CommandAcknowledgement{Status: "ACKNOWLEDGED", ResultSummary: map[string]any{"result_code": "RECEIVED"}})
	if err != nil || repeatedAck.Status != "ACKNOWLEDGED" {
		t.Fatalf("repeated acknowledgement=%+v err=%v", repeatedAck, err)
	}
	items, err = repository.PullCommands(ctx, identity, 10)
	if err != nil || len(items) != 2 {
		t.Fatalf("in-flight recovery pull items=%+v err=%v", items, err)
	}
	for _, item := range items {
		if item.DeliveryAttempts != 3 {
			t.Fatalf("in-flight recovery attempts item=%+v", item)
		}
		if item.ID == successPublicID && item.Status != "ACKNOWLEDGED" {
			t.Fatalf("acknowledged command regressed during recovery pull: %+v", item)
		}
		if item.ID == failurePublicID && item.Status != "DELIVERED" {
			t.Fatalf("delivered command changed unexpectedly during recovery pull: %+v", item)
		}
	}
	completed, err := repository.AcknowledgeCommand(ctx, identity, successPublicID, CommandAcknowledgement{Status: "SUCCEEDED", ResultSummary: map[string]any{"result_code": "APPLIED"}})
	if err != nil || completed.Status != "SUCCEEDED" {
		t.Fatalf("completed=%+v err=%v", completed, err)
	}
	failed, err := repository.AcknowledgeCommand(ctx, identity, failurePublicID, CommandAcknowledgement{Status: "FAILED", ResultSummary: map[string]any{"result_code": "DIAGNOSTIC_FAILED"}})
	if err != nil || failed.Status != "FAILED" {
		t.Fatalf("failed=%+v err=%v", failed, err)
	}
	if _, err := repository.AcknowledgeCommand(ctx, identity, successPublicID, CommandAcknowledgement{Status: "FAILED", ResultSummary: map[string]any{}}); err == nil {
		t.Fatal("terminal command changed from SUCCEEDED to FAILED")
	}
	wrongDevice := identity
	wrongDevice.ID = uuid.NewString()
	if _, err := repository.AcknowledgeCommand(ctx, wrongDevice, failurePublicID, CommandAcknowledgement{Status: "FAILED", ResultSummary: map[string]any{}}); err == nil {
		t.Fatal("different device acknowledged command")
	}
	if _, err := repository.AcknowledgeCommand(ctx, identity, failurePublicID, CommandAcknowledgement{Status: "FAILED", ResultSummary: map[string]any{"invalid": make(chan int)}}); err == nil {
		t.Fatal("unencodable result summary accepted")
	}
	var successResult, failureResult string
	if err := pool.QueryRow(ctx, `SELECT result_summary->>'result_code' FROM device_command WHERE id=$1::uuid`, successID).Scan(&successResult); err != nil || successResult != "APPLIED" {
		t.Fatalf("success result=%q err=%v", successResult, err)
	}
	if err := pool.QueryRow(ctx, `SELECT result_summary->>'result_code' FROM device_command WHERE id=$1::uuid`, failureID).Scan(&failureResult); err != nil || failureResult != "DIAGNOSTIC_FAILED" {
		t.Fatalf("failure result=%q err=%v", failureResult, err)
	}
}