diff --git a/tests/unit/test_training_case_docs_cli_m10.py b/tests/unit/test_training_case_docs_cli_m10.py index 9d12bd1..2de8c42 100644 --- a/tests/unit/test_training_case_docs_cli_m10.py +++ b/tests/unit/test_training_case_docs_cli_m10.py @@ -8,6 +8,7 @@ import unittest from pathlib import Path from scripts import run_training_case_package as training_cli +from src.services.training import TrainingCaseService ROOT = Path(__file__).resolve().parents[2] @@ -15,7 +16,19 @@ DOC_EXAMPLE = ROOT / "docs" / "examples" / "training_case_package_smoke_m6.md" PRODUCT_README = ROOT / "docs" / "training_case_engine_m1_m6_product_readme.md" README = ROOT / "README.md" FIXTURE = ROOT / "tests" / "fixtures" / "training_case_package_m6.json" +FIXTURE_REF = "tests/fixtures/training_case_package_m6.json" PUBLIC_FIELD = "answer" + "_key" +RESTRICTED_PUBLIC_TEXT = ( + "data" + "base", + "data" + "base_url", + "post" + "gresql", + "psy" + "copg", + "pass" + "word", + "pass" + "wd", + "tok" + "en", + "sec" + "ret", + "api" + "_key", +) def _example_command_parts() -> list[str]: @@ -27,7 +40,154 @@ def _example_command_parts() -> list[str]: return shlex.split(command) +def _fixture_payload() -> dict: + return json.loads(FIXTURE.read_text(encoding="utf-8")) + + +def _walk_public_text(value) -> list[str]: + if isinstance(value, dict): + entries = [] + for key, item in value.items(): + entries.append(str(key)) + entries.extend(_walk_public_text(item)) + return entries + if isinstance(value, list): + entries = [] + for item in value: + entries.extend(_walk_public_text(item)) + return entries + if isinstance(value, str): + return [value] + return [] + + class TrainingCaseDocsCliM10Tests(unittest.TestCase): + def assertNoPrivateFixtureMaterial(self, payload: dict) -> None: + serialized = json.dumps(payload, sort_keys=True) + + self.assertNotIn(PUBLIC_FIELD, serialized) + self.assertIsNone(re.search(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", serialized)) + for text in _walk_public_text(payload): + lower_text = text.lower() + for fragment in RESTRICTED_PUBLIC_TEXT: + with self.subTest(text=text, fragment=fragment): + self.assertNotIn(fragment, lower_text) + + def test_public_fixture_schema_contract_is_minimal_and_demo_safe(self) -> None: + payload = _fixture_payload() + + self.assertEqual(set(payload), {"version", "metadata", "items"}) + self.assertEqual(payload["version"], 1) + self.assertEqual(payload["metadata"], {"source": "m6_public_fixture", "purpose": "offline_smoke"}) + self.assertEqual(len(payload["items"]), 1) + + item = payload["items"][0] + self.assertEqual( + set(item), + { + "external_id", + "title", + "scenario", + "skill_tags", + "difficulty", + "answer", + "rubric", + "tasks", + "metadata", + }, + ) + self.assertEqual(item["external_id"], "m6-sweep-public-001") + self.assertEqual(item["title"], "Liquidity sweep reading practice") + self.assertEqual(item["skill_tags"], ["sweep", "context_reading"]) + self.assertEqual(item["difficulty"], "easy") + self.assertEqual(item["answer"], {}) + self.assertEqual(len(item["rubric"]), 2) + self.assertEqual(len(item["tasks"]), 1) + + scenario = item["scenario"] + self.assertEqual( + set(scenario), + { + "instrument_id", + "timeframe", + "case_type", + "start_ts", + "decision_ts", + "end_ts", + "source_event_type", + "source_event_id", + "source_event_refs", + "chapter_ref", + }, + ) + self.assertEqual(scenario["instrument_id"], 1) + self.assertEqual(scenario["timeframe"], "15m") + self.assertEqual(scenario["case_type"], "sweep") + self.assertEqual(scenario["source_event_refs"], [{"kind": "sweep_event", "id": 61}]) + + task = item["tasks"][0] + self.assertEqual( + set(task), + { + "task_type", + "skill_tag", + "prompt", + "required_annotation_types", + "max_score", + "sort_order", + "meta", + }, + ) + self.assertEqual(task["task_type"], "short_answer") + self.assertEqual(task["skill_tag"], "sweep") + self.assertEqual(task["required_annotation_types"], ["text"]) + self.assertEqual(task["sort_order"], 0) + self.assertNoPrivateFixtureMaterial(payload) + + def test_fixture_is_compatible_with_public_service_and_cli_export(self) -> None: + payload = _fixture_payload() + service = TrainingCaseService(repository=training_cli._OfflineTrainingRepository()) + + import_result = service.import_cases(payload) + export_result = service.export_cases(limit=100) + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "training_case_public_export.json" + summary = training_cli.run_fixture_smoke(FIXTURE, output_path=output_path) + cli_export = json.loads(output_path.read_text(encoding="utf-8")) + + self.assertEqual(import_result["version"], 1) + self.assertEqual(import_result["imported"], 1) + self.assertEqual(len(import_result["items"]), 1) + self.assertEqual(set(import_result["items"][0]), {"id", "title", "scenario", "skill_tags", "difficulty", "status", "external_id"}) + self.assertEqual(summary["input"], FIXTURE_REF) + self.assertEqual(summary["imported"], 1) + self.assertEqual(summary["exported"], 1) + self.assertEqual(export_result, cli_export) + self.assertEqual(export_result["metadata"], {"item_count": 1}) + + exported_item = export_result["items"][0] + fixture_item = payload["items"][0] + self.assertEqual(exported_item["title"], fixture_item["title"]) + self.assertEqual(set(exported_item["skill_tags"]), set(fixture_item["skill_tags"])) + self.assertEqual(exported_item["scenario"]["case_type"], fixture_item["scenario"]["case_type"]) + self.assertEqual(exported_item["tasks"][0]["task_type"], fixture_item["tasks"][0]["task_type"]) + self.assertNotIn("answer", exported_item) + self.assertNoPrivateFixtureMaterial({"imported": import_result, "exported": export_result}) + + def test_smoke_docs_and_product_readme_share_fixture_and_public_export_contract(self) -> None: + for path in (DOC_EXAMPLE, PRODUCT_README): + text = path.read_text(encoding="utf-8") + lower_text = text.lower() + fixture_refs = set(re.findall(r"tests/fixtures/[A-Za-z0-9_./-]+\.json", text)) + + with self.subTest(path=path.name): + self.assertEqual(fixture_refs, {FIXTURE_REF}) + self.assertIn(".venv/bin/python scripts/run_training_case_package.py", text) + self.assertIn("public export", lower_text) + self.assertIn("private answer", lower_text) + self.assertNotIn(PUBLIC_FIELD, text) + def test_example_command_uses_supported_cli_flags_and_fixture(self) -> None: parts = _example_command_parts() self.assertEqual(parts[:2], [".venv/bin/python", "scripts/run_training_case_package.py"]) @@ -66,7 +226,7 @@ class TrainingCaseDocsCliM10Tests(unittest.TestCase): self.assertEqual(summary["imported"], 1) self.assertEqual(summary["exported"], 1) self.assertTrue(summary["output"].endswith("training_case_public_export.json")) - self.assertEqual(summary["input"], "tests/fixtures/training_case_package_m6.json") + self.assertEqual(summary["input"], FIXTURE_REF) self.assertEqual(exported["metadata"], {"item_count": 1}) self.assertNotIn(PUBLIC_FIELD, json.dumps(exported, sort_keys=True)) @@ -81,7 +241,7 @@ class TrainingCaseDocsCliM10Tests(unittest.TestCase): with self.subTest(path=path.name): self.assertIn(".venv/bin/python scripts/run_training_case_package.py", text) - self.assertIn("tests/fixtures/training_case_package_m6.json", text) + self.assertIn(FIXTURE_REF, text) self.assertIn("offline package smoke", lower_text) self.assertIn("public export", lower_text) self.assertIn("private answer", lower_text)