195 lines
7.8 KiB
Python
195 lines
7.8 KiB
Python
import json
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
try:
|
|
from jsonschema import Draft202012Validator
|
|
except ModuleNotFoundError: # pragma: no cover - optional test helper dependency
|
|
Draft202012Validator = None
|
|
|
|
from app.services import guardrail
|
|
from app.services.pii_masking_eval import evaluate_fixture, load_cases
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
FIXTURE_PATH = REPO_ROOT / "data" / "privacy" / "pii-masking-ko-fixtures.json"
|
|
INPUT_SCHEMA_PATH = REPO_ROOT / "data" / "privacy" / "pii-masking-eval-input.schema.json"
|
|
REPORT_SCHEMA_PATH = REPO_ROOT / "data" / "privacy" / "pii-masking-eval-report.schema.json"
|
|
SCRIPT_PATH = REPO_ROOT / "scripts" / "evaluate-pii-masking.py"
|
|
|
|
EXPECTED_CATEGORIES = {
|
|
"contact",
|
|
"name",
|
|
"national_id",
|
|
"negative_control",
|
|
"organization",
|
|
"quasi_identifier",
|
|
}
|
|
RAW_IDENTIFIERS = (
|
|
"김서연",
|
|
"박민수",
|
|
"최하늘",
|
|
"한신대학교",
|
|
"상담심리학과",
|
|
"마음봄상담센터",
|
|
"새봄병원",
|
|
"010-1234-5678",
|
|
"seoyeon@example.com",
|
|
"990101-1234567",
|
|
"123456789012",
|
|
"2001년 4월 18일",
|
|
"서울시 강남구 역삼동",
|
|
"1200원",
|
|
)
|
|
|
|
|
|
class PiiMaskingEvalTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.presidio_patch = patch.object(
|
|
guardrail,
|
|
"_try_load_presidio",
|
|
return_value=(None, None),
|
|
)
|
|
self.presidio_patch.start()
|
|
guardrail.set_ko_pii_recognizer(None)
|
|
self.addCleanup(self.presidio_patch.stop)
|
|
self.addCleanup(guardrail.set_ko_pii_recognizer, None)
|
|
|
|
def test_fixture_cases_are_valid_json_list(self) -> None:
|
|
cases = load_cases(FIXTURE_PATH)
|
|
|
|
self.assertEqual(len(cases), 15)
|
|
self.assertTrue(all(case.get("id") for case in cases))
|
|
self.assertTrue(all(case.get("text") for case in cases))
|
|
self.assertTrue(all(case.get("locale") == "ko-KR" for case in cases))
|
|
self.assertTrue(all(case.get("source") == "synthetic" for case in cases))
|
|
self.assertTrue(all("expected_entities" in case for case in cases))
|
|
self.assertTrue(all("forbidden_substrings" in case for case in cases))
|
|
self.assertEqual({case["category"] for case in cases}, EXPECTED_CATEGORIES)
|
|
self._validate_with_schema(cases, INPUT_SCHEMA_PATH)
|
|
|
|
def test_ko_name_org_fixture_passes_without_raw_identifier_leak(self) -> None:
|
|
report = evaluate_fixture(FIXTURE_PATH)
|
|
|
|
self.assertEqual(report["schema_version"], "vignette.pii_masking_eval_report.v1")
|
|
self.assertEqual(report["input_schema_version"], "vignette.pii_masking_eval_input.v1")
|
|
self.assertEqual(report["run_mode"], "technical_dry_run")
|
|
self.assertEqual(report["data_source"], "local_fixture")
|
|
self.assertFalse(report["evidence_text_included"])
|
|
self.assertTrue(report["passed"], report)
|
|
self.assertEqual(report["cases_total"], 15)
|
|
self.assertEqual(report["cases_passed"], 15)
|
|
self.assertEqual(report["cases_failed"], 0)
|
|
self.assertEqual(report["expected_entity_recall"], 1.0)
|
|
self.assertEqual(report["forbidden_substring_removal"], 1.0)
|
|
self.assertEqual(report["unexpected_entity_violations"], 0)
|
|
self.assertEqual(report["by_source"]["synthetic"]["cases_passed"], 15)
|
|
self.assertEqual(set(report["by_category"]), EXPECTED_CATEGORIES)
|
|
self.assertEqual(report["by_severity"]["critical"]["cases_passed"], 2)
|
|
for result in report["results"]:
|
|
self.assertNotIn("masked_text", result)
|
|
self.assertNotIn("forbidden_remaining", result)
|
|
self.assertIn("forbidden_remaining_count", result)
|
|
self._validate_with_schema(report, REPORT_SCHEMA_PATH)
|
|
blob = json.dumps(report, ensure_ascii=False)
|
|
for raw in RAW_IDENTIFIERS:
|
|
self.assertNotIn(raw, blob)
|
|
|
|
def test_natural_language_name_contexts_are_masked_without_label_false_positive(self) -> None:
|
|
cases = {
|
|
"제 이름은 김서연입니다.": "제 이름은 [NAME]입니다.",
|
|
"보호자 이름은 박민수입니다.": "보호자 이름은 [NAME]입니다.",
|
|
"저는 최하늘입니다.": "저는 [NAME]입니다.",
|
|
"김서연입니다.": "[NAME]입니다.",
|
|
"이름은 중요하지 않고 상담 내용만 이야기하고 싶어요.": "이름은 중요하지 않고 상담 내용만 이야기하고 싶어요.",
|
|
}
|
|
|
|
for raw, expected in cases.items():
|
|
with self.subTest(raw=raw):
|
|
result = guardrail.mask_pii(raw)
|
|
self.assertEqual(result.text_masked, expected)
|
|
|
|
def test_optional_ko_recognizer_masks_adapter_spans_before_regex_fallback(self) -> None:
|
|
class FakeKoRecognizer:
|
|
def analyze(self, text: str):
|
|
spans = []
|
|
for entity_type, value in (
|
|
("NAME", "보라별"),
|
|
("ORG", "미래학교상담연구랩"),
|
|
):
|
|
start = text.index(value)
|
|
spans.append(guardrail.PiiEntitySpan(entity_type, start, start + len(value)))
|
|
return spans
|
|
|
|
guardrail.set_ko_pii_recognizer(FakeKoRecognizer())
|
|
|
|
result = guardrail.mask_pii(
|
|
"별명은 보라별이고 기관은 미래학교상담연구랩입니다. 전화는 010-1234-5678입니다."
|
|
)
|
|
|
|
self.assertTrue(result.used_ko_recognizer)
|
|
self.assertFalse(result.used_presidio)
|
|
self.assertIn("NAME", result.entities)
|
|
self.assertIn("ORG", result.entities)
|
|
self.assertIn("PHONE", result.entities)
|
|
self.assertNotIn("보라별", result.text_masked)
|
|
self.assertNotIn("미래학교상담연구랩", result.text_masked)
|
|
self.assertNotIn("010-1234-5678", result.text_masked)
|
|
self.assertIn("[NAME]", result.text_masked)
|
|
self.assertIn("[ORG]", result.text_masked)
|
|
self.assertIn("[PHONE]", result.text_masked)
|
|
|
|
def test_optional_ko_recognizer_failure_keeps_regex_fallback(self) -> None:
|
|
class BrokenKoRecognizer:
|
|
def analyze(self, text: str):
|
|
raise RuntimeError("adapter down")
|
|
|
|
guardrail.set_ko_pii_recognizer(BrokenKoRecognizer())
|
|
|
|
result = guardrail.mask_pii("전화는 010-1234-5678입니다.")
|
|
|
|
self.assertFalse(result.used_ko_recognizer)
|
|
self.assertFalse(result.used_presidio)
|
|
self.assertIn("PHONE", result.entities)
|
|
self.assertIn("[PHONE]", result.text_masked)
|
|
|
|
def test_cli_reports_json_and_nonzero_gate_shape(self) -> None:
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-X",
|
|
"utf8",
|
|
str(SCRIPT_PATH),
|
|
"--fixtures",
|
|
str(FIXTURE_PATH),
|
|
"--json",
|
|
],
|
|
cwd=str(REPO_ROOT),
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
|
|
report = json.loads(completed.stdout)
|
|
self.assertTrue(report["passed"])
|
|
self.assertEqual(report["cases_total"], 15)
|
|
self.assertFalse(report["evidence_text_included"])
|
|
self.assertEqual(set(report["by_category"]), EXPECTED_CATEGORIES)
|
|
self.assertTrue(all("masked_text" not in result for result in report["results"]))
|
|
self._validate_with_schema(report, REPORT_SCHEMA_PATH)
|
|
|
|
def _validate_with_schema(self, instance: object, schema_path: Path) -> None:
|
|
schema = json.loads(schema_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(schema.get("$schema"), "https://json-schema.org/draft/2020-12/schema")
|
|
if Draft202012Validator is None:
|
|
return
|
|
Draft202012Validator.check_schema(schema)
|
|
Draft202012Validator(schema).validate(instance)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|