213 lines
8.5 KiB
Python
213 lines
8.5 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from scripts import run_training_case_package as training_cli
|
|
from src.services.training import TrainingCaseService
|
|
|
|
|
|
PRIVATE_FIELD = "answer" + "_key"
|
|
PRIVATE_DETAILS = (
|
|
"m15-private-side",
|
|
"m15-private-close",
|
|
"m15-private-rubric",
|
|
)
|
|
PRIVATE_FIELD_TEXT = "public text mentions " + PRIVATE_FIELD
|
|
|
|
|
|
class _NestedPrivateRepository:
|
|
def list_cases(self, status=None, skill_tag=None, instrument_id=None, limit=100):
|
|
return [
|
|
{
|
|
"id": 151,
|
|
"instrument_id": 1,
|
|
"timeframe": "15m",
|
|
"case_type": "sweep",
|
|
"skill_tag": "sweep",
|
|
"start_ts": "2026-04-20T13:00:00Z",
|
|
"decision_ts": "2026-04-20T14:00:00Z",
|
|
"end_ts": "2026-04-20T15:00:00Z",
|
|
"difficulty": "easy",
|
|
"status": "published",
|
|
PRIVATE_FIELD: {
|
|
"answer": {"side": PRIVATE_DETAILS[0], "close_back_inside": PRIVATE_DETAILS[1]},
|
|
"rubric": [{"category": PRIVATE_DETAILS[2], "max_score": 5}],
|
|
},
|
|
"meta": {
|
|
"title": "Nested public export privacy case",
|
|
"skill_tags": ["sweep", "context_reading"],
|
|
"rubric": [
|
|
{
|
|
"category": "context",
|
|
"max_score": 3,
|
|
"answer": PRIVATE_DETAILS[0],
|
|
PRIVATE_FIELD: {"rubric": PRIVATE_DETAILS[2]},
|
|
}
|
|
],
|
|
"tasks": [
|
|
{
|
|
"task_type": "short_answer",
|
|
"skill_tag": "sweep",
|
|
"prompt": "Describe the public confirmation.",
|
|
"required_annotation_types": ["text"],
|
|
"max_score": 5,
|
|
"sort_order": 0,
|
|
"meta": {
|
|
"public_hint": "Focus on context.",
|
|
"public_note": PRIVATE_FIELD_TEXT,
|
|
"nested": {
|
|
"answer": PRIVATE_DETAILS[1],
|
|
PRIVATE_FIELD: PRIVATE_DETAILS[2],
|
|
},
|
|
},
|
|
}
|
|
],
|
|
"metadata_note": {
|
|
"public_label": "visible",
|
|
"public_note": PRIVATE_FIELD_TEXT,
|
|
"answer": PRIVATE_DETAILS[0],
|
|
PRIVATE_FIELD: PRIVATE_DETAILS[1],
|
|
},
|
|
},
|
|
}
|
|
][:limit]
|
|
|
|
def get_case(self, case_id: int, **kwargs):
|
|
return self.list_cases(limit=1)[0] | {"id": case_id}
|
|
|
|
|
|
def _case_package_with_nested_private_material() -> dict:
|
|
return {
|
|
"version": 1,
|
|
"metadata": {"source": "m15_unit"},
|
|
"items": [
|
|
{
|
|
"external_id": "m15-public-export-privacy",
|
|
"title": "CLI public export privacy case",
|
|
"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_refs": [{"kind": "sweep_event", "id": 61}],
|
|
},
|
|
"skill_tags": ["sweep", "context_reading"],
|
|
"difficulty": "easy",
|
|
"answer": {"side": PRIVATE_DETAILS[0], "close_back_inside": PRIVATE_DETAILS[1]},
|
|
"rubric": [
|
|
{
|
|
"category": "context",
|
|
"max_score": 3,
|
|
"answer": PRIVATE_DETAILS[0],
|
|
PRIVATE_FIELD: {"rubric": PRIVATE_DETAILS[2]},
|
|
}
|
|
],
|
|
"tasks": [
|
|
{
|
|
"task_type": "short_answer",
|
|
"skill_tag": "sweep",
|
|
"prompt": "Describe the public confirmation.",
|
|
"required_annotation_types": ["text"],
|
|
"max_score": 5,
|
|
"sort_order": 0,
|
|
"meta": {
|
|
"public_hint": "Focus on context.",
|
|
"public_note": PRIVATE_FIELD_TEXT,
|
|
"nested": {
|
|
"answer": PRIVATE_DETAILS[1],
|
|
PRIVATE_FIELD: PRIVATE_DETAILS[2],
|
|
},
|
|
},
|
|
}
|
|
],
|
|
"metadata": {
|
|
"chapter_ref": "m15_public",
|
|
"public_note": PRIVATE_FIELD_TEXT,
|
|
"answer": PRIVATE_DETAILS[0],
|
|
PRIVATE_FIELD: PRIVATE_DETAILS[1],
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _dict_keys(value) -> list[str]:
|
|
if isinstance(value, dict):
|
|
keys = []
|
|
for key, item in value.items():
|
|
keys.append(str(key))
|
|
keys.extend(_dict_keys(item))
|
|
return keys
|
|
if isinstance(value, list):
|
|
keys = []
|
|
for item in value:
|
|
keys.extend(_dict_keys(item))
|
|
return keys
|
|
return []
|
|
|
|
|
|
class TrainingPublicExportPrivacyM15Tests(unittest.TestCase):
|
|
def assertPublicPayloadHasNoPrivateMaterial(self, payload: dict) -> None:
|
|
serialized = json.dumps(payload, sort_keys=True)
|
|
|
|
self.assertNotIn(PRIVATE_FIELD, serialized)
|
|
for detail in PRIVATE_DETAILS:
|
|
with self.subTest(detail=detail):
|
|
self.assertNotIn(detail, serialized)
|
|
self.assertNotIn("answer", _dict_keys(payload))
|
|
self.assertNotIn(PRIVATE_FIELD, _dict_keys(payload))
|
|
|
|
def test_service_public_export_scrubs_nested_solution_material(self) -> None:
|
|
service = TrainingCaseService(repository=_NestedPrivateRepository())
|
|
|
|
export_result = service.export_cases(limit=100)
|
|
|
|
item = export_result["items"][0]
|
|
self.assertEqual(item["rubric"], [{"category": "context", "max_score": 3}])
|
|
self.assertEqual(item["tasks"][0]["meta"]["public_hint"], "Focus on context.")
|
|
self.assertEqual(item["metadata"]["metadata_note"]["public_label"], "visible")
|
|
self.assertIn("quality", item)
|
|
self.assertPublicPayloadHasNoPrivateMaterial(export_result)
|
|
|
|
def test_service_public_list_and_detail_scrub_nested_solution_material(self) -> None:
|
|
service = TrainingCaseService(repository=_NestedPrivateRepository())
|
|
|
|
list_result = service.list_cases(limit=100)
|
|
detail_result = service.get_case(151)
|
|
|
|
self.assertEqual(list_result[0]["meta"]["tasks"][0]["meta"]["public_hint"], "Focus on context.")
|
|
self.assertEqual(detail_result["meta"]["metadata_note"]["public_label"], "visible")
|
|
self.assertIn("quality", list_result[0])
|
|
self.assertIn("quality", detail_result)
|
|
self.assertPublicPayloadHasNoPrivateMaterial({"items": list_result})
|
|
self.assertPublicPayloadHasNoPrivateMaterial(detail_result)
|
|
|
|
def test_cli_smoke_public_export_scrubs_nested_solution_material(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
package_path = Path(tmpdir) / "training_case_package.json"
|
|
output_path = Path(tmpdir) / "training_case_public_export.json"
|
|
package_path.write_text(
|
|
json.dumps(_case_package_with_nested_private_material(), sort_keys=True),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
summary = training_cli.run_fixture_smoke(package_path, output_path=output_path)
|
|
export_result = json.loads(output_path.read_text(encoding="utf-8"))
|
|
|
|
item = export_result["items"][0]
|
|
self.assertTrue(summary["passed"])
|
|
self.assertEqual(summary["imported"], 1)
|
|
self.assertEqual(summary["exported"], 1)
|
|
self.assertEqual(item["rubric"], [{"category": "context", "max_score": 3.0}])
|
|
self.assertEqual(item["tasks"][0]["meta"]["public_hint"], "Focus on context.")
|
|
self.assertEqual(item["metadata"]["chapter_ref"], "m15_public")
|
|
self.assertIn("quality", item)
|
|
self.assertPublicPayloadHasNoPrivateMaterial(export_result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|