-"""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"
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):
@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
@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(
import re
import threading
from functools import lru_cache
+from math import isfinite
from typing import Any
from .entities import LOCATION, PERSON, Span
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),
# 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
# 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)
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.
--- /dev/null
+"""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)
[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",
]
@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)
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
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):
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,
)
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."
--- /dev/null
+"""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() == {}
git.99rst.org / sgasser-llm-shield.git / commitdiff