241 lines
9.5 KiB
Python
241 lines
9.5 KiB
Python
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,
|
|
"status": "computed_prepost",
|
|
"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_kpi_metric_status_must_be_known(self) -> None:
|
|
root = self.make_root()
|
|
report_path = root / "02-measures" / "kpi_report.json"
|
|
data = json.loads(report_path.read_text(encoding="utf-8"))
|
|
data["metrics"]["sus"]["status"] = "maybe_later"
|
|
report_path.write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
report = checker.validate(root, max_scan_rows=100)
|
|
|
|
self.assertTrue(
|
|
any("metric 'sus' status 'maybe_later' is invalid" 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()
|