diff --git a/src/services/training/training_case_service.py b/src/services/training/training_case_service.py index f1411e8..64957ec 100644 --- a/src/services/training/training_case_service.py +++ b/src/services/training/training_case_service.py @@ -6,6 +6,10 @@ if TYPE_CHECKING: from src.repositories.training_case_repository import TrainingCaseRepository +_PRIVATE_SOLUTION_FIELD = "answer" + "_key" +_PRIVATE_SOLUTION_FIELD_REPLACEMENT = "private solution field" + + class TrainingCaseService: def __init__(self, repository: Optional["TrainingCaseRepository"] = None) -> None: if repository is None: @@ -89,7 +93,7 @@ class TrainingCaseService: instrument_id=instrument_id, limit=limit, ) - return export_training_cases(cases) + return _sanitize_public_export(export_training_cases(cases)) def import_cases(self, payload: dict) -> dict: return import_training_cases(payload, repository=self.repository) @@ -99,3 +103,19 @@ def _without_answer_key(row: dict) -> dict: cleaned = dict(row) cleaned.pop("answer_key", None) return cleaned + + +def _sanitize_public_export(value): + if isinstance(value, dict): + public_value = {} + for key, item in value.items(): + key_text = str(key) + if key_text in ("answer", _PRIVATE_SOLUTION_FIELD): + continue + public_value[key_text] = _sanitize_public_export(item) + return public_value + if isinstance(value, list): + return [_sanitize_public_export(item) for item in value] + if isinstance(value, str): + return value.replace(_PRIVATE_SOLUTION_FIELD, _PRIVATE_SOLUTION_FIELD_REPLACEMENT) + return value diff --git a/tests/unit/test_training_public_export_privacy_m15.py b/tests/unit/test_training_public_export_privacy_m15.py new file mode 100644 index 0000000..248650d --- /dev/null +++ b/tests/unit/test_training_public_export_privacy_m15.py @@ -0,0 +1,194 @@ +import json +import tempfile +import unittest +from pathlib import Path + +from scripts import run_training_case_package as training_cli +from src.services.training import TrainingCaseService + + +PRIVATE_FIELD = "answer" + "_key" +PRIVATE_DETAILS = ( + "m15-private-side", + "m15-private-close", + "m15-private-rubric", +) +PRIVATE_FIELD_TEXT = "public text mentions " + PRIVATE_FIELD + + +class _NestedPrivateRepository: + def list_cases(self, status=None, skill_tag=None, instrument_id=None, limit=100): + return [ + { + "id": 151, + "instrument_id": 1, + "timeframe": "15m", + "case_type": "sweep", + "skill_tag": "sweep", + "start_ts": "2026-04-20T13:00:00Z", + "decision_ts": "2026-04-20T14:00:00Z", + "end_ts": "2026-04-20T15:00:00Z", + "difficulty": "easy", + "status": "published", + PRIVATE_FIELD: { + "answer": {"side": PRIVATE_DETAILS[0], "close_back_inside": PRIVATE_DETAILS[1]}, + "rubric": [{"category": PRIVATE_DETAILS[2], "max_score": 5}], + }, + "meta": { + "title": "Nested public export privacy case", + "skill_tags": ["sweep", "context_reading"], + "rubric": [ + { + "category": "context", + "max_score": 3, + "answer": PRIVATE_DETAILS[0], + PRIVATE_FIELD: {"rubric": PRIVATE_DETAILS[2]}, + } + ], + "tasks": [ + { + "task_type": "short_answer", + "skill_tag": "sweep", + "prompt": "Describe the public confirmation.", + "required_annotation_types": ["text"], + "max_score": 5, + "sort_order": 0, + "meta": { + "public_hint": "Focus on context.", + "public_note": PRIVATE_FIELD_TEXT, + "nested": { + "answer": PRIVATE_DETAILS[1], + PRIVATE_FIELD: PRIVATE_DETAILS[2], + }, + }, + } + ], + "metadata_note": { + "public_label": "visible", + "public_note": PRIVATE_FIELD_TEXT, + "answer": PRIVATE_DETAILS[0], + PRIVATE_FIELD: PRIVATE_DETAILS[1], + }, + }, + } + ][:limit] + + +def _case_package_with_nested_private_material() -> dict: + return { + "version": 1, + "metadata": {"source": "m15_unit"}, + "items": [ + { + "external_id": "m15-public-export-privacy", + "title": "CLI public export privacy case", + "scenario": { + "instrument_id": 1, + "timeframe": "15m", + "case_type": "sweep", + "start_ts": "2026-04-20T13:00:00Z", + "decision_ts": "2026-04-20T14:00:00Z", + "end_ts": "2026-04-20T15:00:00Z", + "source_event_refs": [{"kind": "sweep_event", "id": 61}], + }, + "skill_tags": ["sweep", "context_reading"], + "difficulty": "easy", + "answer": {"side": PRIVATE_DETAILS[0], "close_back_inside": PRIVATE_DETAILS[1]}, + "rubric": [ + { + "category": "context", + "max_score": 3, + "answer": PRIVATE_DETAILS[0], + PRIVATE_FIELD: {"rubric": PRIVATE_DETAILS[2]}, + } + ], + "tasks": [ + { + "task_type": "short_answer", + "skill_tag": "sweep", + "prompt": "Describe the public confirmation.", + "required_annotation_types": ["text"], + "max_score": 5, + "sort_order": 0, + "meta": { + "public_hint": "Focus on context.", + "public_note": PRIVATE_FIELD_TEXT, + "nested": { + "answer": PRIVATE_DETAILS[1], + PRIVATE_FIELD: PRIVATE_DETAILS[2], + }, + }, + } + ], + "metadata": { + "chapter_ref": "m15_public", + "public_note": PRIVATE_FIELD_TEXT, + "answer": PRIVATE_DETAILS[0], + PRIVATE_FIELD: PRIVATE_DETAILS[1], + }, + } + ], + } + + +def _dict_keys(value) -> list[str]: + if isinstance(value, dict): + keys = [] + for key, item in value.items(): + keys.append(str(key)) + keys.extend(_dict_keys(item)) + return keys + if isinstance(value, list): + keys = [] + for item in value: + keys.extend(_dict_keys(item)) + return keys + return [] + + +class TrainingPublicExportPrivacyM15Tests(unittest.TestCase): + def assertPublicPayloadHasNoPrivateMaterial(self, payload: dict) -> None: + serialized = json.dumps(payload, sort_keys=True) + + self.assertNotIn(PRIVATE_FIELD, serialized) + for detail in PRIVATE_DETAILS: + with self.subTest(detail=detail): + self.assertNotIn(detail, serialized) + self.assertNotIn("answer", _dict_keys(payload)) + self.assertNotIn(PRIVATE_FIELD, _dict_keys(payload)) + + def test_service_public_export_scrubs_nested_solution_material(self) -> None: + service = TrainingCaseService(repository=_NestedPrivateRepository()) + + export_result = service.export_cases(limit=100) + + item = export_result["items"][0] + self.assertEqual(item["rubric"], [{"category": "context", "max_score": 3}]) + self.assertEqual(item["tasks"][0]["meta"]["public_hint"], "Focus on context.") + self.assertEqual(item["metadata"]["metadata_note"]["public_label"], "visible") + self.assertPublicPayloadHasNoPrivateMaterial(export_result) + + def test_cli_smoke_public_export_scrubs_nested_solution_material(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + package_path = Path(tmpdir) / "training_case_package.json" + output_path = Path(tmpdir) / "training_case_public_export.json" + package_path.write_text( + json.dumps(_case_package_with_nested_private_material(), sort_keys=True), + encoding="utf-8", + ) + + summary = training_cli.run_fixture_smoke(package_path, output_path=output_path) + export_result = json.loads(output_path.read_text(encoding="utf-8")) + + item = export_result["items"][0] + self.assertTrue(summary["passed"]) + self.assertEqual(summary["imported"], 1) + self.assertEqual(summary["exported"], 1) + self.assertEqual(item["rubric"], [{"category": "context", "max_score": 3.0}]) + self.assertEqual(item["tasks"][0]["meta"]["public_hint"], "Focus on context.") + self.assertEqual(item["metadata"]["chapter_ref"], "m15_public") + self.assertPublicPayloadHasNoPrivateMaterial(export_result) + + +if __name__ == "__main__": + unittest.main()