From: Stefan Gasser Date: Sun, 26 Jul 2026 09:55:27 +0000 (+0200) Subject: Add GLiNER semantic backend abstraction (#156) X-Git-Tag: v0.9.0~2 X-Git-Url: http://git.99rst.org/?a=commitdiff_plain;h=1b9d3574754fdd08d632d1b3772518c1d429ad1c;p=sgasser-llm-shield.git Add GLiNER semantic backend abstraction (#156) * Add GLiNER semantic backend abstraction * Preserve detector environment compatibility * Improve model configuration errors --- diff --git a/detector/detector/__init__.py b/detector/detector/__init__.py index fb07dd4..b9ced70 100644 --- a/detector/detector/__init__.py +++ b/detector/detector/__init__.py @@ -1,4 +1,3 @@ -"""PasteGuard PII detector: an /analyze service over a deterministic -regex/checksum layer plus multilingual GLiNER NER.""" +"""PasteGuard PII detector: deterministic checks plus a semantic backend.""" __version__ = "0.1.0" diff --git a/detector/detector/app.py b/detector/detector/app.py index 3729ab4..1cd2ca9 100644 --- a/detector/detector/app.py +++ b/detector/detector/app.py @@ -7,8 +7,8 @@ from fastapi import FastAPI from pydantic import BaseModel from .deterministic import detect_deterministic -from .gliner_layer import detect_gliner, load_model from .merge import merge +from .semantic_backend import backend_info, detect_semantic, load_semantic_backend def _utf16_mapper(text: str): @@ -39,8 +39,9 @@ class Entity(BaseModel): @asynccontextmanager async def lifespan(_: FastAPI): - # Load the model before serving so /health == ready (PasteGuard polls it). - load_model() + # Load the selected backend before serving so /health == ready + # (PasteGuard polls it). + load_semantic_backend() yield @@ -49,14 +50,14 @@ app = FastAPI(title="PasteGuard Detector", version="0.1.0", lifespan=lifespan) @app.get("/health") def health() -> dict[str, str]: - return {"status": "ok"} + return {"status": "ok", **backend_info()} @app.post("/analyze", response_model=list[Entity]) def analyze(req: AnalyzeRequest) -> list[Entity]: deterministic = detect_deterministic(req.text, req.phone_regions) - fuzzy = detect_gliner(req.text, req.score_threshold) - spans = merge(deterministic, fuzzy, req.entities, 0.0) + semantic = detect_semantic(req.text, req.score_threshold) + spans = merge(deterministic, semantic, req.entities, 0.0) to_u16 = _utf16_mapper(req.text) return [ Entity( diff --git a/detector/detector/gliner_layer.py b/detector/detector/gliner_layer.py index be58f10..7c134d2 100644 --- a/detector/detector/gliner_layer.py +++ b/detector/detector/gliner_layer.py @@ -12,6 +12,7 @@ import os import re import threading from functools import lru_cache +from math import isfinite from typing import Any from .entities import LOCATION, PERSON, Span @@ -22,12 +23,49 @@ MODEL_INFERENCE_CACHE_SIZE = 4096 DEFAULT_MODEL = "urchade/gliner_multi_pii-v1" +def _env(name: str, legacy_name: str | None = None) -> tuple[str | None, str]: + value = os.environ.get(name) + if value is not None: + return value, name + if legacy_name is not None: + legacy_value = os.environ.get(legacy_name) + if legacy_value is not None: + return legacy_value, legacy_name + return None, name + + def _floor(label: str, default: float) -> float: - return float(os.environ.get(f"DETECTOR_FLOOR_{label.upper()}", default)) + name = f"GLINER_FLOOR_{label.upper()}" + value, source_name = _env(name, f"DETECTOR_FLOOR_{label.upper()}") + if value is None: + return default + try: + parsed = float(value) + except ValueError: + raise ValueError(f"{source_name} must be a number between 0 and 1; got {value!r}") from None + if not isfinite(parsed) or not 0.0 <= parsed <= 1.0: + raise ValueError(f"{source_name} must be a number between 0 and 1; got {value!r}") + return parsed + + +def _max_tokens(default: int = 384) -> int: + name = "GLINER_MAX_TOKENS" + value, source_name = _env(name, "DETECTOR_MAX_TOKENS") + if value is None: + return default + try: + parsed = int(value) + except ValueError: + raise ValueError( + f"{source_name} must be an integer of at least 64; got {value!r}" + ) from None + if parsed < 64: + raise ValueError(f"{source_name} must be an integer of at least 64; got {value!r}") + return parsed # Per-label confidence floors (calibrated against the accuracy benchmark; -# overridable via env, e.g. DETECTOR_FLOOR_LOCATION=0.6). Role nouns are demoted +# overridable via env, e.g. GLINER_FLOOR_LOCATION=0.6). Role nouns are demoted # by the suppressor labels (below), not by this floor. PER_LABEL_FLOOR = { "person": _floor("person", 0.95), @@ -59,7 +97,7 @@ _PREDICT_FLOOR = min(PER_LABEL_FLOOR.values()) - 0.1 # drop PII past the cut. Split into overlapping windows; the splitter mirrors # GLiNER's WhitespaceTokenSplitter so window sizes match. _TOKEN_RE = re.compile(r"\w+(?:[-_]\w+)*|\S") -_MAX_TOKENS = int(os.environ.get("DETECTOR_MAX_TOKENS", "384")) +_MAX_TOKENS = _max_tokens() _WINDOW = max(64, _MAX_TOKENS - 64) # headroom under the hard limit _OVERLAP = 64 # >= longest expected entity, so boundary-straddling spans survive @@ -84,28 +122,34 @@ def _windows(text: str): # GLiNER ships no type stubs, so the loaded model is untyped (Any). _model: Any = None +_loaded_model_name: str | None = None _lock = threading.Lock() # Torch inference is not guaranteed thread-safe; serialize concurrent /analyze calls. _infer_lock = threading.Lock() -def _model_name() -> str: - return ( - os.environ.get("DETECTOR_MODEL_PATH") or os.environ.get("DETECTOR_MODEL") or DEFAULT_MODEL - ) - - -def load_model() -> None: - """Load the model once. Safe to call at startup or lazily.""" - global _model +def load_model(model_name: str = DEFAULT_MODEL) -> None: + """Load the selected GLiNER model once.""" + global _loaded_model_name, _model if _model is not None: + if _loaded_model_name != model_name: + loaded = _loaded_model_name or "an unknown checkpoint" + raise RuntimeError( + f"GLiNER is already loaded with {loaded!r}; cannot load {model_name!r}" + ) return with _lock: if _model is not None: + if _loaded_model_name != model_name: + loaded = _loaded_model_name or "an unknown checkpoint" + raise RuntimeError( + f"GLiNER is already loaded with {loaded!r}; cannot load {model_name!r}" + ) return from gliner import GLiNER - _model = GLiNER.from_pretrained(_model_name()) + _model = GLiNER.from_pretrained(model_name) + _loaded_model_name = model_name @lru_cache(maxsize=MODEL_INFERENCE_CACHE_SIZE) @@ -116,7 +160,8 @@ def _predict_window(text: str): def detect_gliner(text: str, score_threshold: float = 0.0) -> list[Span]: if not text: return [] - load_model() + if _model is None: + raise RuntimeError("GLiNER model not loaded; load_semantic_backend() selects and loads it") n = len(text) # Run each window, shift spans back to absolute offsets, dedupe overlaps # (same span+label) keeping the max score. diff --git a/detector/detector/semantic_backend.py b/detector/detector/semantic_backend.py new file mode 100644 index 0000000..7590c23 --- /dev/null +++ b/detector/detector/semantic_backend.py @@ -0,0 +1,245 @@ +"""Semantic detector backend selection and model resolution.""" + +from __future__ import annotations + +import json +import os +import threading +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol + +from huggingface_hub import hf_hub_download +from huggingface_hub.errors import ( + HfHubHTTPError, + HFValidationError, + LocalEntryNotFoundError, + RemoteEntryNotFoundError, + RepositoryNotFoundError, +) +from huggingface_hub.utils import ( + validate_repo_id, # pyright: ignore[reportPrivateImportUsage] +) + +from . import gliner_layer +from .entities import Span + +DEFAULT_BACKEND = "gliner" +_GLINER_CONFIG = "gliner_config.json" +_GLINER_WEIGHTS = ("model.safetensors", "pytorch_model.bin") + + +class SemanticBackend(Protocol): + """A loaded semantic detector with provider-neutral identity metadata.""" + + @property + def name(self) -> str: ... + + @property + def model(self) -> str: ... + + def load(self) -> None: ... + + def detect(self, text: str, score_threshold: float = 0.0) -> list[Span]: ... + + +@dataclass(frozen=True) +class _FunctionBackend: + name: str + model: str + _load_model: Callable[[str], None] + _detect: Callable[[str, float], list[Span]] + + def load(self) -> None: + self._load_model(self.model) + + def detect(self, text: str, score_threshold: float = 0.0) -> list[Span]: + return self._detect(text, score_threshold) + + +@dataclass(frozen=True) +class _BackendDefinition: + default_model: str + resolve_model: Callable[[str, bool, str], str] + build: Callable[[str], SemanticBackend] + + +def _build_gliner(model: str) -> SemanticBackend: + return _FunctionBackend( + name="gliner", + model=model, + _load_model=gliner_layer.load_model, + _detect=gliner_layer.detect_gliner, + ) + + +_backend: SemanticBackend | None = None +_backend_lock = threading.Lock() + + +def _configured_model() -> tuple[str | None, str]: + """Return the configured model and the environment variable that supplied it.""" + legacy_path = os.environ.get("DETECTOR_MODEL_PATH") + if legacy_path: + return legacy_path, "DETECTOR_MODEL_PATH" + model = os.environ.get("DETECTOR_MODEL") + return model or None, "DETECTOR_MODEL" + + +def _validate_gliner_config(config_path: Path, model: str, source_name: str) -> None: + try: + config = json.loads(config_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise ValueError( + f"{source_name} {model!r} has an unreadable {_GLINER_CONFIG}: {exc}" + ) from exc + if not isinstance(config, dict): + raise ValueError( + f"{source_name} {model!r} has an invalid {_GLINER_CONFIG}: " + "the top-level value must be an object" + ) + + +def _validate_local_gliner_model(path: Path, configured_value: str, source_name: str) -> str: + if not path.is_dir(): + raise ValueError(f"{source_name} local path is not a directory: {configured_value}") + + config_path = path / _GLINER_CONFIG + if not config_path.is_file(): + raise ValueError( + f"{source_name} is not a complete GLiNER checkpoint: " + f"missing {_GLINER_CONFIG} in {configured_value}" + ) + _validate_gliner_config(config_path, configured_value, source_name) + + if not any((path / filename).is_file() for filename in _GLINER_WEIGHTS): + expected = " or ".join(_GLINER_WEIGHTS) + raise ValueError( + f"{source_name} is not a complete GLiNER checkpoint: " + f"missing {expected} in {configured_value}" + ) + + return str(path.resolve()) + + +def _looks_like_local_path(value: str) -> bool: + return value.startswith(("/", "./", "../", "~")) + + +def _validate_remote_gliner_model(model: str, source_name: str) -> None: + try: + config_path = Path(hf_hub_download(model, _GLINER_CONFIG)) + except RemoteEntryNotFoundError: + raise ValueError( + f"{source_name} {model!r} is not a GLiNER checkpoint: missing {_GLINER_CONFIG}" + ) from None + except RepositoryNotFoundError: + raise ValueError( + f"{source_name} {model!r} was not found or is not accessible; " + "check the Hugging Face model ID and authentication" + ) from None + except LocalEntryNotFoundError: + raise ValueError( + f"{source_name} {model!r} could not be validated because " + f"{_GLINER_CONFIG} is not cached and Hugging Face is unavailable" + ) from None + except HfHubHTTPError as exc: + raise ValueError( + f"{source_name} {model!r} could not be validated with Hugging Face: {exc}" + ) from exc + + _validate_gliner_config(config_path, model, source_name) + + +def _resolve_gliner_model( + configured_value: str, + is_default: bool = False, + source_name: str = "DETECTOR_MODEL", +) -> str: + value = configured_value.strip() + if not value: + raise ValueError(f"{source_name} must not be blank") + + try: + path = Path(value).expanduser() + except RuntimeError: + path = Path(value) + if path.exists(): + return _validate_local_gliner_model(path, configured_value, source_name) + if _looks_like_local_path(value): + raise ValueError(f"{source_name} local path does not exist: {configured_value}") + + try: + validate_repo_id(value) + except HFValidationError: + raise ValueError( + f"{source_name} {configured_value!r} is neither an existing local " + "directory nor a valid Hugging Face model ID" + ) from None + + # The built-in checkpoint identity is known. Custom repositories are + # checked before the much larger model download starts. + if not is_default: + _validate_remote_gliner_model(value, source_name) + return value + + +_BACKENDS = { + "gliner": _BackendDefinition( + default_model=gliner_layer.DEFAULT_MODEL, + resolve_model=_resolve_gliner_model, + build=_build_gliner, + ) +} + + +def load_semantic_backend() -> SemanticBackend: + """Resolve, validate, and load the configured semantic backend once.""" + global _backend + if _backend is not None: + return _backend + + with _backend_lock: + if _backend is not None: + return _backend + + backend_name = os.environ.get("DETECTOR_BACKEND", DEFAULT_BACKEND) or DEFAULT_BACKEND + definition = _BACKENDS.get(backend_name) + if definition is None: + supported = ", ".join(sorted(_BACKENDS)) + raise ValueError( + f"Unknown DETECTOR_BACKEND {backend_name!r}. Supported backends: {supported}" + ) + + configured_model, model_source = _configured_model() + selected_model = configured_model or definition.default_model + selected_model = definition.resolve_model( + selected_model, + configured_model is None or selected_model == definition.default_model, + model_source, + ) + + backend = definition.build(selected_model) + try: + backend.load() + except Exception as exc: + raise RuntimeError( + f"Failed to load semantic backend {backend.name!r} " + f"with model {backend.model!r}: {type(exc).__name__}: {exc}" + ) from exc + + _backend = backend + return backend + + +def backend_info() -> dict[str, str]: + """Return provider-neutral identity metadata for the loaded backend.""" + if _backend is None: + return {} + return {"backend": _backend.name, "model": _backend.model} + + +def detect_semantic(text: str, score_threshold: float = 0.0) -> list[Span]: + """Run the configured semantic backend and return canonical spans.""" + return load_semantic_backend().detect(text, score_threshold) diff --git a/detector/pyproject.toml b/detector/pyproject.toml index 2210da6..d9c69b4 100644 --- a/detector/pyproject.toml +++ b/detector/pyproject.toml @@ -5,14 +5,17 @@ build-backend = "setuptools.build_meta" [project] name = "pasteguard-detector" version = "0.1.0" -description = "/analyze PII detector: deterministic regex/checksum + multilingual GLiNER NER." +description = "/analyze PII detector with deterministic checks and a semantic backend." requires-python = ">=3.10" license = { text = "Apache-2.0" } dependencies = [ "fastapi>=0.110", "uvicorn>=0.29", "pydantic>=2", - "gliner>=0.2.13", + "gliner>=0.2.27,<0.3", + "huggingface-hub>=1,<2", + "torch>=2.6", + "transformers>=5.6.2,<5.7", "python-stdnum>=1.20", "phonenumbers>=8.13", ] diff --git a/detector/tests/test_analyze.py b/detector/tests/test_analyze.py index 698de32..35d99d8 100644 --- a/detector/tests/test_analyze.py +++ b/detector/tests/test_analyze.py @@ -13,9 +13,14 @@ from detector.entities import LOCATION, PERSON, PHONE_NUMBER, Span @pytest.fixture def client(monkeypatch): - monkeypatch.setattr(appmod, "load_model", lambda: None) + monkeypatch.setattr(appmod, "load_semantic_backend", lambda: None) + monkeypatch.setattr( + appmod, + "backend_info", + lambda: {"backend": "gliner", "model": "urchade/gliner_multi_pii-v1"}, + ) - def fake_gliner(text, score_threshold=0.0): + def fake_semantic(text, score_threshold=0.0): spans = [] for needle, etype in (("Mario Rossi", PERSON), ("München", LOCATION)): i = text.find(needle) @@ -25,7 +30,7 @@ def client(monkeypatch): spans.append(Span(etype, i, i + len(needle), 0.95)) return spans - monkeypatch.setattr(appmod, "detect_gliner", fake_gliner) + monkeypatch.setattr(appmod, "detect_semantic", fake_semantic) with TestClient(appmod.app) as c: yield c @@ -33,7 +38,24 @@ def client(monkeypatch): def test_health(client): r = client.get("/health") assert r.status_code == 200 - assert r.json() == {"status": "ok"} + assert r.json() == { + "status": "ok", + "backend": "gliner", + "model": "urchade/gliner_multi_pii-v1", + } + + +def test_unknown_backend_failure_aborts_startup(monkeypatch): + def fail_startup(): + raise ValueError("Unknown DETECTOR_BACKEND 'unknown'") + + monkeypatch.setattr(appmod, "load_semantic_backend", fail_startup) + + with ( + pytest.raises(ValueError, match="Unknown DETECTOR_BACKEND 'unknown'"), + TestClient(appmod.app), + ): + pass def test_response_shape_and_offsets(client): diff --git a/detector/tests/test_gliner_layer.py b/detector/tests/test_gliner_layer.py index 55a255a..62d2c11 100644 --- a/detector/tests/test_gliner_layer.py +++ b/detector/tests/test_gliner_layer.py @@ -8,14 +8,18 @@ from unittest.mock import Mock import pytest +import detector.gliner_layer as layer from detector.gliner_layer import ( _MAX_TOKENS, _SUPPRESS_LABELS, _TOKEN_RE, PER_LABEL_FLOOR, Span, + _floor, + _max_tokens, _windows, detect_gliner, + load_model, ) @@ -61,6 +65,120 @@ def test_per_label_floors_present_and_ordered(): assert PER_LABEL_FLOOR["location"] <= PER_LABEL_FLOOR["person"] +def test_gliner_loads_selected_model_once(monkeypatch): + from gliner import GLiNER + + loaded_model = object() + from_pretrained = Mock(return_value=loaded_model) + monkeypatch.setattr(layer, "_model", None) + monkeypatch.setattr(layer, "_loaded_model_name", None) + monkeypatch.setattr(GLiNER, "from_pretrained", from_pretrained) + + load_model("org/custom-gliner") + load_model("org/custom-gliner") + + from_pretrained.assert_called_once_with("org/custom-gliner") + assert layer._model is loaded_model + + +def test_gliner_rejects_different_model_after_loading(monkeypatch): + monkeypatch.setattr(layer, "_model", object()) + monkeypatch.setattr(layer, "_loaded_model_name", "org/already-loaded") + + with pytest.raises( + RuntimeError, + match=("GLiNER is already loaded with 'org/already-loaded'; cannot load 'org/different'"), + ): + load_model("org/different") + + +def test_detect_requires_loaded_model(monkeypatch): + monkeypatch.setattr(layer, "_model", None) + + with pytest.raises(RuntimeError, match="GLiNER model not loaded"): + detect_gliner("Alice", 0.0) + + +def test_floor_env_defaults(monkeypatch): + monkeypatch.delenv("GLINER_FLOOR_PERSON", raising=False) + monkeypatch.delenv("DETECTOR_FLOOR_PERSON", raising=False) + + assert _floor("person", 0.95) == 0.95 + + +def test_gliner_floor_env_wins_over_existing_legacy_name(monkeypatch): + monkeypatch.setenv("GLINER_FLOOR_PERSON", "0.91") + monkeypatch.setenv("DETECTOR_FLOOR_PERSON", "0.50") + + assert _floor("person", 0.95) == 0.91 + + +def test_existing_floor_env_remains_supported(monkeypatch): + monkeypatch.delenv("GLINER_FLOOR_PERSON", raising=False) + monkeypatch.setenv("DETECTOR_FLOOR_PERSON", "0.90") + + assert _floor("person", 0.95) == 0.90 + + +def test_invalid_existing_floor_env_names_the_source(monkeypatch): + monkeypatch.delenv("GLINER_FLOOR_PERSON", raising=False) + monkeypatch.setenv("DETECTOR_FLOOR_PERSON", "bad") + + with pytest.raises( + ValueError, + match="DETECTOR_FLOOR_PERSON must be a number between 0 and 1", + ): + _floor("person", 0.95) + + +@pytest.mark.parametrize("value", ["bad", "-0.1", "1.1", "nan", "inf"]) +def test_invalid_gliner_floor_fails_clearly(monkeypatch, value): + monkeypatch.setenv("GLINER_FLOOR_PERSON", value) + + with pytest.raises(ValueError, match="GLINER_FLOOR_PERSON must be a number between 0 and 1"): + _floor("person", 0.95) + + +def test_max_tokens_env_defaults(monkeypatch): + monkeypatch.delenv("GLINER_MAX_TOKENS", raising=False) + monkeypatch.delenv("DETECTOR_MAX_TOKENS", raising=False) + + assert _max_tokens() == 384 + + +def test_gliner_max_tokens_env_wins_over_existing_legacy_name(monkeypatch): + monkeypatch.setenv("GLINER_MAX_TOKENS", "512") + monkeypatch.setenv("DETECTOR_MAX_TOKENS", "256") + + assert _max_tokens() == 512 + + +def test_existing_max_tokens_env_remains_supported(monkeypatch): + monkeypatch.delenv("GLINER_MAX_TOKENS", raising=False) + monkeypatch.setenv("DETECTOR_MAX_TOKENS", "256") + + assert _max_tokens() == 256 + + +def test_invalid_existing_max_tokens_env_names_the_source(monkeypatch): + monkeypatch.delenv("GLINER_MAX_TOKENS", raising=False) + monkeypatch.setenv("DETECTOR_MAX_TOKENS", "many") + + with pytest.raises( + ValueError, + match="DETECTOR_MAX_TOKENS must be an integer of at least 64", + ): + _max_tokens() + + +@pytest.mark.parametrize("value", ["63", "384.5", "many"]) +def test_invalid_gliner_max_tokens_fails_clearly(monkeypatch, value): + monkeypatch.setenv("GLINER_MAX_TOKENS", value) + + with pytest.raises(ValueError, match="GLINER_MAX_TOKENS must be an integer of at least 64"): + _max_tokens() + + def test_window_inference_cache(monkeypatch): repeated_text = "This is going to repeat, don't waste resources more than once!" cache_missed_text = "This appears for the first time, it will increase the model call count." diff --git a/detector/tests/test_semantic_backend.py b/detector/tests/test_semantic_backend.py new file mode 100644 index 0000000..2832c0e --- /dev/null +++ b/detector/tests/test_semantic_backend.py @@ -0,0 +1,357 @@ +"""Tests for semantic backend selection and model validation (no downloads).""" + +from unittest.mock import Mock + +import pytest +from httpx import Request, Response +from huggingface_hub.errors import ( + LocalEntryNotFoundError, + RemoteEntryNotFoundError, + RepositoryNotFoundError, +) + +import detector.semantic_backend as semantic_backend +from detector.entities import PERSON, Span + + +@pytest.fixture(autouse=True) +def reset_semantic_backend(monkeypatch): + monkeypatch.delenv("DETECTOR_BACKEND", raising=False) + monkeypatch.delenv("DETECTOR_MODEL", raising=False) + monkeypatch.delenv("DETECTOR_MODEL_PATH", raising=False) + monkeypatch.setattr(semantic_backend, "_backend", None) + + +def _mock_gliner_loader(monkeypatch): + load = Mock() + monkeypatch.setattr(semantic_backend.gliner_layer, "load_model", load) + return load + + +def _valid_local_model(path): + path.mkdir(parents=True) + (path / "gliner_config.json").write_text("{}") + (path / "model.safetensors").touch() + return path + + +def _hub_error(error_type): + response = Response(404, request=Request("HEAD", "https://huggingface.co")) + return error_type("not found", response=response) + + +def test_gliner_is_the_only_default_backend(monkeypatch): + load = _mock_gliner_loader(monkeypatch) + download_config = Mock() + monkeypatch.setattr(semantic_backend, "hf_hub_download", download_config) + + backend = semantic_backend.load_semantic_backend() + + assert backend.name == "gliner" + assert backend.model == "urchade/gliner_multi_pii-v1" + load.assert_called_once_with("urchade/gliner_multi_pii-v1") + download_config.assert_not_called() + + +def test_explicit_gliner_selection(monkeypatch): + monkeypatch.setenv("DETECTOR_BACKEND", "gliner") + load = _mock_gliner_loader(monkeypatch) + + semantic_backend.load_semantic_backend() + + load.assert_called_once_with("urchade/gliner_multi_pii-v1") + + +@pytest.mark.parametrize("backend", ["unknown", "openai_privacy_filter"]) +def test_unknown_or_disabled_backend_fails_clearly(monkeypatch, backend): + monkeypatch.setenv("DETECTOR_BACKEND", backend) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises( + ValueError, + match=rf"Unknown DETECTOR_BACKEND {backend!r}. Supported backends: gliner", + ): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_selected_backend_loads_once(monkeypatch): + load = _mock_gliner_loader(monkeypatch) + + first = semantic_backend.load_semantic_backend() + second = semantic_backend.load_semantic_backend() + + assert first is second + load.assert_called_once() + + +def test_valid_local_model_is_resolved_and_loaded(monkeypatch, tmp_path): + model_path = _valid_local_model(tmp_path / "custom-checkpoint") + monkeypatch.setenv("DETECTOR_MODEL", str(model_path)) + load = _mock_gliner_loader(monkeypatch) + + backend = semantic_backend.load_semantic_backend() + + assert backend.model == str(model_path.resolve()) + load.assert_called_once_with(str(model_path.resolve())) + + +def test_existing_relative_model_directory_shadows_hub_id(monkeypatch, tmp_path): + model_path = _valid_local_model(tmp_path / "org" / "custom-gliner") + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("DETECTOR_MODEL", "org/custom-gliner") + load = _mock_gliner_loader(monkeypatch) + download_config = Mock() + monkeypatch.setattr(semantic_backend, "hf_hub_download", download_config) + + semantic_backend.load_semantic_backend() + + load.assert_called_once_with(str(model_path.resolve())) + download_config.assert_not_called() + + +def test_missing_local_model_path_fails_before_loading(monkeypatch, tmp_path): + missing = tmp_path / "missing-model" + monkeypatch.setenv("DETECTOR_MODEL", str(missing)) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises(ValueError, match="DETECTOR_MODEL local path does not exist"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_unresolvable_tilde_model_fails_actionably(monkeypatch): + model = "~unknown-pasteguard-user/custom-checkpoint" + monkeypatch.setenv("DETECTOR_MODEL", model) + monkeypatch.setattr( + semantic_backend.Path, + "expanduser", + Mock(side_effect=RuntimeError("Could not determine home directory")), + ) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises( + ValueError, + match=r"DETECTOR_MODEL local path does not exist: ~unknown-pasteguard-user/", + ): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_local_model_path_must_be_a_directory(monkeypatch, tmp_path): + model_file = tmp_path / "model.bin" + model_file.touch() + monkeypatch.setenv("DETECTOR_MODEL", str(model_file)) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises(ValueError, match="DETECTOR_MODEL local path is not a directory"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_local_model_requires_gliner_config(monkeypatch, tmp_path): + model_path = tmp_path / "incomplete-checkpoint" + model_path.mkdir() + (model_path / "model.safetensors").touch() + monkeypatch.setenv("DETECTOR_MODEL", str(model_path)) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises(ValueError, match=r"missing gliner_config\.json"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_local_model_requires_valid_gliner_config(monkeypatch, tmp_path): + model_path = tmp_path / "invalid-checkpoint" + model_path.mkdir() + (model_path / "gliner_config.json").write_text("not json") + (model_path / "pytorch_model.bin").touch() + monkeypatch.setenv("DETECTOR_MODEL", str(model_path)) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises(ValueError, match=r"unreadable gliner_config\.json"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_local_model_requires_weights(monkeypatch, tmp_path): + model_path = tmp_path / "incomplete-checkpoint" + model_path.mkdir() + (model_path / "gliner_config.json").write_text("{}") + monkeypatch.setenv("DETECTOR_MODEL", str(model_path)) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises(ValueError, match=r"missing model\.safetensors or pytorch_model\.bin"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_custom_hugging_face_model_is_validated_and_loaded(monkeypatch, tmp_path): + config = tmp_path / "gliner_config.json" + config.write_text("{}") + monkeypatch.setenv("DETECTOR_MODEL", "org/custom-gliner") + load = _mock_gliner_loader(monkeypatch) + download_config = Mock(return_value=str(config)) + monkeypatch.setattr(semantic_backend, "hf_hub_download", download_config) + + semantic_backend.load_semantic_backend() + + download_config.assert_called_once_with("org/custom-gliner", "gliner_config.json") + load.assert_called_once_with("org/custom-gliner") + + +def test_non_gliner_hugging_face_model_fails_before_loading(monkeypatch): + monkeypatch.setenv("DETECTOR_MODEL", "org/token-classifier") + load = _mock_gliner_loader(monkeypatch) + monkeypatch.setattr( + semantic_backend, + "hf_hub_download", + Mock(side_effect=_hub_error(RemoteEntryNotFoundError)), + ) + + with pytest.raises(ValueError, match="is not a GLiNER checkpoint"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_unknown_hugging_face_model_fails_actionably(monkeypatch): + monkeypatch.setenv("DETECTOR_MODEL", "org/missing-model") + load = _mock_gliner_loader(monkeypatch) + monkeypatch.setattr( + semantic_backend, + "hf_hub_download", + Mock(side_effect=_hub_error(RepositoryNotFoundError)), + ) + + with pytest.raises(ValueError, match="was not found or is not accessible"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_offline_uncached_model_fails_actionably(monkeypatch): + monkeypatch.setenv("DETECTOR_MODEL", "org/custom-gliner") + load = _mock_gliner_loader(monkeypatch) + monkeypatch.setattr( + semantic_backend, + "hf_hub_download", + Mock(side_effect=LocalEntryNotFoundError("not cached")), + ) + + with pytest.raises(ValueError, match="is not cached and Hugging Face is unavailable"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +@pytest.mark.parametrize("value", ["models/nested/path", "models/", "not a model!"]) +def test_invalid_model_value_fails_before_loading(monkeypatch, value): + monkeypatch.setenv("DETECTOR_MODEL", value) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises(ValueError, match="neither an existing local directory nor a valid"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_empty_model_env_uses_default(monkeypatch): + monkeypatch.setenv("DETECTOR_MODEL", "") + load = _mock_gliner_loader(monkeypatch) + + semantic_backend.load_semantic_backend() + + load.assert_called_once_with("urchade/gliner_multi_pii-v1") + + +def test_blank_model_env_fails_clearly(monkeypatch): + monkeypatch.setenv("DETECTOR_MODEL", " ") + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises(ValueError, match="DETECTOR_MODEL must not be blank"): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_existing_model_path_alias_remains_supported(monkeypatch, tmp_path): + model_path = _valid_local_model(tmp_path / "legacy-checkpoint") + monkeypatch.setenv("DETECTOR_MODEL_PATH", str(model_path)) + load = _mock_gliner_loader(monkeypatch) + + semantic_backend.load_semantic_backend() + + load.assert_called_once_with(str(model_path.resolve())) + + +def test_existing_model_path_alias_wins_over_baked_detector_model(monkeypatch, tmp_path): + model_path = _valid_local_model(tmp_path / "mounted-checkpoint") + monkeypatch.setenv("DETECTOR_MODEL", "urchade/gliner_multi_pii-v1") + monkeypatch.setenv("DETECTOR_MODEL_PATH", str(model_path)) + load = _mock_gliner_loader(monkeypatch) + + semantic_backend.load_semantic_backend() + + load.assert_called_once_with(str(model_path.resolve())) + + +def test_missing_model_path_alias_error_names_the_source(monkeypatch, tmp_path): + missing = tmp_path / "missing-checkpoint" + monkeypatch.setenv("DETECTOR_MODEL", "urchade/gliner_multi_pii-v1") + monkeypatch.setenv("DETECTOR_MODEL_PATH", str(missing)) + load = _mock_gliner_loader(monkeypatch) + + with pytest.raises( + ValueError, + match="DETECTOR_MODEL_PATH local path does not exist", + ): + semantic_backend.load_semantic_backend() + + load.assert_not_called() + + +def test_backend_info_reports_loaded_identity(monkeypatch): + _mock_gliner_loader(monkeypatch) + + assert semantic_backend.backend_info() == {} + + semantic_backend.load_semantic_backend() + + assert semantic_backend.backend_info() == { + "backend": "gliner", + "model": "urchade/gliner_multi_pii-v1", + } + + +def test_detect_semantic_uses_loaded_backend(monkeypatch): + _mock_gliner_loader(monkeypatch) + expected = [Span(PERSON, 0, 5, 0.9)] + detect = Mock(return_value=expected) + monkeypatch.setattr(semantic_backend.gliner_layer, "detect_gliner", detect) + + assert semantic_backend.detect_semantic("Alice", 0.7) == expected + detect.assert_called_once_with("Alice", 0.7) + + +def test_gliner_load_failure_includes_backend_and_model(monkeypatch): + load = Mock(side_effect=OSError("weights are unreadable")) + monkeypatch.setattr(semantic_backend.gliner_layer, "load_model", load) + + with pytest.raises( + RuntimeError, + match=( + "Failed to load semantic backend 'gliner' with model " + "'urchade/gliner_multi_pii-v1': OSError: weights are unreadable" + ), + ): + semantic_backend.load_semantic_backend() + + assert semantic_backend.backend_info() == {}