Janady13/missionmatch / internal/integration/aioe_integration_test.go

1//go:build integration 2 3// AIOE isolation and closed loop integration proofs, against a REAL Postgres. These prove at 4// runtime what migration 0043 promises: the dedicated missionmatch_aioe role can read outcomes and 5// write profiles but is DENIED on eeo, on candidates, and on every other tenant table; the app 6// role records outcomes under RLS; and one updater pass produces a versioned, audited profile the 7// scorer then loads. 8// 9// Env: MM_AIOE_DSN (optional; these tests skip without it, exactly like the compliance DSN). 10package integration 11 12import ( 13 "context" 14 "encoding/json" 15 "fmt" 16 "os" 17 "testing" 18 "time" 19 20 "github.com/jackc/pgx/v5" 21 "github.com/jackc/pgx/v5/pgxpool" 22 23 "github.com/missionworkforce/missionmatch/internal/aioe" 24 "github.com/missionworkforce/missionmatch/internal/platform/database" 25) 26 27func aioePool(t *testing.T) *pgxpool.Pool { 28 t.Helper() 29 dsn := os.Getenv("MM_AIOE_DSN") 30 if dsn == "" { 31 t.Skip("MM_AIOE_DSN not set, skipping aioe isolation tests") 32 } 33 pool, err := database.NewPool(context.Background(), dsn) 34 if err != nil { 35 t.Fatalf("aioe pool: %v", err) 36 } 37 t.Cleanup(pool.Close) 38 if err := database.AssertRole(context.Background(), pool, "missionmatch_aioe"); err != nil { 39 t.Fatalf("aioe role assert: %v", err) 40 } 41 return pool 42} 43 44// TestAioeRoleIsDeniedEverywhereButItsOwnTables is THE guardrail proof: the updater's credentials 45// are physically unable to read protected or personal data. Every denial must be a permission 46// error (42501), not an empty result. 47func TestAioeRoleIsDeniedEverywhereButItsOwnTables(t *testing.T) { 48 pool := aioePool(t) 49 ctx := context.Background() 50 51 denied := []string{ 52 "SELECT 1 FROM eeo.candidate_demographics LIMIT 1", 53 "SELECT 1 FROM candidates LIMIT 1", 54 "SELECT 1 FROM candidate_evaluations LIMIT 1", 55 "SELECT 1 FROM requirements LIMIT 1", 56 "SELECT 1 FROM users LIMIT 1", 57 "SELECT 1 FROM auth_sessions LIMIT 1", 58 "SELECT 1 FROM slate_members LIMIT 1", 59 "SELECT 1 FROM audit_logs LIMIT 1", 60 "SELECT 1 FROM organizations LIMIT 1", 61 "INSERT INTO outcome_events (organization_id, module, subject_id, event_type, signal_value) VALUES (gen_random_uuid(), 'screen', gen_random_uuid(), 'screening_decision', 1)", 62 } 63 for _, q := range denied { 64 _, err := pool.Exec(ctx, q) 65 if pgCode(err) != "42501" { 66 t.Fatalf("aioe role must be DENIED (42501) on %q, got %v", q, err) 67 } 68 } 69 70 // And its own tables ARE readable (the grants are narrow, not broken). 71 allowed := []string{ 72 "SELECT count(*) FROM outcome_events", 73 "SELECT count(*) FROM aioe_weight_profiles", 74 "SELECT count(*) FROM aioe_weight_adjustments", 75 "SELECT count(*) FROM aioe_runs", 76 } 77 for _, q := range allowed { 78 if _, err := pool.Exec(ctx, q); err != nil { 79 t.Fatalf("aioe role should run %q, got %v", q, err) 80 } 81 } 82} 83 84// TestAioeClosedLoop drives the whole loop against the real database: the app role records enough 85// positive outcomes with denormalized dimension scores, one updater pass produces an ACTIVE 86// versioned profile with audit rows, and the scorer's loader returns the new profile for that org. 87func TestAioeClosedLoop(t *testing.T) { 88 pool := aioePool(t) 89 ctx := context.Background() 90 org := seedOrg(t) 91 92 // Record 40 positive outcomes through the APP role under RLS, the capture path shape: high 93 // skills score, low location score, so skills should rise and location fall. 94 scores := map[string]string{ 95 "clearance": "100", "skills": "95", "certifications": "50", 96 "customer_experience": "50", "experience": "60", "location": "10", 97 } 98 meta, _ := json.Marshal(map[string]any{"dimension_scores": scores}) 99 err := appDB.RunInTenantTx(ctxOrg(org), func(ctx context.Context, tx pgx.Tx) error { 100 for i := 0; i < 40; i++ { 101 if _, err := tx.Exec(ctx, ` 102 INSERT INTO outcome_events (organization_id, module, subject_id, event_type, signal_value, reason_code, occurred_at, metadata) 103 VALUES ($1, 'screen', gen_random_uuid(), 'screening_decision', 1, 'advance', now() - interval '1 day', $2)`, 104 org, meta); err != nil { 105 return err 106 } 107 } 108 return nil 109 }) 110 if err != nil { 111 t.Fatalf("record outcomes: %v", err) 112 } 113 114 updater := aioe.NewUpdater(pool, aioe.DefaultParams(), nil) 115 report, err := updater.RunAll(ctx, time.Now(), "admin") 116 if err != nil { 117 t.Fatalf("updater: %v", err) 118 } 119 if report.ProfilesUpdated < 1 { 120 t.Fatalf("expected at least one profile update, got %+v", report) 121 } 122 123 // The org now has exactly one ACTIVE profile with audit rows naming the window. 124 var version, adjustments int 125 if err := pool.QueryRow(ctx, ` 126 SELECT profile_version FROM aioe_weight_profiles 127 WHERE organization_id = $1 AND status = 'active'`, org).Scan(&version); err != nil { 128 t.Fatalf("active profile: %v", err) 129 } 130 if version < 1 { 131 t.Fatalf("active profile version = %d", version) 132 } 133 if err := pool.QueryRow(ctx, ` 134 SELECT count(*) FROM aioe_weight_adjustments 135 WHERE organization_id = $1 AND produced_version = $2`, org, version).Scan(&adjustments); err != nil { 136 t.Fatalf("adjustments: %v", err) 137 } 138 if adjustments < 1 { 139 t.Fatal("a produced profile must carry adjustment audit rows, nothing changes silently") 140 } 141 142 // The scorer's loader (app role, RLS scoped) picks the new profile up for this org. 143 err = appDB.RunInTenantTx(ctxOrg(org), func(ctx context.Context, tx pgx.Tx) error { 144 w, label, lerr := aioe.LoadActiveWeightsOnTx(ctx, tx) 145 if lerr != nil { 146 return lerr 147 } 148 if label != aioe.ProfileLabel(version) { 149 return fmt.Errorf("loader returned %q, want %q", label, aioe.ProfileLabel(version)) 150 } 151 if err := w.Validate(); err != nil { 152 return fmt.Errorf("loaded profile invalid: %w", err) 153 } 154 d := aioe.DefaultWeightsMap() 155 if w.Skills.LessThanOrEqual(d["skills"]) { 156 return fmt.Errorf("skills should have risen above the default, got %s", w.Skills) 157 } 158 return nil 159 }) 160 if err != nil { 161 t.Fatal(err) 162 } 163 164 // Determinism: a second pass over the SAME day and window changes nothing further only if the 165 // evidence is fully absorbed; what it must NEVER do is fail or produce an unaudited change. 166 before := version 167 if _, err := updater.RunAll(ctx, time.Now(), "admin"); err != nil { 168 t.Fatalf("second pass: %v", err) 169 } 170 var after int 171 if err := pool.QueryRow(ctx, ` 172 SELECT profile_version FROM aioe_weight_profiles 173 WHERE organization_id = $1 AND status = 'active'`, org).Scan(&after); err != nil { 174 t.Fatal(err) 175 } 176 if after > before { 177 var count int 178 if err := pool.QueryRow(ctx, ` 179 SELECT count(*) FROM aioe_weight_adjustments 180 WHERE organization_id = $1 AND produced_version = $2`, org, after).Scan(&count); err != nil || count == 0 { 181 t.Fatalf("version advanced to %d without audit rows (err %v)", after, err) 182 } 183 } 184 185} 186