test: strengthen training package cli ux contract

This commit is contained in:
2026-05-01 11:34:23 -04:00
parent 2cdecf7b54
commit 4d5bca4dd9
2 changed files with 82 additions and 3 deletions
+17 -2
View File
@@ -79,10 +79,12 @@ def run_fixture_smoke(input_path: Path, output_path: Optional[Path] = None) -> d
def main(argv=None) -> int:
parser = _build_parser()
args = parser.parse_args(argv)
input_path = Path(args.input)
output_path = Path(args.output) if args.output else None
try:
summary = run_fixture_smoke(Path(args.input), output_path=Path(args.output) if args.output else None)
summary = run_fixture_smoke(input_path, output_path=output_path)
except (OSError, ValueError, json.JSONDecodeError) as exc:
print(json.dumps({"passed": False, "error": _public_error(exc)}, ensure_ascii=False))
print(json.dumps(_error_summary(input_path, output_path, exc), ensure_ascii=False))
return 2
print(json.dumps(summary, ensure_ascii=False))
@@ -114,11 +116,24 @@ def _display_path(path: Optional[Path]) -> Optional[str]:
def _public_error(exc: Exception) -> str:
if isinstance(exc, FileNotFoundError):
return "input_file_not_found"
if isinstance(exc, json.JSONDecodeError):
return "invalid_json_body"
message = str(exc)
if "answer" + "_key" in message:
return "invalid_json_field"
return message
def _error_summary(input_path: Path, output_path: Optional[Path], exc: Exception) -> dict:
return {
"passed": False,
"input": _display_path(input_path),
"output": _display_path(output_path),
"error": _public_error(exc),
}
if __name__ == "__main__":
raise SystemExit(main())
+65 -1
View File
@@ -2,6 +2,8 @@ import contextlib
import io
import json
import re
import shlex
import subprocess
import sys
import tempfile
import unittest
@@ -16,6 +18,14 @@ 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<command>.*?)\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 = []
@@ -114,11 +124,34 @@ class TrainingCaseCliM6Tests(unittest.TestCase):
self.assertRegex(success_line, r'^\{"passed": true, "input": ".+", "output": ".+", "imported": 1, "exported": 1\}$')
self.assertEqual(error_code, 2)
self.assertEqual(list(error_summary), ["passed", "error"])
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_exported_public_package_is_offline_projection_without_solution_fields(self) -> None:
fixture_item = json.loads(FIXTURE.read_text(encoding="utf-8"))["items"][0]
@@ -213,6 +246,37 @@ class TrainingCaseCliM6Tests(unittest.TestCase):
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={}))