fix(enterprise): run detect_label_issues single-process by default; test FPE with and without pyffx (#433)

detect_label_issues passed cleanlab's default n_jobs, which starts one worker per core. On spawn platforms (macOS, Windows) a call from a plain script without a __main__ guard re-imported the script in every worker and never returned. Default n_jobs to 1; an explicit n_jobs is passed through.

test_fpe_falls_back_to_surrogate_without_crypto asserted the surrogate mode without hiding pyffx, so it failed whenever the [privacy] or [all] extra was installed. Hide pyffx in that test and add a test for the crypto_fpe path when pyffx is present.

Author: kevincostner17

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

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]")

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():