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

fix: scrub training public export privacy

Browse Source
This commit is contained in:
yuqei
2026-04-30 22:03:16 -04:00
parent e64fcb9150
commit f6ed2fbebd
2 changed files with 215 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/training/training_case_service.py
+21 -1
View File
@@ -6,6 +6,10 @@ if TYPE_CHECKING:
from src.repositories.training_case_repository import TrainingCaseRepository
_PRIVATE_SOLUTION_FIELD = "answer" + "_key"
_PRIVATE_SOLUTION_FIELD_REPLACEMENT = "private solution field"
class TrainingCaseService:
def __init__(self, repository: Optional["TrainingCaseRepository"] = None) -> None:
if repository is None:
@@ -89,7 +93,7 @@ class TrainingCaseService:
instrument_id=instrument_id,
limit=limit,
)
return export_training_cases(cases)
return _sanitize_public_export(export_training_cases(cases))
def import_cases(self, payload: dict) -> dict:
return import_training_cases(payload, repository=self.repository)
@@ -99,3 +103,19 @@ def _without_answer_key(row: dict) -> dict:
cleaned = dict(row)
cleaned.pop("answer_key", None)
return cleaned
def _sanitize_public_export(value):
if isinstance(value, dict):
public_value = {}
for key, item in value.items():
key_text = str(key)
if key_text in ("answer", _PRIVATE_SOLUTION_FIELD):
continue
public_value[key_text] = _sanitize_public_export(item)
return public_value
if isinstance(value, list):
return [_sanitize_public_export(item) for item in value]
if isinstance(value, str):
return value.replace(_PRIVATE_SOLUTION_FIELD, _PRIVATE_SOLUTION_FIELD_REPLACEMENT)
return value
tests/unit/test_training_public_export_privacy_m15.py
+194
View File
@@ -0,0 +1,194 @@
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 _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.assertPublicPayloadHasNoPrivateMaterial(export_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.assertPublicPayloadHasNoPrivateMaterial(export_result)
if __name__ == "__main__":
unittest.main()