156 lines
6.4 KiB
Python
156 lines
6.4 KiB
Python
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()
|