diff --git a/src/freshdata/enterprise/cleaner.py b/src/freshdata/enterprise/cleaner.py index 6a2eede..297be54 100644 --- a/src/freshdata/enterprise/cleaner.py +++ b/src/freshdata/enterprise/cleaner.py @@ -768,8 +768,14 @@ def detect_label_issues( out-of-sample predicted probabilities. Returns indices ranked by ``return_indices_ranked_by`` (a valid Cleanlab option). Raises a clear :class:`ImportError` if Cleanlab is not installed. + + ``n_jobs`` defaults to 1. Cleanlab's own default starts one worker per core, + and on platforms that spawn workers (macOS, Windows) a call from a plain + script without an ``if __name__ == "__main__"`` guard then never returns. + Pass ``n_jobs`` explicitly to use more workers. """ _require_cleanlab() + kwargs.setdefault("n_jobs", 1) from cleanlab.filter import find_label_issues # pragma: no cover - needs cleanlab return find_label_issues( # pragma: no cover - needs cleanlab diff --git a/tests/test_enterprise_cleaner.py b/tests/test_enterprise_cleaner.py index 50f5dfa..4a95d8e 100644 --- a/tests/test_enterprise_cleaner.py +++ b/tests/test_enterprise_cleaner.py @@ -1,5 +1,10 @@ """Clustering, PII masking, semantic validation, and cleanlab-guard tests.""" +import os +import subprocess +import sys +import types + import pandas as pd import pytest @@ -394,3 +399,58 @@ def test_cleanlab_wrappers_raise_clear_error_when_absent(): detect_label_issues([0, 1], [[0.9, 0.1], [0.2, 0.8]]) with pytest.raises(ImportError, match="cleanlab"): detect_outliers([[1, 2], [3, 4]]) + + +def _fake_cleanlab(monkeypatch) -> list: + calls: list = [] + filter_module = types.ModuleType("cleanlab.filter") + + def find_label_issues(**kwargs): + calls.append(kwargs) + return [2] + + filter_module.find_label_issues = find_label_issues + package = types.ModuleType("cleanlab") + package.filter = filter_module + monkeypatch.setitem(sys.modules, "cleanlab", package) + monkeypatch.setitem(sys.modules, "cleanlab.filter", filter_module) + return calls + + +def test_detect_label_issues_runs_single_process_by_default(monkeypatch): + calls = _fake_cleanlab(monkeypatch) + assert detect_label_issues([0, 1], [[0.9, 0.1], [0.2, 0.8]]) == [2] + assert calls[0]["n_jobs"] == 1 + assert calls[0]["return_indices_ranked_by"] == "self_confidence" + + +def test_detect_label_issues_keeps_an_explicit_n_jobs(monkeypatch): + calls = _fake_cleanlab(monkeypatch) + detect_label_issues([0, 1], [[0.9, 0.1], [0.2, 0.8]], n_jobs=4) + assert calls[0]["n_jobs"] == 4 + + +def test_detect_label_issues_returns_from_an_unguarded_script(tmp_path): + # With cleanlab's default worker pool, spawn start methods re-import this + # script in every worker and the call never returns. + pytest.importorskip("cleanlab") + script = tmp_path / "label_issues.py" + script.write_text( + "import numpy as np\n" + "from freshdata.enterprise.cleaner import detect_label_issues\n" + "labels = np.array([0, 1, 0, 1, 0, 1])\n" + "probs = np.array([[0.9, 0.1], [0.2, 0.8], [0.1, 0.9]," + " [0.3, 0.7], [0.8, 0.2], [0.05, 0.95]])\n" + "print([int(i) for i in detect_label_issues(labels, probs)])\n" + ) + env = {**os.environ, "PYTHONPATH": os.pathsep.join(sys.path)} + done = subprocess.run( + [sys.executable, str(script)], + capture_output=True, + text=True, + timeout=120, + env=env, + check=False, + ) + assert done.returncode == 0, done.stderr[-2000:] + assert done.stdout.strip().endswith("[2]") diff --git a/tests/test_enterprise_privacy.py b/tests/test_enterprise_privacy.py index 54ace97..c801418 100644 --- a/tests/test_enterprise_privacy.py +++ b/tests/test_enterprise_privacy.py @@ -2,6 +2,8 @@ from __future__ import annotations +import sys + import pandas as pd import pytest @@ -169,13 +171,26 @@ def test_surrogate_preserves_shape(): assert report.metadata["fpe_mode"] == "surrogate_format_preserving_not_crypto_fpe" -def test_fpe_falls_back_to_surrogate_without_crypto(): +def test_fpe_falls_back_to_surrogate_without_crypto(monkeypatch): + # Hide pyffx so the fallback runs even when the [privacy] extra is installed. + monkeypatch.setitem(sys.modules, "pyffx", None) + df = pd.DataFrame({"acct": ["1234567890"]}) + rule = MaskingRule(name="f", columns=("acct",), strategy="fpe", key="K", preserve_format=True) + out, report = anonymize(df, rules=(rule,)) + masked = out["acct"].iloc[0] + assert len(masked) == 10 and masked.isdigit() + assert report.metadata["fpe_mode"] == "surrogate_format_preserving_not_crypto_fpe" + + +def test_fpe_uses_crypto_when_pyffx_is_installed(): + pytest.importorskip("pyffx") df = pd.DataFrame({"acct": ["1234567890"]}) rule = MaskingRule(name="f", columns=("acct",), strategy="fpe", key="K", preserve_format=True) out, report = anonymize(df, rules=(rule,)) masked = out["acct"].iloc[0] + assert masked != "1234567890" assert len(masked) == 10 and masked.isdigit() - assert report.metadata["fpe_mode"].startswith("surrogate_format_preserving") + assert report.metadata["fpe_mode"] == "crypto_fpe" def test_detection_driven_anonymization_scrubs_spans():