Janady13/missionmatch / internal/integration/bench_import_integration_test.go

1//go:build integration 2 3// WS9 bench ingestion, exercised against a REAL Postgres 16 + pgvector with RLS in force. It drives the 4// two WS9 seams the controller validates on staging: sourcing.IngestBenchRecord on the migrator pool (the 5// candidate, skills, the precomputed embed, and the bench_import_log row, all in one migrator tx) and 6// sourcing.TagCandidateMissions on the app pool inside RunInTenantTx (the mission tag write under RLS). It 7// does NOT shell out to the import-bench binary; it drives the library seams directly so the data layer 8// properties are proven against live Postgres. 9// 10// The proven invariants are: INVARIANT 3, clearance stays 'unknown' on import so a resume regex can never 11// score as held; the candidate skills resolve to canonical ontology rows (the Signals Intelligence case); 12// resume_embedding is populated when a vector is passed and NULL when nil; the mission tags land under RLS 13// and a second org cannot see them; one bench_import_log row per imported record carries the operator 14// legal_basis and the bench source; and a re run of the same record ENRICHES rather than duplicating. 15package integration 16 17import ( 18 "context" 19 "testing" 20 21 "github.com/google/uuid" 22 "github.com/jackc/pgx/v5" 23 pgvector "github.com/pgvector/pgvector-go" 24 25 "github.com/missionworkforce/missionmatch/internal/platform/benchimport" 26 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 27 "github.com/missionworkforce/missionmatch/internal/platform/embeddings" 28 "github.com/missionworkforce/missionmatch/internal/platform/extraction" 29 "github.com/missionworkforce/missionmatch/internal/platform/missionclass" 30 "github.com/missionworkforce/missionmatch/internal/platform/sourcing" 31) 32 33// benchTestLegalBasis is the operator provided controller basis the test records against every imported 34// record, mirroring the required -legal-basis flag. It is never the PDL string (invariant 7). 35const benchTestLegalBasis = "client provided cleared bench, client is the data controller" 36 37// importOneBenchRecord drives the WS9 ingest seams for one record exactly as the import command does: it 38// runs sourcing.IngestBenchRecord in one migrator transaction (candidate, skills, the precomputed vector, 39// and one bench_import_log row), commits, then tags the candidate's missions on the app pool inside 40// RunInTenantTx. The embed vector is precomputed by the CALLER (invariant 6), so a nil vector commits the 41// row without an embedding. It returns the resolved candidate id and whether the row was a new insert. 42func importOneBenchRecord(t *testing.T, org uuid.UUID, rec benchimport.BenchRecord, vec *pgvector.Vector, source, fileOrBatch string, idx missionclass.AliasIndex) (uuid.UUID, bool) { 43 t.Helper() 44 ctx := context.Background() 45 ext := extraction.NewDeterministic() 46 sp := rec.ToSourcedPerson() 47 48 tx, err := migrator.Begin(ctx) 49 if err != nil { 50 t.Fatalf("begin migrator tx: %v", err) 51 } 52 committed := false 53 defer func() { 54 if !committed { 55 _ = tx.Rollback(ctx) 56 } 57 }() 58 59 out, err := sourcing.IngestBenchRecord(ctx, dbgen.New(tx), org, sp, ext, vec, source, fileOrBatch, benchTestLegalBasis) 60 if err != nil { 61 t.Fatalf("IngestBenchRecord: %v", err) 62 } 63 if err := tx.Commit(ctx); err != nil { 64 t.Fatalf("commit migrator tx: %v", err) 65 } 66 committed = true 67 68 // Invariant 2: tag the candidate's missions on the APP pool inside RunInTenantTx after the candidate 69 // commits, with the org from WithOrg so the RLS WITH CHECK validates it. The headline carries the SIGINT 70 // token for a record whose JobTitle resolves to an intelligence mission. 71 headline := rec.JobTitle 72 if err := sourcing.TagCandidateMissions(ctxOrg(org), appDB, out.CandidateID, headline, sp.Skills, idx); err != nil { 73 t.Fatalf("TagCandidateMissions: %v", err) 74 } 75 return out.CandidateID, out.New 76} 77 78func benchImportLogCount(t *testing.T, org, candID uuid.UUID) int { 79 t.Helper() 80 var n int 81 if err := migrator.QueryRow(context.Background(), 82 "SELECT count(*) FROM bench_import_log WHERE organization_id=$1 AND candidate_id=$2", org, candID).Scan(&n); err != nil { 83 t.Fatalf("bench_import_log count: %v", err) 84 } 85 return n 86} 87 88// TestBenchImportIngestsUnderRLS proves the WS9 ingest seams end to end on live Postgres: a structured 89// record with a Signals Intelligence skill and a SIGINT title plus a resume fixture both become candidates 90// with the clearance unknown invariant held, the skill resolves to a canonical ontology row, the embed 91// vector lands only when passed, the mission tag is RLS scoped, and one compliance row is written per 92// record with the bench source and the operator legal basis. 93func TestBenchImportIngestsUnderRLS(t *testing.T) { 94 ctx := context.Background() 95 org, otherOrg := seedOrg(t), seedOrg(t) 96 // An empty AliasIndex: the SIGINT tag comes from the literal keyword in the title, so the test does not 97 // depend on the curated alias rows being present in the database under test (mirrors the WS7 tag test). 98 idx := missionclass.NewAliasIndex(map[string]string{}) 99 100 // A structured record with a Signals Intelligence skill and a SIGINT analyst title, embedded with a 101 // deterministic vector so resume_embedding is populated. The clearance code is carried for triage only 102 // and must never reach candidates.clearance_level (invariant 3). 103 emb := embeddings.NewDeterministic(384) 104 rawVec, _ := emb.Embed(ctx, "SIGINT Analyst | Signals Intelligence, Python, Splunk") 105 pv := pgvector.NewVector(rawVec) 106 csvRec := benchimport.BenchRecord{ 107 FullName: "Sam Signals", 108 Location: "Honolulu, HI", 109 JobTitle: "SIGINT Analyst", 110 Skills: []string{"Signals Intelligence", "Python", "Splunk"}, 111 Years: 8, 112 ClearanceCode: "ts_sci", // triage only, never written to candidates.clearance_level 113 ExternalID: "bench-csv-" + uuid.NewString(), 114 Email: "sam.signals." + uuid.NewString()[:8] + "@example.com", 115 } 116 csvID, csvNew := importOneBenchRecord(t, org, csvRec, &pv, "bench_csv", "bench_roster.csv", idx) 117 if !csvNew { 118 t.Fatalf("expected the CSV record to insert a new candidate") 119 } 120 121 // A second record from a resume fixture path, ingested WITHOUT a vector so resume_embedding stays NULL, 122 // proving the embed skip path commits the row (the embed-backfill only-missing fallback fills it later). 123 resumeRec := benchimport.BenchRecord{ 124 FullName: "Rita Resume", 125 Location: "Albuquerque, NM", 126 JobTitle: "Cyber Security Engineer", 127 Skills: []string{"AWS", "Kubernetes"}, 128 Years: 6, 129 ExternalID: "bench-file-" + uuid.NewString(), 130 } 131 resumeID, resumeNew := importOneBenchRecord(t, org, resumeRec, nil, "bench_file", "/bench/resumes/rita.pdf", idx) 132 if !resumeNew { 133 t.Fatalf("expected the resume record to insert a new candidate") 134 } 135 136 // ---- candidate rows: full_name set, clearance_level 'unknown' (INVARIANT 3) ---- 137 for _, tc := range []struct { 138 id uuid.UUID 139 name string 140 }{{csvID, "Sam Signals"}, {resumeID, "Rita Resume"}} { 141 var fullName, clearance string 142 if err := migrator.QueryRow(ctx, 143 "SELECT full_name, clearance_level FROM candidates WHERE id=$1", tc.id).Scan(&fullName, &clearance); err != nil { 144 t.Fatalf("read candidate %s: %v", tc.id, err) 145 } 146 if fullName != tc.name { 147 t.Errorf("candidate full_name = %q, want %q", fullName, tc.name) 148 } 149 if clearance != "unknown" { 150 t.Errorf("INVARIANT 3 BREACH: candidate %s clearance_level = %q, must stay 'unknown' on import", tc.id, clearance) 151 } 152 } 153 154 // ---- candidate_skills resolved to canonical ontology skills (Signals Intelligence) ---- 155 skills := skillNamesFor(t, csvID) 156 if !skills["Signals Intelligence"] { 157 t.Errorf("the Signals Intelligence skill did not resolve to a candidate_skills ontology row; got %v", skills) 158 } 159 for _, want := range []string{"Python", "Splunk"} { 160 if !skills[want] { 161 t.Errorf("structured skill %q not written to candidate_skills; got %v", want, skills) 162 } 163 } 164 // The skill is a real row in the global ontology skills table, reachable for matching. 165 var skillN int 166 if err := migrator.QueryRow(ctx, "SELECT count(*) FROM skills WHERE name='Signals Intelligence'").Scan(&skillN); err != nil { 167 t.Fatalf("count Signals Intelligence skill: %v", err) 168 } 169 if skillN == 0 { 170 t.Errorf("the Signals Intelligence ontology skill row was not created") 171 } 172 173 // ---- resume_embedding populated when a vector is passed, NULL when nil ---- 174 var hasVec bool 175 if err := migrator.QueryRow(ctx, "SELECT resume_embedding IS NOT NULL FROM candidates WHERE id=$1", csvID).Scan(&hasVec); err != nil { 176 t.Fatalf("read csv embedding presence: %v", err) 177 } 178 if !hasVec { 179 t.Errorf("resume_embedding must be populated when a vector is passed (invariant 6 in tx write)") 180 } 181 var resumeHasVec bool 182 if err := migrator.QueryRow(ctx, "SELECT resume_embedding IS NOT NULL FROM candidates WHERE id=$1", resumeID).Scan(&resumeHasVec); err != nil { 183 t.Fatalf("read resume embedding presence: %v", err) 184 } 185 if resumeHasVec { 186 t.Errorf("resume_embedding must be NULL when a nil vector is passed (the embed skip path)") 187 } 188 189 // ---- candidate_mission_tags present after TagCandidateMissions, RLS scoped ---- 190 var code, mechanism string 191 if err := appDB.RunInTenantTx(ctxOrg(org), func(ctx context.Context, tx pgx.Tx) error { 192 return tx.QueryRow(ctx, 193 "SELECT mission_code, mechanism FROM candidate_mission_tags WHERE candidate_id=$1 AND mission_code='SIGINT'", 194 csvID).Scan(&code, &mechanism) 195 }); err != nil { 196 t.Fatalf("read SIGINT tag for the bench candidate: %v", err) 197 } 198 if code != "SIGINT" || mechanism != missionclass.MechKeyword { 199 t.Errorf("SIGINT tag = code %q mechanism %q, want SIGINT/keyword", code, mechanism) 200 } 201 // A second org MUST NOT see the tag: a count under its session is zero (RLS default deny). 202 var foreignN int 203 if err := appDB.RunInTenantTx(ctxOrg(otherOrg), func(ctx context.Context, tx pgx.Tx) error { 204 return tx.QueryRow(ctx, "SELECT count(*) FROM candidate_mission_tags WHERE candidate_id=$1", csvID).Scan(&foreignN) 205 }); err != nil { 206 t.Fatalf("foreign org tag count: %v", err) 207 } 208 if foreignN != 0 { 209 t.Fatalf("ISOLATION BREACH: a second org saw %d of the bench candidate's mission tags (want 0)", foreignN) 210 } 211 212 // ---- one bench_import_log row per imported record with the legal basis and the bench source ---- 213 if n := benchImportLogCount(t, org, csvID); n != 1 { 214 t.Errorf("expected exactly one bench_import_log row for the csv record, got %d", n) 215 } 216 if n := benchImportLogCount(t, org, resumeID); n != 1 { 217 t.Errorf("expected exactly one bench_import_log row for the resume record, got %d", n) 218 } 219 var basis, source string 220 if err := migrator.QueryRow(ctx, 221 "SELECT legal_basis, source FROM bench_import_log WHERE candidate_id=$1", csvID).Scan(&basis, &source); err != nil { 222 t.Fatalf("read csv bench_import_log: %v", err) 223 } 224 if basis != benchTestLegalBasis { 225 t.Errorf("bench_import_log legal_basis = %q, want the operator basis %q", basis, benchTestLegalBasis) 226 } 227 if source != "bench_csv" { 228 t.Errorf("bench_import_log source = %q, want bench_csv", source) 229 } 230 var resumeSource string 231 if err := migrator.QueryRow(ctx, 232 "SELECT source FROM bench_import_log WHERE candidate_id=$1", resumeID).Scan(&resumeSource); err != nil { 233 t.Fatalf("read resume bench_import_log: %v", err) 234 } 235 if resumeSource != "bench_file" { 236 t.Errorf("resume bench_import_log source = %q, want bench_file", resumeSource) 237 } 238 239 // ---- dedup: re run the SAME csv record, assert it ENRICHES (no duplicate candidate) ---- 240 before := candidateOrgCount(t, org) 241 reID, reNew := importOneBenchRecord(t, org, csvRec, &pv, "bench_csv", "bench_roster.csv", idx) 242 if reNew { 243 t.Fatalf("dedup failed: re running the same record reported a NEW candidate, not an enrich") 244 } 245 if reID != csvID { 246 t.Fatalf("dedup resolved to a different candidate %s, want the original %s", reID, csvID) 247 } 248 if after := candidateOrgCount(t, org); after != before { 249 t.Fatalf("dedup failed: candidate count went %d to %d (a duplicate was created)", before, after) 250 } 251 // The re run wrote a second compliance row for the same candidate (one per imported record), proving the 252 // log is append friendly and audits every import event, while the candidate row was enriched not cloned. 253 if n := benchImportLogCount(t, org, csvID); n != 2 { 254 t.Errorf("expected two bench_import_log rows after the dedup re run (one per import), got %d", n) 255 } 256} 257 258// TestBenchImportDistinctRecordsWithoutExternalID is the regression guard for the empty external id collapse. 259// A bench file usually carries NO external id, and the candidates partial unique index is 260// (organization_id, external_id) WHERE external_id IS NOT NULL. Passing the external id as an empty STRING 261// (not NULL) made every external id free record conflict on (org, ”) and collapse onto one row. The fix 262// passes NULL when empty so identity is resolved only by the email then name plus location dedup. This test 263// imports two distinct people with no external id and asserts two distinct candidates, not one. 264func TestBenchImportDistinctRecordsWithoutExternalID(t *testing.T) { 265 org := seedOrg(t) 266 idx := missionclass.NewAliasIndex(nil) 267 268 recA := benchimport.BenchRecord{ 269 FullName: "Distinct Alpha", Location: "Honolulu HI", JobTitle: "Intelligence Analyst", 270 Skills: []string{"SIGINT"}, Email: "distinct.alpha@example.com", 271 } 272 recB := benchimport.BenchRecord{ 273 FullName: "Distinct Bravo", Location: "Albuquerque NM", JobTitle: "Systems Engineer", 274 Skills: []string{"Kubernetes"}, Email: "distinct.bravo@example.com", 275 } 276 277 idA, newA := importOneBenchRecord(t, org, recA, nil, "bench_csv", "roster.csv", idx) 278 idB, newB := importOneBenchRecord(t, org, recB, nil, "bench_csv", "roster.csv", idx) 279 280 if !newA || !newB { 281 t.Fatalf("two external id free records with distinct identity must both be NEW inserts, got newA=%v newB=%v", newA, newB) 282 } 283 if idA == idB { 284 t.Fatalf("REGRESSION: two external id free bench records collapsed onto one candidate %s (the empty external_id partial index bug)", idA) 285 } 286 if n := candidateOrgCount(t, org); n != 2 { 287 t.Fatalf("expected two distinct candidates from two external id free records, got %d", n) 288 } 289} 290