import contextlib import io import json import re import shlex import subprocess import sys import tempfile import unittest from pathlib import Path from scripts import run_offline_checks from scripts import run_training_case_package as training_cli FIXTURE = Path("tests/fixtures/training_case_package_m6.json") DOC_SNIPPET = Path("docs/examples/training_case_package_smoke_m6.md") PRIVATE_FIELD = "answer" + "_key" def _documented_package_command_parts() -> list[str]: text = DOC_SNIPPET.read_text(encoding="utf-8") match = re.search(r"```bash\n(?P.*?)\n```", text, flags=re.DOTALL) if match is None: raise AssertionError("missing documented package smoke command") return shlex.split(re.sub(r"\\\n\s*", " ", match.group("command").strip())) def _mapping_keys(value) -> list[str]: if isinstance(value, dict): keys = [] for key, item in value.items(): keys.append(str(key)) keys.extend(_mapping_keys(item)) return keys if isinstance(value, list): keys = [] for item in value: keys.extend(_mapping_keys(item)) return keys return [] class TrainingCaseCliM6Tests(unittest.TestCase): def test_fixture_omits_internal_solution_key_or_private_answer_detail(self) -> None: payload = json.loads(FIXTURE.read_text(encoding="utf-8")) serialized = json.dumps(payload, sort_keys=True) self.assertEqual(payload["version"], 1) self.assertEqual(len(payload["items"]), 1) self.assertEqual(payload["items"][0]["answer"], {}) self.assertNotIn(PRIVATE_FIELD, serialized) def test_fixture_schema_has_complete_offline_sample_contract(self) -> None: payload = json.loads(FIXTURE.read_text(encoding="utf-8")) item = payload["items"][0] scenario = item["scenario"] task = item["tasks"][0] self.assertEqual(payload["metadata"], {"source": "m6_public_fixture", "purpose": "offline_smoke"}) self.assertRegex(item["external_id"], r"^m6-[a-z-]+-\d{3}$") self.assertEqual(len(item["skill_tags"]), len(set(item["skill_tags"]))) self.assertEqual(scenario["start_ts"], "2026-04-20T13:00:00Z") self.assertLess(scenario["start_ts"], scenario["decision_ts"]) self.assertLess(scenario["decision_ts"], scenario["end_ts"]) self.assertEqual(scenario["source_event_refs"], [{"kind": scenario["source_event_type"], "id": scenario["source_event_id"]}]) self.assertEqual(item["metadata"]["chapter_ref"], scenario["chapter_ref"]) self.assertEqual(task["skill_tag"], item["skill_tags"][0]) self.assertEqual(task["required_annotation_types"], ["text"]) self.assertEqual(task["sort_order"], 0) self.assertEqual( task["max_score"], sum(section["max_score"] for section in item["rubric"]), ) def test_cli_imports_and_exports_fixture_without_private_answer_detail(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: output_path = Path(tmpdir) / "public_training_case_package.json" stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = training_cli.main(["--input", str(FIXTURE), "--output", str(output_path)]) summary = json.loads(stdout.getvalue()) exported = json.loads(output_path.read_text(encoding="utf-8")) combined = json.dumps({"summary": summary, "exported": exported}, sort_keys=True) self.assertEqual(exit_code, 0) self.assertTrue(summary["passed"]) self.assertEqual(summary["imported"], 1) self.assertEqual(summary["exported"], 1) self.assertEqual(exported["metadata"], {"item_count": 1}) self.assertEqual(exported["items"][0]["title"], "Liquidity sweep reading practice") self.assertNotIn(PRIVATE_FIELD, combined) def test_cli_summary_uses_readable_key_order_for_success_and_error(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: output_path = Path(tmpdir) / "public_training_case_package.json" stdout = io.StringIO() with contextlib.redirect_stdout(stdout): success_code = training_cli.main(["--input", str(FIXTURE), "--output", str(output_path)]) success_line = stdout.getvalue().strip() success_summary = json.loads(success_line) input_path = Path(tmpdir) / "invalid_training_case_package.json" input_path.write_text(json.dumps({"version": 1, "items": [{"title": "Incomplete"}]}), encoding="utf-8") error_stdout = io.StringIO() with contextlib.redirect_stdout(error_stdout): error_code = training_cli.main(["--input", str(input_path)]) error_line = error_stdout.getvalue().strip() error_summary = json.loads(error_line) self.assertEqual(success_code, 0) self.assertEqual(list(success_summary), ["passed", "input", "output", "imported", "exported"]) self.assertTrue(success_summary["passed"]) self.assertEqual(success_summary["input"], str(FIXTURE)) self.assertEqual(success_summary["output"], str(output_path)) self.assertNotIn("\n", success_line) self.assertRegex(success_line, r'^\{"passed": true, "input": ".+", "output": ".+", "imported": 1, "exported": 1\}$') self.assertEqual(error_code, 2) self.assertEqual(list(error_summary), ["passed", "input", "output", "error"]) self.assertFalse(error_summary["passed"]) self.assertEqual(error_summary["input"], str(input_path)) self.assertIsNone(error_summary["output"]) self.assertEqual(error_summary["error"], "missing_required_json_field:items[0].scenario") self.assertNotIn("\n", error_line) def test_cli_error_summary_includes_user_paths_and_stable_missing_input_code(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: input_path = Path(tmpdir) / "missing_training_case_package.json" output_path = Path(tmpdir) / "public_training_case_package.json" stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = training_cli.main(["--input", str(input_path), "--output", str(output_path)]) line = stdout.getvalue().strip() summary = json.loads(line) self.assertEqual(exit_code, 2) self.assertEqual(list(summary), ["passed", "input", "output", "error"]) self.assertFalse(summary["passed"]) self.assertEqual(summary["input"], str(input_path)) self.assertEqual(summary["output"], str(output_path)) self.assertEqual(summary["error"], "input_file_not_found") self.assertNotIn("\n", line) self.assertNotIn("[Errno", line) def test_cli_error_summary_uses_stable_code_for_directory_input_path(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: input_path = Path(tmpdir) output_path = Path(tmpdir) / "public_training_case_package.json" stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = training_cli.main(["--input", str(input_path), "--output", str(output_path)]) line = stdout.getvalue().strip() summary = json.loads(line) self.assertEqual(exit_code, 2) self.assertEqual(list(summary), ["passed", "input", "output", "error"]) self.assertFalse(summary["passed"]) self.assertEqual(summary["input"], str(input_path)) self.assertEqual(summary["output"], str(output_path)) self.assertEqual(summary["error"], "input_path_is_directory") self.assertNotIn("\n", line) self.assertNotIn("[Errno", line) def test_cli_exported_public_package_is_offline_projection_without_solution_fields(self) -> None: fixture_item = json.loads(FIXTURE.read_text(encoding="utf-8"))["items"][0] with tempfile.TemporaryDirectory() as tmpdir: output_path = Path(tmpdir) / "public_training_case_package.json" stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = training_cli.main(["--input", str(FIXTURE), "--output", str(output_path)]) summary = json.loads(stdout.getvalue()) exported = json.loads(output_path.read_text(encoding="utf-8")) keys = _mapping_keys(exported) item = exported["items"][0] self.assertEqual(exit_code, 0) self.assertEqual(exported["metadata"]["item_count"], summary["exported"]) self.assertEqual( set(item), {"id", "title", "scenario", "skill_tags", "difficulty", "status", "rubric", "tasks", "metadata", "quality"}, ) self.assertEqual(item["title"], fixture_item["title"]) self.assertEqual(item["scenario"], fixture_item["scenario"]) self.assertEqual(item["rubric"], fixture_item["rubric"]) self.assertEqual(item["metadata"]["external_id"], fixture_item["external_id"]) self.assertEqual(item["quality"]["rank"], "high") self.assertEqual(item["tasks"][0]["status"], "active") self.assertEqual(item["tasks"][0]["meta"], fixture_item["tasks"][0]["meta"]) self.assertNotIn("answer", keys) self.assertNotIn(PRIVATE_FIELD, keys) def test_cli_export_omits_private_answer_detail_from_temp_payload(self) -> None: private_answer_marker = "-".join(("m6", "temporary", "answer", "detail")) payload = json.loads(FIXTURE.read_text(encoding="utf-8")) payload["items"][0]["answer"] = {"expected_label": private_answer_marker} with tempfile.TemporaryDirectory() as tmpdir: input_path = Path(tmpdir) / "training_case_package.json" output_path = Path(tmpdir) / "public_training_case_package.json" input_path.write_text(json.dumps(payload), encoding="utf-8") stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = training_cli.main(["--input", str(input_path), "--output", str(output_path)]) combined = json.dumps( { "summary": json.loads(stdout.getvalue()), "exported": json.loads(output_path.read_text(encoding="utf-8")), }, sort_keys=True, ) self.assertEqual(exit_code, 0) self.assertNotIn(PRIVATE_FIELD, combined) self.assertNotIn(private_answer_marker, combined) def test_cli_returns_nonzero_for_invalid_package(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: input_path = Path(tmpdir) / "invalid_training_case_package.json" input_path.write_text(json.dumps({"version": 1, "items": [{"title": "Incomplete"}]}), encoding="utf-8") stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = training_cli.main(["--input", str(input_path)]) summary = json.loads(stdout.getvalue()) self.assertEqual(exit_code, 2) self.assertFalse(summary["passed"]) self.assertIn("missing_required_json_field", summary["error"]) self.assertNotIn("traceback", stdout.getvalue().lower()) def test_cli_error_output_does_not_echo_internal_solution_field(self) -> None: private_answer_marker = "-".join(("m6", "temporary", "answer", "detail")) payload = json.loads(FIXTURE.read_text(encoding="utf-8")) payload["items"][0]["answer" + "_key"] = {"expected_label": private_answer_marker} with tempfile.TemporaryDirectory() as tmpdir: input_path = Path(tmpdir) / "invalid_training_case_package.json" input_path.write_text(json.dumps(payload), encoding="utf-8") stdout = io.StringIO() with contextlib.redirect_stdout(stdout): exit_code = training_cli.main(["--input", str(input_path)]) summary = json.loads(stdout.getvalue()) self.assertEqual(exit_code, 2) self.assertFalse(summary["passed"]) self.assertEqual(summary["error"], "invalid_json_field") self.assertNotIn(PRIVATE_FIELD, stdout.getvalue()) self.assertNotIn(private_answer_marker, stdout.getvalue()) def test_documented_package_command_runs_real_script_with_public_export_path(self) -> None: parts = _documented_package_command_parts() with tempfile.TemporaryDirectory() as tmpdir: output_path = Path(tmpdir) / "training_case_public_export.json" output_index = parts.index("--output") + 1 command = [sys.executable, *parts[1:]] command[output_index] = str(output_path) process = subprocess.run( command, cwd=Path.cwd(), capture_output=True, text=True, check=False, ) summary = json.loads(process.stdout) exported = json.loads(output_path.read_text(encoding="utf-8")) self.assertEqual(parts[:2], [".venv/bin/python", "scripts/run_training_case_package.py"]) self.assertEqual(process.returncode, 0) self.assertEqual(process.stderr, "") self.assertEqual(list(summary), ["passed", "input", "output", "imported", "exported"]) self.assertTrue(summary["passed"]) self.assertEqual(summary["input"], str(FIXTURE)) self.assertEqual(summary["output"], str(output_path)) self.assertEqual(summary["imported"], 1) self.assertEqual(summary["exported"], exported["metadata"]["item_count"]) self.assertEqual(exported["items"][0]["metadata"]["external_id"], "m6-sweep-public-001") def test_offline_checks_include_training_case_package_smoke(self) -> None: commands = dict(run_offline_checks.build_check_commands(env={})) self.assertEqual( commands["training_case_package"], [sys.executable, "scripts/run_training_case_package.py"], ) def test_docs_show_offline_fixture_command(self) -> None: text = DOC_SNIPPET.read_text(encoding="utf-8") self.assertIn("scripts/run_training_case_package.py", text) self.assertIn("tests/fixtures/training_case_package_m6.json", text) self.assertNotIn(PRIVATE_FIELD, text) def test_m6_sources_do_not_add_private_literals_or_trading_writes(self) -> None: paths = [ Path("scripts/run_training_case_package.py"), Path("scripts/run_offline_checks.py"), FIXTURE, DOC_SNIPPET, Path(__file__), ] source = "\n".join(path.read_text(encoding="utf-8") for path in paths if path.exists()).lower() for term in ( "data" + "base", "post" + "gresql", "psy" + "copg", "pass" + "word", "pass" + "wd", "tok" + "en", "sec" + "ret", "api" + "_key", ): self.assertNotIn(term, source) for table_name in ("trade_" + "signals", "trade_" + "executions", "backtest_" + "results"): forbidden = re.compile(rf"\b(insert\s+into|update|delete\s+from|truncate\s+table)\s+{table_name}\b") self.assertIsNone(forbidden.search(source), f"training M6 writes to {table_name}") if __name__ == "__main__": unittest.main()