import hashlib import importlib.util import json import sys import tempfile import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[3] CHECKER_PATH = REPO_ROOT / "scripts" / "check-phase3-artifacts.py" spec = importlib.util.spec_from_file_location("phase3_artifact_checker", CHECKER_PATH) assert spec is not None and spec.loader is not None checker = importlib.util.module_from_spec(spec) sys.modules[spec.name] = checker spec.loader.exec_module(checker) def write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() class Phase3ArtifactCheckerTests(unittest.TestCase): def make_root(self) -> Path: tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) root = Path(tmp.name) write_text( root / "00-intake" / "pilot_roster.csv", "participant_id,cohort_id,consent_version,consent_signed_at,withdrawal_state,enrolled_at\n" "P3-001,phase3,v1,2026-06-27T00:00:00Z,active,2026-06-27T00:00:00Z\n", ) write_text( root / "00-intake" / "consent_receipts.csv", "participant_id,consent_version,signed_at,signer_role,receipt_id\n" "P3-001,v1,2026-06-27T00:00:00Z,self,R-001\n", ) write_text( root / "01-sessions" / "session_completion.csv", "participant_id,session_id,persona_id,started_at,ended_at,completion_state,turns_count,supervisor_reviewed_at\n" "P3-001,S-001,P1,2026-06-27T00:00:00Z,2026-06-27T00:50:00Z,completed,12,2026-06-27T01:00:00Z\n", ) write_text( root / "02-measures" / "prepost_measures.csv", "participant_id,measure_name,timepoint,score,collected_at\n" "P3-001,self_efficacy,pre,3,2026-06-27T00:00:00Z\n" "P3-001,self_efficacy,post,4,2026-06-27T01:00:00Z\n", ) write_text( root / "02-measures" / "sus_responses.csv", "participant_id,item,response,collected_at\n" "P3-001,1,5,2026-06-27T01:00:00Z\n", ) write_text( root / "04-privacy" / "withdrawal_log.csv", "participant_id,requested_at,effective_at,scope,status,attestation_path\n", ) write_text( root / "04-privacy" / "privacy_audit.md", "# Privacy audit\n\nLegal/privacy reviewer: reviewer@example.invalid\n", ) write_text(root / "03-export" / "anonymized_dataset.jsonl", "{}\n") metrics = { name: { "value": 1, "threshold": 1, "pass": True, "numerator": 1, "denominator": 1, "method": "fixture", "source_files": ["fixture"], } for name in checker.KPI_METRICS } write_text( root / "02-measures" / "kpi_report.json", json.dumps( { "pilot_id": "phase3-fixture", "generated_at": "2026-06-27T00:00:00Z", "source_window": { "started_at": "2026-06-27T00:00:00Z", "ended_at": "2026-06-27T01:00:00Z", }, "cohort_size": 1, "metrics": metrics, "exclusions": [], "open_schema_gaps": [], "review": { "operator": "test", "reviewed_at": "2026-06-27T01:00:00Z", "decision": "fixture", }, }, ensure_ascii=False, ) + "\n", ) dataset_path = root / "03-export" / "anonymized_dataset.jsonl" write_text( root / "03-export" / "export_manifest.json", json.dumps( { "export_id": "phase3-fixture", "dataset_name": "vignette_phase3_recursive_learning_seed", "export_status": checker.APPROVED_EXPORT_STATUS, "created_at": "2026-06-27T00:00:00Z", "purpose": "test", "source_window": { "started_at": "2026-06-27T00:00:00Z", "ended_at": "2026-06-27T01:00:00Z", }, "source_tables": ["app.sessions"], "selection_criteria": { "include_withdrawn": False, "min_completed_sessions": 2, }, "consent_scope": { "allowed_uses": ["education_quality_review", "recursive_learning_seed"], "participants_included": 1, "participants_excluded": 0, }, "anonymization": { "text_transform": "masked_text_only", "direct_identifier_policy": "blocked", }, "pii_scan": {"status": "pass"}, "agreement": {"kappa": 0.60, "icc": 0.75}, "files": [ { "path": "03-export/anonymized_dataset.jsonl", "rows": 1, "sha256": sha256(dataset_path), "schema": "phase3_dataset_item_v1", } ], "approvals": { "data_steward": "steward", "legal_or_privacy_reviewer": "privacy", "technical_operator": "operator", "approved_at": "2026-06-27T01:00:00Z", }, "known_limitations": [], }, ensure_ascii=False, ) + "\n", ) return root def test_valid_approved_fixture_passes(self) -> None: root = self.make_root() report = checker.validate(root, max_scan_rows=100) self.assertEqual([], report.errors) def test_approved_manifest_requires_privacy_and_agreement_gates(self) -> None: root = self.make_root() manifest_path = root / "03-export" / "export_manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["pii_scan"]["status"] = "pending" manifest["agreement"]["kappa"] = 0.59 manifest["agreement"]["icc"] = 0.74 manifest["selection_criteria"]["include_withdrawn"] = True manifest["consent_scope"]["allowed_uses"] = ["education_quality_review"] manifest_path.write_text(json.dumps(manifest), encoding="utf-8") report = checker.validate(root, max_scan_rows=100) errors = "\n".join(report.errors) self.assertIn("pii_scan.status='pass'", errors) self.assertIn("agreement.kappa >= 0.60", errors) self.assertIn("agreement.icc >= 0.75", errors) self.assertIn("include_withdrawn=false", errors) self.assertIn("recursive_learning_seed consent scope", errors) def test_kpi_metric_required_fields_are_errors(self) -> None: root = self.make_root() report_path = root / "02-measures" / "kpi_report.json" data = json.loads(report_path.read_text(encoding="utf-8")) del data["metrics"]["sus"]["source_files"] report_path.write_text(json.dumps(data), encoding="utf-8") report = checker.validate(root, max_scan_rows=100) self.assertTrue( any("metric 'sus' missing 'source_files'" in error for error in report.errors), report.errors, ) def test_csv_enum_values_are_validated(self) -> None: root = self.make_root() write_text( root / "01-sessions" / "session_completion.csv", "participant_id,session_id,persona_id,started_at,ended_at,completion_state,turns_count,supervisor_reviewed_at\n" "P3-001,S-001,P1,2026-06-27T00:00:00Z,2026-06-27T00:50:00Z,done,12,2026-06-27T01:00:00Z\n", ) report = checker.validate(root, max_scan_rows=100) self.assertTrue( any("invalid completion_state 'done'" in error for error in report.errors), report.errors, ) def test_approved_manifest_hash_must_match_existing_file(self) -> None: root = self.make_root() manifest_path = root / "03-export" / "export_manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["files"][0]["sha256"] = "0" * 64 manifest_path.write_text(json.dumps(manifest), encoding="utf-8") report = checker.validate(root, max_scan_rows=100) self.assertTrue(any("sha256 mismatch" in error for error in report.errors), report.errors) if __name__ == "__main__": unittest.main()