Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
196 lines
10 KiB
Go
196 lines
10 KiB
Go
package privacy
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/cache"
|
|
"data-terminal-cloud/internal/database"
|
|
"data-terminal-cloud/internal/outbox"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
const (
|
|
privacyDeleteSubjectID = "90000000-0000-0000-0000-000000000741"
|
|
privacyDeleteSubjectPublicID = "90000000-0000-0000-0000-000000000742"
|
|
privacyDeleteRequestID = "90000000-0000-0000-0000-000000000743"
|
|
privacyDeleteRequestPublicID = "90000000-0000-0000-0000-000000000744"
|
|
)
|
|
|
|
type privacyDeleteStore struct {
|
|
removed []string
|
|
failure error
|
|
}
|
|
|
|
func (s *privacyDeleteStore) Remove(_ context.Context, reference string) error {
|
|
if s.failure != nil {
|
|
return s.failure
|
|
}
|
|
s.removed = append(s.removed, reference)
|
|
return nil
|
|
}
|
|
|
|
func TestDeleteProcessorReconcilesAllSubsystemsAndCreatesTombstone(t *testing.T) {
|
|
databaseURL := os.Getenv("TEST_DATABASE_URL")
|
|
redisAddress := os.Getenv("TEST_REDIS_ADDR")
|
|
if databaseURL == "" || redisAddress == "" {
|
|
t.Skip("TEST_DATABASE_URL and TEST_REDIS_ADDR are required")
|
|
}
|
|
ctx := context.Background()
|
|
pool, err := database.Open(ctx, databaseURL, "privacy-delete-test")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer pool.Close()
|
|
cleanupPrivacyDeleteFixture(t, ctx, pool)
|
|
defer cleanupPrivacyDeleteFixture(t, context.Background(), pool)
|
|
|
|
if _, err := pool.Exec(ctx, `INSERT INTO subject_profile (
|
|
id,public_id,tenant_id,household_id,preferred_name,status
|
|
) VALUES (
|
|
$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001',
|
|
'10000000-0000-0000-0000-000000000201','隐私删除集成长者','active'
|
|
)`, privacyDeleteSubjectID, privacyDeleteSubjectPublicID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := pool.Exec(ctx, `INSERT INTO privacy_request (
|
|
id,public_id,tenant_id,subject_id,request_type,scope,requested_by,idempotency_key,request_digest
|
|
) VALUES (
|
|
$1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001',$3::uuid,
|
|
'DELETE','{"reason_code":"subject_request"}'::jsonb,
|
|
'10000000-0000-0000-0000-000000000101','privacy-delete-integration',repeat('9',64)
|
|
)`, privacyDeleteRequestID, privacyDeleteRequestPublicID, privacyDeleteSubjectID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := pool.Exec(ctx, `INSERT INTO export_task (
|
|
id,public_id,tenant_id,subject_id,requested_by,scope,status,object_ref,object_size,content_digest,
|
|
expires_at,completed_at,idempotency_key,request_digest
|
|
) VALUES (
|
|
'90000000-0000-0000-0000-000000000745','90000000-0000-0000-0000-000000000746',
|
|
'10000000-0000-0000-0000-000000000001',$1::uuid,'10000000-0000-0000-0000-000000000101',
|
|
'{}'::jsonb,'READY','privacy-delete/export.zip',128,repeat('a',64),now()+interval '1 hour',now(),
|
|
'privacy-delete-export-integration',repeat('b',64)
|
|
)`, privacyDeleteSubjectID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := pool.Exec(ctx, `INSERT INTO privacy_subtask (
|
|
public_id,privacy_request_id,subsystem,target_type,target_ref
|
|
) SELECT gen_random_uuid(),$1::uuid,subsystem,'subject_data',subsystem
|
|
FROM unnest(ARRAY['redis','export_files','object_storage','pgvector','provider_index','postgresql','call_memory']) AS subsystem`, privacyDeleteRequestID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
redisClient := cache.Open(redisAddress, "", 0)
|
|
defer redisClient.Close()
|
|
store := &privacyDeleteStore{}
|
|
processor, err := NewDeleteProcessor(pool, redisClient, store)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
store.failure = outbox.RetryAfter(errors.New("provider deletion rate limited"), time.Minute)
|
|
if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err == nil || !outbox.IsRetryable(err) || outbox.RetryDelay(err) != time.Minute {
|
|
t.Fatalf("dependency failure lost retry metadata: %v", err)
|
|
}
|
|
var failureStatus, failedSubsystem, taskStatus, lastError string
|
|
var nextAttempt time.Time
|
|
if err := pool.QueryRow(ctx, `SELECT pr.status,pr.result_summary->>'failed_subsystem',ps.status,ps.last_error,ps.next_attempt_at FROM privacy_request pr JOIN privacy_subtask ps ON ps.privacy_request_id=pr.id AND ps.subsystem='export_files' WHERE pr.id=$1::uuid`, privacyDeleteRequestID).Scan(&failureStatus, &failedSubsystem, &taskStatus, &lastError, &nextAttempt); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if failureStatus != "PARTIAL_FAILED" || failedSubsystem != "export_files" || taskStatus != "FAILED" || lastError != "provider deletion rate limited" || time.Until(nextAttempt) < 55*time.Second {
|
|
t.Fatalf("failure was hidden or cooldown lost: request=%s subsystem=%s task=%s error=%q next=%v", failureStatus, failedSubsystem, taskStatus, lastError, nextAttempt)
|
|
}
|
|
// A dependency without Retry-After must use the same minimum cooldown in
|
|
// the durable receipt and the error returned to the outbox worker.
|
|
store.failure = errors.New("temporary object removal failure")
|
|
if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err == nil || outbox.RetryDelay(err) != 30*time.Second {
|
|
t.Fatalf("ordinary dependency failure did not preserve the receipt cooldown: %v", err)
|
|
}
|
|
var redisAttempts, exportAttempts int
|
|
if err := pool.QueryRow(ctx, `SELECT attempts FROM privacy_subtask WHERE privacy_request_id=$1::uuid AND subsystem='redis'`, privacyDeleteRequestID).Scan(&redisAttempts); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT attempts,next_attempt_at FROM privacy_subtask WHERE privacy_request_id=$1::uuid AND subsystem='export_files'`, privacyDeleteRequestID).Scan(&exportAttempts, &nextAttempt); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if redisAttempts != 1 || exportAttempts != 2 || time.Until(nextAttempt) < 25*time.Second {
|
|
t.Fatalf("retry repeated completed work or advertised a different cooldown: redis=%d export=%d next=%v", redisAttempts, exportAttempts, nextAttempt)
|
|
}
|
|
store.failure = nil
|
|
if _, err := pool.Exec(ctx, `UPDATE memory_user_state SET legacy_projection_pending=true WHERE subject_id=$1::uuid`, privacyDeleteSubjectID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err == nil || outbox.RetryDelay(err) != time.Minute {
|
|
t.Fatalf("unfinished call memory cleanup was acknowledged: %v", err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT pr.status,pr.result_summary->>'failed_subsystem',ps.last_error FROM privacy_request pr JOIN privacy_subtask ps ON ps.privacy_request_id=pr.id AND ps.subsystem='call_memory' WHERE pr.id=$1::uuid`, privacyDeleteRequestID).Scan(&failureStatus, &failedSubsystem, &lastError); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if failureStatus != "PARTIAL_FAILED" || failedSubsystem != "call_memory" || lastError != "call memory deletion pending: legacy_projection" {
|
|
t.Fatalf("cleanup recovery reason was hidden: status=%s subsystem=%s reason=%q", failureStatus, failedSubsystem, lastError)
|
|
}
|
|
if _, err := pool.Exec(ctx, `UPDATE memory_user_state SET legacy_projection_pending=false WHERE subject_id=$1::uuid`, privacyDeleteSubjectID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var requestStatus, subjectStatus, preferredName string
|
|
var completed, skipped, tombstones, audits int
|
|
if err := pool.QueryRow(ctx, `SELECT status FROM privacy_request WHERE id=$1::uuid`, privacyDeleteRequestID).Scan(&requestStatus); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT status,preferred_name FROM subject_profile WHERE id=$1::uuid`, privacyDeleteSubjectID).Scan(&subjectStatus, &preferredName); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FILTER(WHERE status='COMPLETED'),count(*) FILTER(WHERE status='SKIPPED') FROM privacy_subtask WHERE privacy_request_id=$1::uuid`, privacyDeleteRequestID).Scan(&completed, &skipped); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM tombstone WHERE deletion_request_id=$1::uuid`, privacyDeleteRequestID).Scan(&tombstones); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT count(*) FROM audit_event WHERE action='privacy.delete.complete' AND resource_id=$1::uuid`, privacyDeleteRequestPublicID).Scan(&audits); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var exportStatus string
|
|
var exportReference *string
|
|
if err := pool.QueryRow(ctx, `SELECT status,object_ref FROM export_task WHERE id='90000000-0000-0000-0000-000000000745'::uuid`).Scan(&exportStatus, &exportReference); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if requestStatus != "COMPLETED" || subjectStatus != "deleted" || preferredName != "[已删除]" || completed != 6 || skipped != 1 || tombstones != 1 || audits != 1 || len(store.removed) != 1 || store.removed[0] != "privacy-delete/export.zip" || exportStatus != "EXPIRED" || exportReference != nil {
|
|
t.Fatalf("request=%s subject=%s name=%q completed=%d skipped=%d tombstones=%d audits=%d removed=%v", requestStatus, subjectStatus, preferredName, completed, skipped, tombstones, audits, store.removed)
|
|
}
|
|
if err := processor.Handle(ctx, outbox.Event{AggregateID: privacyDeleteRequestID}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := pool.QueryRow(ctx, `SELECT attempts FROM privacy_subtask WHERE privacy_request_id=$1::uuid AND subsystem='export_files'`, privacyDeleteRequestID).Scan(&exportAttempts); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if exportAttempts != 3 || len(store.removed) != 1 {
|
|
t.Fatalf("completed deletion was executed again: attempts=%d removed=%d", exportAttempts, len(store.removed))
|
|
}
|
|
if err := processor.Handle(ctx, outbox.Event{AggregateID: "90000000-0000-0000-0000-000000000799"}); err != nil {
|
|
t.Fatalf("missing deletion request error=%v", err)
|
|
}
|
|
}
|
|
|
|
func cleanupPrivacyDeleteFixture(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
|
|
t.Helper()
|
|
statements := []string{
|
|
`DELETE FROM tombstone WHERE deletion_request_id='90000000-0000-0000-0000-000000000743'::uuid`,
|
|
`DELETE FROM audit_event WHERE action='privacy.delete.complete' AND resource_id='90000000-0000-0000-0000-000000000744'::uuid`,
|
|
`DELETE FROM privacy_request WHERE id='90000000-0000-0000-0000-000000000743'::uuid`,
|
|
`DELETE FROM export_task WHERE id='90000000-0000-0000-0000-000000000745'::uuid`,
|
|
`DELETE FROM memory_user_state WHERE subject_id='90000000-0000-0000-0000-000000000741'::uuid`,
|
|
`DELETE FROM subject_profile WHERE id='90000000-0000-0000-0000-000000000741'::uuid`,
|
|
}
|
|
for _, statement := range statements {
|
|
if _, err := pool.Exec(ctx, statement); err != nil {
|
|
t.Fatalf("cleanup privacy delete fixture: %v", err)
|
|
}
|
|
}
|
|
}
|