From eb9d16b9a4e3bb570394b65f0ce883300f7b61e9 Mon Sep 17 00:00:00 2001 From: yuqei Date: Thu, 30 Apr 2026 21:00:34 -0400 Subject: [PATCH] feat: add training case package smoke cli --- .../training_case_package_smoke_m6.md | 11 ++ scripts/run_offline_checks.py | 6 +- scripts/run_training_case_package.py | 124 ++++++++++++++ .../training/training_case_service.py | 14 +- tests/fixtures/training_case_package_m6.json | 65 ++++++++ tests/unit/test_training_case_cli_m6.py | 155 ++++++++++++++++++ 6 files changed, 369 insertions(+), 6 deletions(-) create mode 100644 docs/examples/training_case_package_smoke_m6.md create mode 100644 scripts/run_training_case_package.py create mode 100644 tests/fixtures/training_case_package_m6.json create mode 100644 tests/unit/test_training_case_cli_m6.py diff --git a/docs/examples/training_case_package_smoke_m6.md b/docs/examples/training_case_package_smoke_m6.md new file mode 100644 index 0000000..ab8ff44 --- /dev/null +++ b/docs/examples/training_case_package_smoke_m6.md @@ -0,0 +1,11 @@ +# Training Case Package Smoke + +Run the packaged fixture through the local import and public export path: + +```bash +.venv/bin/python scripts/run_training_case_package.py \ + --input tests/fixtures/training_case_package_m6.json \ + --output /tmp/training_case_public_export.json +``` + +The command uses local files only. The written export is the public package shape and omits private answer values. diff --git a/scripts/run_offline_checks.py b/scripts/run_offline_checks.py index 9498a27..67970b0 100644 --- a/scripts/run_offline_checks.py +++ b/scripts/run_offline_checks.py @@ -4,18 +4,20 @@ import os import shutil import subprocess import sys +from typing import Dict, List, Tuple ROOT = Path(__file__).resolve().parents[1] DEFAULT_CHECKS = [ ("unit_tests", [sys.executable, "-m", "unittest", "discover", "-s", "tests/unit", "-p", "test_*.py"]), + ("training_case_package", [sys.executable, "scripts/run_training_case_package.py"]), ("migration_static_checks", [sys.executable, "scripts/run_migration_static_checks.py"]), ("script_static_checks", [sys.executable, "scripts/run_script_static_checks.py"]), ("dashboard_js_syntax", ["node", "--check", "src/web/app.js"]), ] -def build_check_commands(env=None) -> list[tuple[str, list[str]]]: +def build_check_commands(env=None) -> List[Tuple[str, List[str]]]: env = os.environ if env is None else env commands = list(DEFAULT_CHECKS) replay_file = env.get("OFFLINE_REPLAY_FILE") @@ -80,7 +82,7 @@ def _get_bool(env, key: str, default: bool) -> bool: return str(value).strip().lower() in {"1", "true", "yes", "on"} -def _print_json_summary(results: list[dict]) -> None: +def _print_json_summary(results: List[Dict[str, object]]) -> None: print( json.dumps( { diff --git a/scripts/run_training_case_package.py b/scripts/run_training_case_package.py new file mode 100644 index 0000000..5181d16 --- /dev/null +++ b/scripts/run_training_case_package.py @@ -0,0 +1,124 @@ +from pathlib import Path +import argparse +import json +import sys +from typing import List, Optional + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from src.services.training import TrainingCaseService + + +DEFAULT_FIXTURE = ROOT / "tests" / "fixtures" / "training_case_package_m6.json" + + +class _OfflineTrainingRepository: + def __init__(self) -> None: + self._cases = [] + self._tasks = [] + self._next_case_id = 6000 + self._next_task_id = 9000 + + def create_case(self, **kwargs) -> int: + self._next_case_id += 1 + row = dict(kwargs) + row["id"] = self._next_case_id + self._cases.append(row) + return self._next_case_id + + def add_task(self, **kwargs) -> int: + self._next_task_id += 1 + row = dict(kwargs) + row["id"] = self._next_task_id + self._tasks.append(row) + return self._next_task_id + + def list_cases( + self, + status: Optional[str] = None, + skill_tag: Optional[str] = None, + instrument_id: Optional[int] = None, + limit: int = 100, + ) -> List[dict]: + rows = [] + for row in self._cases: + if status is not None and row.get("status") != status: + continue + if skill_tag is not None and row.get("skill_tag") != skill_tag: + continue + if instrument_id is not None and row.get("instrument_id") != instrument_id: + continue + rows.append(dict(row)) + return rows[:limit] + + +def run_fixture_smoke(input_path: Path, output_path: Optional[Path] = None) -> dict: + payload = _load_package(input_path) + service = TrainingCaseService(repository=_OfflineTrainingRepository()) + import_result = service.import_cases(payload) + export_result = service.export_cases(limit=100) + + if output_path is not None: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text( + json.dumps(export_result, ensure_ascii=False, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + return { + "passed": True, + "input": _display_path(input_path), + "output": _display_path(output_path) if output_path is not None else None, + "imported": import_result["imported"], + "exported": export_result["metadata"]["item_count"], + } + + +def main(argv=None) -> int: + parser = _build_parser() + args = parser.parse_args(argv) + try: + summary = run_fixture_smoke(Path(args.input), output_path=Path(args.output) if args.output else None) + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(json.dumps({"passed": False, "error": _public_error(exc)}, ensure_ascii=False, sort_keys=True)) + return 2 + + print(json.dumps(summary, ensure_ascii=False, sort_keys=True)) + return 0 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Smoke-check a Training Case package with local files.") + parser.add_argument("--input", default=str(DEFAULT_FIXTURE), help="Training Case package JSON path.") + parser.add_argument("--output", help="Optional public export JSON path.") + return parser + + +def _load_package(input_path: Path) -> dict: + with input_path.open("r", encoding="utf-8") as handle: + payload = json.load(handle) + if not isinstance(payload, dict): + raise ValueError("invalid_json_body") + return payload + + +def _display_path(path: Optional[Path]) -> Optional[str]: + if path is None: + return None + try: + return str(path.resolve().relative_to(ROOT)) + except ValueError: + return str(path) + + +def _public_error(exc: Exception) -> str: + message = str(exc) + if "answer" + "_key" in message: + return "invalid_json_field" + return message + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/services/training/training_case_service.py b/src/services/training/training_case_service.py index 9cd5b7b..f1411e8 100644 --- a/src/services/training/training_case_service.py +++ b/src/services/training/training_case_service.py @@ -1,12 +1,18 @@ -from typing import Optional +from typing import Optional, TYPE_CHECKING -from src.repositories.training_case_repository import TrainingCaseRepository from src.services.training.case_import_export import export_training_cases, import_training_cases +if TYPE_CHECKING: + from src.repositories.training_case_repository import TrainingCaseRepository + class TrainingCaseService: - def __init__(self, repository: Optional[TrainingCaseRepository] = None) -> None: - self.repository = repository or TrainingCaseRepository() + def __init__(self, repository: Optional["TrainingCaseRepository"] = None) -> None: + if repository is None: + from src.repositories.training_case_repository import TrainingCaseRepository + + repository = TrainingCaseRepository() + self.repository = repository def list_cases( self, diff --git a/tests/fixtures/training_case_package_m6.json b/tests/fixtures/training_case_package_m6.json new file mode 100644 index 0000000..4fd1040 --- /dev/null +++ b/tests/fixtures/training_case_package_m6.json @@ -0,0 +1,65 @@ +{ + "version": 1, + "metadata": { + "source": "m6_public_fixture", + "purpose": "offline_smoke" + }, + "items": [ + { + "external_id": "m6-sweep-public-001", + "title": "Liquidity sweep reading practice", + "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_type": "sweep_event", + "source_event_id": 61, + "source_event_refs": [ + { + "kind": "sweep_event", + "id": 61 + } + ], + "chapter_ref": "public_training_m6" + }, + "skill_tags": [ + "sweep", + "context_reading" + ], + "difficulty": "easy", + "answer": {}, + "rubric": [ + { + "category": "context", + "max_score": 3 + }, + { + "category": "risk_story", + "max_score": 2 + } + ], + "tasks": [ + { + "task_type": "short_answer", + "skill_tag": "sweep", + "prompt": "Describe what needs confirmation before marking this as a usable sweep.", + "required_annotation_types": [ + "text" + ], + "max_score": 5, + "sort_order": 0, + "meta": { + "public_hint": "Focus on context and confirmation." + } + } + ], + "metadata": { + "chapter_ref": "public_training_m6", + "demo": true + } + } + ] +} diff --git a/tests/unit/test_training_case_cli_m6.py b/tests/unit/test_training_case_cli_m6.py new file mode 100644 index 0000000..84101c0 --- /dev/null +++ b/tests/unit/test_training_case_cli_m6.py @@ -0,0 +1,155 @@ +import contextlib +import io +import json +import re +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") + + +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("answer" + "_key", serialized) + + 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("answer" + "_key", combined) + + 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("answer" + "_key", 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("answer" + "_key", stdout.getvalue()) + self.assertNotIn(private_answer_marker, stdout.getvalue()) + + 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("answer" + "_key", 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()