yuqei/ai-exchange
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: lock training package fixture contract

Browse Source
This commit is contained in:
yuqei
2026-04-30 21:45:18 -04:00
parent 9193d6ca21
commit a432945a18
1 changed files with 162 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_training_case_docs_cli_m10.py
+162 -2
View File
@@ -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)