diff --git a/src/freshdata/enterprise/config.py b/src/freshdata/enterprise/config.py index 433fcd7..4d9762c 100644 --- a/src/freshdata/enterprise/config.py +++ b/src/freshdata/enterprise/config.py @@ -294,7 +294,10 @@ class PIIDetectionConfig: The fallback detector (regex + context keywords) needs no extra dependencies. When ``use_ner`` is set and the optional ``freshdata-cleaner[privacy]`` extra (Presidio) is installed, an NER pass is - layered on top; otherwise it is skipped silently. + layered on top. If Presidio is not installed or its analyzer fails to start, + :func:`~freshdata.enterprise.detect_pii` emits one ``UserWarning``, runs the + regex/context detector alone and records ``ner_active=False`` plus the + ``ner_error`` in the report metadata. """ enabled: bool = True diff --git a/src/freshdata/enterprise/privacy.py b/src/freshdata/enterprise/privacy.py index b790fdd..c3f60ee 100644 --- a/src/freshdata/enterprise/privacy.py +++ b/src/freshdata/enterprise/privacy.py @@ -34,6 +34,7 @@ import json import os import re +import warnings from abc import ABC, abstractmethod from collections.abc import Callable from dataclasses import dataclass, field @@ -55,6 +56,16 @@ _PREVIEW_LEN = 24 +def _is_missing_scalar(value: Any) -> bool: + """True for ``None`` and any scalar missing marker (``NaN``, ``pd.NA``, ``NaT``). + + Nullable ``string``/``Int64``/``boolean`` columns hold ``pd.NA`` and datetime + columns hold ``NaT``; both must be passed through like ``None`` rather than + stringified to ``""``/``"NaT"`` and masked as if they were values. + """ + return value is None or (pd.api.types.is_scalar(value) and bool(pd.isna(value))) + + # ===================================================================== # Entity patterns, context keywords, and HIPAA/GDPR maps # ===================================================================== @@ -449,28 +460,60 @@ def _ner_entities( _PRESIDIO_ANALYZER: Any = None +#: ``"ExceptionType: message"`` from the first failed Presidio start-up. Cached +#: so a missing package or language model is not retried for every cell. +_PRESIDIO_ERROR: str | None = None -def _get_presidio_analyzer() -> Any: # pragma: no cover - requires optional Presidio - global _PRESIDIO_ANALYZER - if _PRESIDIO_ANALYZER is None: +def _get_presidio_analyzer() -> Any: + """Return the shared Presidio analyzer, or ``None`` when it cannot start. + + The first failure is recorded in :data:`_PRESIDIO_ERROR` and not retried. + """ + global _PRESIDIO_ANALYZER, _PRESIDIO_ERROR + if _PRESIDIO_ANALYZER is None and _PRESIDIO_ERROR is None: try: from presidio_analyzer import AnalyzerEngine _PRESIDIO_ANALYZER = AnalyzerEngine() - except Exception: - _PRESIDIO_ANALYZER = None + except Exception as exc: + _PRESIDIO_ERROR = f"{type(exc).__name__}: {exc}" return _PRESIDIO_ANALYZER +def _duplicated_labels(frame: pd.DataFrame) -> list[Any]: + """Column labels that occur more than once, in first-seen order.""" + return list(dict.fromkeys(frame.columns[frame.columns.duplicated()])) + + def detect_pii(df: Any, *, config: PIIDetectionConfig | None = None) -> PIIScanReport: """Scan the text columns of *df* for PII; return a :class:`PIIScanReport`. Read-only. Only object/string columns are scanned. Raw matched substrings are redacted in the report unless ``config.redact_samples=False``. + + Raises :class:`ValueError` when *df* has duplicate column labels, because a + duplicated label does not identify a single column to scan. """ cfg = config or PIIDetectionConfig() frame = to_pandas(df) + duplicated = _duplicated_labels(frame) + if duplicated: + raise ValueError( + f"detect_pii requires unique column labels; duplicated: {duplicated}" + ) + ner_active = False + ner_error: str | None = None + if cfg.use_ner: + ner_active = _get_presidio_analyzer() is not None + if not ner_active: + ner_error = _PRESIDIO_ERROR or "presidio analyzer unavailable" + warnings.warn( + f"detect_pii: use_ner=True but the Presidio NER pass is unavailable " + f"({ner_error}); only the regex/context detector ran", + UserWarning, + stacklevel=2, + ) entities: list[PIIEntity] = [] scanned: list[str] = [] for col in frame.columns: @@ -479,19 +522,26 @@ def detect_pii(df: Any, *, config: PIIDetectionConfig | None = None) -> PIIScanR continue scanned.append(str(col)) for row, value in series.items(): - if value is None or (isinstance(value, float) and pd.isna(value)): + if _is_missing_scalar(value): continue text = str(value) cell_entities = detect_in_text(text, column=str(col), config=cfg) - if cfg.use_ner: + if ner_active: cell_entities = _merge_ner(cell_entities, _ner_entities(text, str(col), cfg)) for e in cell_entities: e.metadata["row"] = int(row) if isinstance(row, (int, float)) else row entities.extend(cell_entities) + metadata: dict[str, Any] = { + "ner": ner_active, + "ner_requested": bool(cfg.use_ner), + "ner_active": ner_active, + } + if ner_error is not None: + metadata["ner_error"] = ner_error return PIIScanReport( entities=entities, columns_scanned=tuple(scanned), - metadata={"ner": bool(cfg.use_ner)}, + metadata=metadata, ) @@ -955,25 +1005,53 @@ def anonymize( "detection_config=PIIDetectionConfig() to say what to mask." ) frame = to_pandas(df).copy() + duplicated = _duplicated_labels(frame) + if duplicated: + # A duplicated label selects several columns at once, so neither the + # detection pass nor a rule aimed at it can address a single column. + if detection_config is not None and detection_config.enabled: + raise ValueError( + "anonymize requires unique column labels for PII detection; " + f"duplicated: {duplicated}" + ) + for rule in rules: + targeted = _resolve_columns(rule, duplicated) + if targeted: + raise ValueError( + f"anonymize requires unique column labels; rule {rule.name!r} " + f"targets duplicated: {targeted}" + ) events: list[MaskingEvent] = [] changed_cols: list[str] = [] cells_changed = 0 metadata: dict[str, Any] = {} + fpe_modes: dict[str, dict[str, int]] = {} for rule in rules: key = _resolve_key(rule) vault = _vault_for(rule) for column in _resolve_columns(rule, list(frame.columns)): if column not in frame.columns: continue - n, fpe_mode = _apply_rule_column( + n, mode_counts = _apply_rule_column( frame, column, rule, key, vault, events, audit_include_pii ) if n: cells_changed += n changed_cols.append(str(column)) - if fpe_mode: - metadata["fpe_mode"] = fpe_mode + if mode_counts: + per_column = fpe_modes.setdefault(str(column), {}) + for mode, count in mode_counts.items(): + per_column[mode] = per_column.get(mode, 0) + count + + # One mode overall keeps the plain mode string; a mix of modes (per cell, + # column or rule) is reported as "mixed" with per-column cell counts. + modes_used = {mode for per_column in fpe_modes.values() for mode in per_column} + if len(modes_used) == 1: + metadata["fpe_mode"] = next(iter(modes_used)) + elif modes_used: + metadata["fpe_mode"] = "mixed" + metadata["fpe_modes"] = fpe_modes entities_found = 0 if detection_config is not None and detection_config.enabled: @@ -1059,7 +1137,8 @@ def _apply_rule_column( vault: TokenVault, events: list[MaskingEvent], include_pii: bool, -) -> tuple[int, str | None]: +) -> tuple[int, dict[str, int]]: + """Mask one column in place; return ``(cells_changed, {fpe_mode: cell_count})``.""" if rule.strategy == "drop": n = int(frame[column].notna().sum()) _record_event( @@ -1068,31 +1147,36 @@ def _apply_rule_column( source="column", original="", masked="", include_pii=include_pii, ) frame.drop(columns=[column], inplace=True) - return n, None + return n, {} - reversible = rule.strategy in ("tokenize", "fpe") and rule.reversible format_preserving = rule.strategy in ("fpe", "surrogate") or rule.preserve_format if rule.strategy in ("tokenize", "fpe") and rule.reversible and not key: raise ValueError( f"masking rule {rule.name!r}: reversible {rule.strategy} requires key= or key_env=" ) - fpe_mode: str | None = None + mode_counts: dict[str, int] = {} series = frame[column] entity_type = _entity_for_rule(rule, str(column)) n_changed = 0 new_values: list[Any] = [] for row, value in series.items(): - if value is None or (isinstance(value, float) and pd.isna(value)): + if _is_missing_scalar(value): new_values.append(value) continue original = str(value) masked, mode = _mask_one(original, rule, key, vault) if mode: - fpe_mode = mode + mode_counts[mode] = mode_counts.get(mode, 0) + 1 new_values.append(masked) if masked != original: n_changed += 1 + # Only a vault token or real FPE can be reversed; the surrogate + # fallback of ``fpe`` is one-way, whatever the rule asked for. + reversible = rule.reversible and ( + rule.strategy == "tokenize" + or (rule.strategy == "fpe" and mode == "crypto_fpe") + ) _record_event( events, column=str(column), row=row, entity_type=entity_type, rule=rule, strategy=rule.strategy, reversible=reversible, @@ -1100,7 +1184,7 @@ def _apply_rule_column( original=original, masked=masked, include_pii=include_pii, ) frame[column] = pd.Series(new_values, index=series.index) - return n_changed, fpe_mode + return n_changed, mode_counts def _mask_one( @@ -1152,7 +1236,7 @@ def _anonymize_detected( touched = False new_values: list[Any] = [] for row, value in series.items(): - if value is None or (isinstance(value, float) and pd.isna(value)): + if _is_missing_scalar(value): new_values.append(value) continue text = str(value) @@ -1248,7 +1332,17 @@ def check_k_anonymity( raise ValueError("check_k_anonymity requires at least one quasi-identifier") n_rows = len(frame) - sizes = frame.groupby(list(quasi_identifiers), dropna=False).size() + # Categorical keys would make groupby emit every category combination, + # including empty ones. Group on the observed values instead; observed=True + # is avoided because it mishandles dropna=False on pandas 1.5. + keys = frame[list(quasi_identifiers)] + categorical = { + col: object for col, dtype in keys.dtypes.items() if isinstance(dtype, pd.CategoricalDtype) + } + if categorical: + keys = keys.astype(categorical) + sizes = keys.groupby(list(quasi_identifiers), dropna=False).size() + sizes = sizes[sizes > 0] n_classes = int(len(sizes)) smallest = int(sizes.min()) if n_classes else 0 violating_mask = sizes < k diff --git a/src/freshdata/enterprise/privacy_policy.py b/src/freshdata/enterprise/privacy_policy.py index 304ea51..20b38e1 100644 --- a/src/freshdata/enterprise/privacy_policy.py +++ b/src/freshdata/enterprise/privacy_policy.py @@ -48,6 +48,7 @@ MaskingEvent, PrivacyReport, TokenVault, + _is_missing_scalar, _luhn_ok, detect_in_text, detokenize_value, @@ -688,7 +689,7 @@ def apply_privacy_policy( used_vault = tok_vault reversible = rule is None or rule.reversible for value in series: - if value is None or (isinstance(value, float) and pd.isna(value)): + if _is_missing_scalar(value): new_values.append(value) continue original = str(value) diff --git a/tests/test_privacy_missing_and_labels.py b/tests/test_privacy_missing_and_labels.py new file mode 100644 index 0000000..c1dce6c --- /dev/null +++ b/tests/test_privacy_missing_and_labels.py @@ -0,0 +1,467 @@ +"""Privacy engine correctness: missing values, categorical k-anonymity, labels and metadata. + +* #243: ``pd.NA``/``NaT`` cells stay missing under every strategy and action. +* #244: categorical quasi-identifiers do not produce empty equivalence classes. +* #265: duplicate column labels raise a clear ``ValueError``. +* #281: fpe audit metadata follows the mode each cell actually used. +* #282: detect_pii reports whether the NER pass actually ran. +""" + +from __future__ import annotations + +import sys +import types + +import numpy as np +import pandas as pd +import pytest + +from freshdata.enterprise import ( + EnterpriseConfig, + KAnonymityConfig, + MaskingRule, + PIIDetectionConfig, + PrivacyPolicy, + PrivacyRule, + anonymize, + apply_privacy_policy, + check_k_anonymity, + clean_enterprise, + detect_pii, + load_compliance_pack, +) + +KEY = "unit-test-key" + + +@pytest.fixture +def no_pyffx(monkeypatch): + """Force the surrogate fallback of ``fpe`` whether or not pyffx is installed.""" + monkeypatch.setitem(sys.modules, "pyffx", None) + + +# -------------------------------------------------------------------------- +# #243: missing cells are passed through, not masked +# -------------------------------------------------------------------------- + +#: dtype label -> column with a missing cell at position 1 and two real values. +_NULLABLE_COLUMNS = { + "string": lambda: pd.array(["alice@example.com", None, "bob@example.com"], dtype="string"), + "Int64": lambda: pd.array([1234567890, None, 9876543210], dtype="Int64"), + "boolean": lambda: pd.array([True, None, False], dtype="boolean"), + "datetime64[ns]": lambda: pd.to_datetime(["2024-01-02", None, "2024-03-04"]), + "object_pd_NA": lambda: pd.Series( + ["alice@example.com", pd.NA, "bob@example.com"], dtype=object + ), +} + +_RULES = { + "hash": MaskingRule(name="r", columns=("c",), strategy="hash", salt="fixed-salt"), + "redact": MaskingRule(name="r", columns=("c",), strategy="redact"), + "partial": MaskingRule(name="r", columns=("c",), strategy="partial"), + "regex_scrub": MaskingRule( + name="r", columns=("c",), strategy="regex_scrub", scrub_patterns=(), regexes=(r".+",) + ), + "tokenize": MaskingRule(name="r", columns=("c",), strategy="tokenize", key=KEY), + "surrogate": MaskingRule(name="r", columns=("c",), strategy="surrogate"), + "fpe": MaskingRule(name="r", columns=("c",), strategy="fpe", key=KEY), +} + + +@pytest.mark.parametrize("dtype", list(_NULLABLE_COLUMNS)) +@pytest.mark.parametrize("strategy", list(_RULES)) +def test_rule_strategies_keep_missing_cells_missing(dtype, strategy, no_pyffx): + df = pd.DataFrame({"c": _NULLABLE_COLUMNS[dtype]()}) + out, report = anonymize(df, rules=(_RULES[strategy],)) + assert pd.isna(out["c"].iloc[1]) + assert not pd.isna(out["c"].iloc[0]) and not pd.isna(out["c"].iloc[2]) + assert report.cells_changed == 2 + assert sorted(e.row for e in report.events) == [0, 2] + + +@pytest.mark.parametrize("dtype", list(_NULLABLE_COLUMNS)) +def test_drop_counts_only_present_cells(dtype): + df = pd.DataFrame({"c": _NULLABLE_COLUMNS[dtype](), "keep": [1, 2, 3]}) + out, report = anonymize(df, rules=(MaskingRule(name="r", columns=("c",), strategy="drop"),)) + assert "c" not in out.columns + assert report.cells_changed == 2 + + +_POLICY_CASES = [ + (dtype, action) + for action in ("tokenize", "pseudonymize", "redact", "quarantine") + for dtype in _NULLABLE_COLUMNS + # quarantine writes a string placeholder with Series.where, which masked + # Int64/boolean arrays reject; that is separate from missing-value handling. + if not (action == "quarantine" and dtype in ("Int64", "boolean")) +] + + +@pytest.mark.parametrize(("dtype", "action"), _POLICY_CASES) +def test_policy_actions_keep_missing_cells_missing(dtype, action, no_pyffx): + df = pd.DataFrame({"c": _NULLABLE_COLUMNS[dtype]()}) + rule = PrivacyRule(id="r", action=action, columns=("c",), classification="s") + out, report = apply_privacy_policy(df, PrivacyPolicy(name="p", rules=(rule,), key=KEY)) + assert pd.isna(out["c"].iloc[1]) + assert not pd.isna(out["c"].iloc[0]) and not pd.isna(out["c"].iloc[2]) + assert report.cells_changed == 2 + + +@pytest.mark.parametrize("dtype", ["string", "object_pd_NA"]) +def test_detection_path_keeps_missing_cells_missing(dtype): + values = ["call 123-45-6789 today", None, "ssn 987-65-4321"] + series = ( + pd.array(values, dtype="string") + if dtype == "string" + else pd.Series([values[0], pd.NA, values[2]], dtype=object) + ) + df = pd.DataFrame({"notes": series}) + out, report = anonymize(df, detection_config=PIIDetectionConfig()) + assert pd.isna(out["notes"].iloc[1]) + assert "" in out["notes"].iloc[0] and "" in out["notes"].iloc[2] + assert report.entities_found == 2 + scan = detect_pii(df) + assert sorted(e.metadata["row"] for e in scan.entities) == [0, 2] + + +def test_missing_masked_output_matches_object_none_control(): + rule = MaskingRule(name="h", columns=("c",), strategy="hash", salt="fixed-salt") + nullable, _ = anonymize( + pd.DataFrame({"c": pd.array(["a@b.com", None], dtype="string")}), rules=(rule,) + ) + control, _ = anonymize(pd.DataFrame({"c": ["a@b.com", None]}), rules=(rule,)) + assert nullable["c"].iloc[0] == control["c"].iloc[0] + + +def test_issue_243_redact_repro(): + df = pd.DataFrame({"email": pd.array(["a@b.com", None], dtype="string")}) + out, report = anonymize( + df, rules=(MaskingRule(name="r", columns=("email",), strategy="redact"),) + ) + assert out["email"].iloc[0] == "***" + assert pd.isna(out["email"].iloc[1]) + assert report.cells_changed == 1 + + +def test_issue_243_tokenize_repro(): + df = pd.DataFrame({"email": pd.array(["a@b.com", None], dtype="string")}) + out, report = anonymize( + df, rules=(MaskingRule(name="t", columns=("email",), strategy="tokenize", key="k"),) + ) + assert out["email"].iloc[0].startswith("tok_") + assert pd.isna(out["email"].iloc[1]) + assert report.cells_changed == 1 + + +def test_issue_243_policy_repro(): + df = pd.DataFrame({"email": pd.array(["a@b.com", None], dtype="string")}) + policy = PrivacyPolicy(packs=(load_compliance_pack("hipaa"),), jurisdiction="US") + out, report = apply_privacy_policy(df, policy) + assert out["email"].iloc[0] == "" + assert pd.isna(out["email"].iloc[1]) + assert report.cells_changed == 1 + + +# -------------------------------------------------------------------------- +# #244: k-anonymity with categorical quasi-identifiers +# -------------------------------------------------------------------------- + + +def _zip_sex_frame() -> pd.DataFrame: + return pd.DataFrame({"zip": ["10001"] * 5 + ["20002"] * 5, "sex": ["F"] * 5 + ["M"] * 5}) + + +def test_issue_244_categorical_matches_object(): + df = _zip_sex_frame() + obj = check_k_anonymity(df, ["zip", "sex"], k=5) + cat = check_k_anonymity(df.astype("category"), ["zip", "sex"], k=5) + for report in (obj, cat): + assert report.ok + assert report.smallest_class_size == 5 + assert report.n_equivalence_classes == 2 + assert report.rows_violating_k == 0 + assert report.high_risk_groups == [] + assert cat.to_dict() == obj.to_dict() + + +def test_categorical_quasi_identifier_with_missing_value(): + df = pd.DataFrame({"zip": ["10001", "10001", np.nan, "20002"], "sex": ["F", "F", "M", "M"]}) + obj = check_k_anonymity(df, ["zip", "sex"], k=2) + cat = check_k_anonymity(df.astype("category"), ["zip", "sex"], k=2) + assert cat.n_equivalence_classes == obj.n_equivalence_classes == 3 + assert cat.smallest_class_size == obj.smallest_class_size == 1 + assert cat.rows_violating_k == obj.rows_violating_k == 2 + assert {"zip": None, "sex": "M"} in [g["group"] for g in cat.high_risk_groups] + assert all(g["size"] > 0 for g in cat.high_risk_groups) + + +def test_unused_categories_are_not_classes(): + df = pd.DataFrame( + { + "zip": pd.Categorical(["10001"] * 3, categories=["10001", "20002", "30003"]), + "sex": pd.Categorical(["F"] * 3, categories=["F", "M", "X"]), + } + ) + report = check_k_anonymity(df, ["zip", "sex"], k=3) + assert report.ok + assert report.n_equivalence_classes == 1 + assert report.smallest_class_size == 3 + + +def test_mixed_categorical_and_object_quasi_identifiers(): + df = _zip_sex_frame() + df["zip"] = df["zip"].astype("category") + report = check_k_anonymity(df, ["zip", "sex"], k=5) + assert report.ok + assert report.n_equivalence_classes == 2 + assert report.smallest_class_size == 5 + + +def test_clean_enterprise_k_anonymity_with_categorical_quasi_identifiers(): + df = _zip_sex_frame().astype("category") + df["value"] = range(len(df)) + config = EnterpriseConfig( + k_anonymity=KAnonymityConfig(enabled=True, quasi_identifiers=("zip", "sex"), k=5) + ) + result = clean_enterprise(df, enterprise=config) + report = result.k_anonymity_report + assert report is not None + assert report.ok + assert report.n_equivalence_classes == 2 + assert report.smallest_class_size == 5 + + +# -------------------------------------------------------------------------- +# #265 part 1: duplicate column labels in detect_pii / anonymize +# -------------------------------------------------------------------------- + + +def _duplicate_email_frame() -> pd.DataFrame: + return pd.DataFrame([["a@b.com", "c@d.com"]], columns=["email", "email"]) + + +def test_issue_265_detect_pii_rejects_duplicate_labels(): + with pytest.raises(ValueError, match=r"detect_pii requires unique column labels.*'email'"): + detect_pii(_duplicate_email_frame()) + + +def test_issue_265_anonymize_detection_rejects_duplicate_labels(): + df = _duplicate_email_frame() + with pytest.raises(ValueError, match=r"requires unique column labels.*'email'"): + anonymize(df, detection_config=PIIDetectionConfig()) + assert df.iloc[0].tolist() == ["a@b.com", "c@d.com"] + + +def test_anonymize_rule_targeting_duplicated_label_raises(): + df = _duplicate_email_frame() + rule = MaskingRule(name="r", columns=("email",), strategy="redact") + with pytest.raises(ValueError, match=r"rule 'r'.*'email'"): + anonymize(df, rules=(rule,)) + + +def test_anonymize_rule_on_unique_column_with_duplicates_elsewhere(): + df = pd.DataFrame([["x", "y", "a@b.com"]], columns=["dup", "dup", "email"]) + rule = MaskingRule(name="r", columns=("email",), strategy="redact") + out, report = anonymize(df, rules=(rule,)) + assert out.columns.tolist() == ["dup", "dup", "email"] + assert out.iloc[0].tolist() == ["x", "y", "***"] + assert report.cells_changed == 1 + + +def test_anonymize_disabled_detection_ignores_duplicates_elsewhere(): + df = pd.DataFrame([["x", "y", "a@b.com"]], columns=["dup", "dup", "email"]) + rule = MaskingRule(name="r", columns=("email",), strategy="redact") + out, _ = anonymize(df, rules=(rule,), detection_config=PIIDetectionConfig(enabled=False)) + assert out.iloc[0].tolist() == ["x", "y", "***"] + + +# -------------------------------------------------------------------------- +# #281 parts 1-2: fpe audit metadata follows the mode actually used +# -------------------------------------------------------------------------- + +_SURROGATE_MODE = "surrogate_format_preserving_not_crypto_fpe" + + +class _StubInteger: + """Stand-in for ``pyffx.Integer`` that cannot encrypt the number 999.""" + + def __init__(self, key, length): + self.length = length + + def encrypt(self, n): + if n == 999: + raise ValueError("stub cannot encrypt 999") + return (n * 7 + 3) % (10**self.length) + + +@pytest.fixture +def stub_pyffx(monkeypatch): + module = types.ModuleType("pyffx") + module.Integer = _StubInteger + monkeypatch.setitem(sys.modules, "pyffx", module) + + +def test_issue_281_surrogate_fallback_is_not_reported_reversible(no_pyffx): + df = pd.DataFrame({"ssn": ["123-45-6789"]}) + rule = MaskingRule(name="r", columns=("ssn",), strategy="fpe", reversible=True, key="k") + out, report = anonymize(df, rules=(rule,)) + assert out["ssn"].iloc[0] == "119-85-2634" + assert [e.reversible for e in report.events] == [False] + assert report.metadata == {"fpe_mode": _SURROGATE_MODE} + + +def test_issue_281_mixed_modes_in_one_column_are_counted(stub_pyffx): + df = pd.DataFrame({"x": ["abc-def", "123-45-6789", "999"]}) + rule = MaskingRule(name="r", columns=("x",), strategy="fpe", reversible=True, key="k") + out, report = anonymize(df, rules=(rule,)) + assert out["x"].iloc[0] == "xhs-rps" + assert out["x"].iloc[1] == "864-19-7526" + assert report.metadata["fpe_mode"] == "mixed" + assert report.metadata["fpe_modes"] == {"x": {_SURROGATE_MODE: 2, "crypto_fpe": 1}} + assert {e.row: e.reversible for e in report.events} == {0: False, 1: True, 2: False} + assert "fpe_mode: mixed" in report.summary() + + +def test_mixed_modes_across_columns_are_counted_per_column(stub_pyffx): + df = pd.DataFrame({"a": ["123-45-6789"], "b": ["123-45-6789"]}) + rules = ( + MaskingRule(name="fa", columns=("a",), strategy="fpe", key="k"), + MaskingRule(name="sb", columns=("b",), strategy="surrogate"), + ) + _, report = anonymize(df, rules=rules) + assert report.metadata == { + "fpe_mode": "mixed", + "fpe_modes": {"a": {"crypto_fpe": 1}, "b": {_SURROGATE_MODE: 1}}, + } + + +def test_crypto_fpe_single_mode_reports_mode_string(stub_pyffx): + df = pd.DataFrame({"x": ["123-45-6789", "555-12-3456"]}) + rule = MaskingRule(name="r", columns=("x",), strategy="fpe", key="k") + _, report = anonymize(df, rules=(rule,)) + assert report.metadata == {"fpe_mode": "crypto_fpe"} + assert [e.reversible for e in report.events] == [False, False] + + +def test_reversible_tokenize_events_stay_reversible(): + df = pd.DataFrame({"email": ["a@b.com", "c@d.com"]}) + rule = MaskingRule(name="t", columns=("email",), strategy="tokenize", reversible=True, key=KEY) + _, report = anonymize(df, rules=(rule,)) + assert [e.reversible for e in report.events] == [True, True] + assert "fpe_mode" not in report.metadata + + +def test_single_mode_report_is_unchanged(no_pyffx): + df = pd.DataFrame({"ssn": ["123-45-6789", "987-65-4321"], "acct": ["1234567890", None]}) + rules = ( + MaskingRule( + name="s", columns=("ssn",), strategy="surrogate", preserve_format=True, visible=4 + ), + MaskingRule(name="f", columns=("acct",), strategy="fpe", key="K", preserve_format=True), + MaskingRule(name="d", columns=("ssn",), strategy="drop"), + ) + _, report = anonymize(df, rules=rules) + assert report.metadata == {"fpe_mode": _SURROGATE_MODE} + assert report.summary().endswith(f"fpe_mode: {_SURROGATE_MODE}") + assert all(e.reversible is False for e in report.events) + + +def test_rules_without_fpe_add_no_fpe_metadata(): + df = pd.DataFrame({"a": ["x"], "b": ["y"]}) + rules = ( + MaskingRule(name="r", columns=("a",), strategy="redact"), + MaskingRule(name="d", columns=("b",), strategy="drop"), + ) + _, report = anonymize(df, rules=rules) + assert report.metadata == {} + + +# -------------------------------------------------------------------------- +# #282: NER metadata when the Presidio analyzer cannot start +# -------------------------------------------------------------------------- + + +@pytest.fixture +def presidio_globals(monkeypatch): + """Start each test with no cached analyzer or start-up error.""" + monkeypatch.setattr("freshdata.enterprise.privacy._PRESIDIO_ANALYZER", None) + monkeypatch.setattr("freshdata.enterprise.privacy._PRESIDIO_ERROR", None) + + +def _install_presidio_stub(monkeypatch, engine_cls) -> None: + module = types.ModuleType("presidio_analyzer") + module.AnalyzerEngine = engine_cls + monkeypatch.setitem(sys.modules, "presidio_analyzer", module) + + +def _freshdata_user_warnings(record) -> list: + return [w for w in record if issubclass(w.category, UserWarning)] + + +def test_issue_282_failed_analyzer_start_is_reported(monkeypatch, presidio_globals): + constructed: list[int] = [] + + class FailingEngine: + def __init__(self): + constructed.append(1) + raise RuntimeError("spaCy model en_core_web_lg not installed") + + _install_presidio_stub(monkeypatch, FailingEngine) + df = pd.DataFrame({"t": ["Alice Johnson visited", "Bob wrote to a@b.com", "Carol"]}) + config = PIIDetectionConfig(use_ner=True) + + with pytest.warns(UserWarning, match="NER pass is unavailable") as record: + report = detect_pii(df, config=config) + assert len(_freshdata_user_warnings(record)) == 1 + assert report.metadata == { + "ner": False, + "ner_requested": True, + "ner_active": False, + "ner_error": "RuntimeError: spaCy model en_core_web_lg not installed", + } + assert report.entity_types == {"EMAIL"} + assert constructed == [1] + + with pytest.warns(UserWarning, match="spaCy model"): + again = detect_pii(df, config=config) + assert again.metadata["ner_error"] == report.metadata["ner_error"] + assert constructed == [1] + + +def test_working_analyzer_marks_ner_active(monkeypatch, presidio_globals): + analyzed: list[str] = [] + + class WorkingEngine: + def analyze(self, text, language): + analyzed.append(text) + return [] + + _install_presidio_stub(monkeypatch, WorkingEngine) + df = pd.DataFrame({"t": ["Alice Johnson visited", None]}) + report = detect_pii(df, config=PIIDetectionConfig(use_ner=True)) + assert report.metadata == {"ner": True, "ner_requested": True, "ner_active": True} + assert analyzed == ["Alice Johnson visited"] + + +def test_missing_presidio_package_records_import_error(monkeypatch, presidio_globals): + monkeypatch.setitem(sys.modules, "presidio_analyzer", None) + with pytest.warns(UserWarning, match="presidio_analyzer") as record: + report = detect_pii(pd.DataFrame({"t": ["x"]}), config=PIIDetectionConfig(use_ner=True)) + assert len(_freshdata_user_warnings(record)) == 1 + assert report.metadata["ner"] is False + assert report.metadata["ner_active"] is False + assert report.metadata["ner_error"].split(":")[0] in ("ImportError", "ModuleNotFoundError") + + +def test_ner_not_requested_never_starts_the_analyzer(monkeypatch, presidio_globals): + constructed: list[int] = [] + + class FailingEngine: + def __init__(self): + constructed.append(1) + raise RuntimeError("should not start") + + _install_presidio_stub(monkeypatch, FailingEngine) + report = detect_pii(pd.DataFrame({"t": ["a@b.com"]})) + assert report.metadata == {"ner": False, "ner_requested": False, "ner_active": False} + assert constructed == []