upstream-archive/presidio
1
0
Fork 0
mirror of https://github.com/data-privacy-stack/presidio.git synced 2026-07-23 11:20:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

Revert "test slim nlp"

Browse Source 
This reverts commit 0d00f9d962.
This commit is contained in:
Sharon Hart
2026-03-22 14:12:21 +02:00
parent 0d00f9d962
commit 9a6cfb42ac
13 changed files with 23 additions and 727 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
presidio-analyzer/Dockerfile
View File

@@ -1,7 +1,7 @@
FROM python:3.12-slim@sha256:f3fa41d74a768c2fce8016b98c191ae8c1bacd8f1152870a3f9f87d350920b7c
ARG NLP_CONF_FILE=presidio_analyzer/conf/slim_nlp.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/slim.yaml
ARG NLP_CONF_FILE=presidio_analyzer/conf/default.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/default_analyzer.yaml
ARG RECOGNIZER_REGISTRY_CONF_FILE=presidio_analyzer/conf/default_recognizers.yaml
ENV PIP_NO_CACHE_DIR=1
ENV POETRY_VIRTUALENVS_CREATE=false

4
presidio-analyzer/Dockerfile.dev
View File

@@ -2,8 +2,8 @@ FROM python:3.12-slim@sha256:f3fa41d74a768c2fce8016b98c191ae8c1bacd8f1152870a3f9
ARG DEV_MODE=dev
ARG POETRY_EXTRAS=""
ARG NLP_CONF_FILE=presidio_analyzer/conf/slim_nlp.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/slim.yaml
ARG NLP_CONF_FILE=presidio_analyzer/conf/default.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/default_analyzer.yaml
ARG RECOGNIZER_REGISTRY_CONF_FILE=presidio_analyzer/conf/default_recognizers.yaml
ENV DEV_MODE=${DEV_MODE}

4
presidio-analyzer/Dockerfile.stanza
View File

@@ -1,7 +1,7 @@
FROM python:3.12-slim@sha256:f3fa41d74a768c2fce8016b98c191ae8c1bacd8f1152870a3f9f87d350920b7c
ARG NLP_CONF_FILE=presidio_analyzer/conf/slim_nlp.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/slim.yaml
ARG NLP_CONF_FILE=presidio_analyzer/conf/default.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/default_analyzer.yaml
ARG RECOGNIZER_REGISTRY_CONF_FILE=presidio_analyzer/conf/default_recognizers.yaml
ENV PIP_NO_CACHE_DIR=1
ENV POETRY_VIRTUALENVS_CREATE=false

4
presidio-analyzer/Dockerfile.windows
View File

@@ -1,7 +1,7 @@
FROM python:3.12-windowsservercore@sha256:f2ab91b547b695c40a69cb1931cf382cb4019f56e1df5a83c86372d0cc8fb5a4
ARG NLP_CONF_FILE=presidio_analyzer/conf/slim_nlp.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/slim.yaml
ARG NLP_CONF_FILE=presidio_analyzer/conf/default.yaml
ARG ANALYZER_CONF_FILE=presidio_analyzer/conf/default_analyzer.yaml
ARG RECOGNIZER_REGISTRY_CONF_FILE=presidio_analyzer/conf/default_recognizers.yaml
ENV PIP_NO_CACHE_DIR=1
ENV POETRY_VIRTUALENVS_CREATE=false

6
presidio-analyzer/install_nlp_models.py
View File

@@ -52,7 +52,7 @@ def install_models(conf_file: str) -> None:
def _download_model(engine_name: str, model_name: Union[str, Dict[str, str]]) -> None:
if engine_name == "spacy" or engine_name == "slim":
if engine_name == "spacy":
spacy_download(model_name)
elif engine_name == "stanza":
if stanza:
@@ -102,8 +102,8 @@ if __name__ == "__main__":
parser.add_argument(
"--conf_file",
required=False,
default="presidio_analyzer/conf/slim_nlp.yaml",
help="Location of nlp configuration yaml file. Default: conf/slim_nlp.yaml",
default="presidio_analyzer/conf/default.yaml",
help="Location of nlp configuration yaml file. Default: conf/default.yaml",
)
args = parser.parse_args()

2
presidio-analyzer/presidio_analyzer/analyzer_engine_provider.py
View File

@@ -159,7 +159,7 @@ class AnalyzerEngineProvider:
@staticmethod
def _get_full_conf_path(
default_conf_file: Union[Path, str] = "slim.yaml",
default_conf_file: Union[Path, str] = "default_analyzer.yaml",
) -> Path:
"""Return a Path to the default conf file."""
return Path(Path(__file__).parent, "conf", default_conf_file)

120
presidio-analyzer/presidio_analyzer/conf/slim.yaml
View File

@@ -1,120 +0,0 @@
supported_languages:
- en
default_score_threshold: 0
nlp_configuration:
nlp_engine_name: slim
models:
- lang_code: en
model_name: en_core_web_sm
recognizer_registry:
# global_regex_flags: 26
recognizers:
# GLiNERRecognizer replaces SpacyRecognizer for NER-based entity detection.
# The slim NLP engine provides only tokenization and lemmatization;
# entity extraction is handled by self-contained recognizers like GLiNER.
- name: GLiNERRecognizer
type: predefined
enabled: true
- name: CreditCardRecognizer
supported_languages:
- language: en
context: [credit, card, visa, mastercard, cc, amex, discover, jcb, diners, maestro, instapayment]
type: predefined
- name: UsBankRecognizer
type: predefined
- name: UsLicenseRecognizer
type: predefined
- name: UsItinRecognizer
type: predefined
- name: UsPassportRecognizer
type: predefined
- name: UsSsnRecognizer
type: predefined
- name: NhsRecognizer
type: predefined
- name: UkNinoRecognizer
type: predefined
enabled: false
- name: UkPassportRecognizer
type: predefined
enabled: false
- name: SgFinRecognizer
type: predefined
enabled: false
- name: AuAbnRecognizer
type: predefined
enabled: false
- name: AuAcnRecognizer
type: predefined
enabled: false
- name: AuTfnRecognizer
type: predefined
enabled: false
- name: AuMedicareRecognizer
type: predefined
enabled: false
- name: InPanRecognizer
type: predefined
enabled: false
- name: InAadhaarRecognizer
supported_languages:
- en
type: predefined
enabled: false
- name: InVehicleRegistrationRecognizer
type: predefined
enabled: false
- name: InPassportRecognizer
type: predefined
enabled: false
- name: CryptoRecognizer
type: predefined
- name: DateRecognizer
type: predefined
- name: EmailRecognizer
type: predefined
- name: IbanRecognizer
type: predefined
- name: IpRecognizer
type: predefined
- name: MedicalLicenseRecognizer
type: predefined
- name: PhoneRecognizer
type: predefined
- name: UrlRecognizer
type: predefined
- name: InVoterRecognizer
type: predefined
enabled: false
- name: InGstinRecognizer
type: predefined
enabled: false

2
presidio-analyzer/presidio_analyzer/nlp_engine/__init__.py
View File

@@ -4,7 +4,6 @@ from .device_detector import device_detector
from .ner_model_configuration import NerModelConfiguration
from .nlp_artifacts import NlpArtifacts
from .nlp_engine import NlpEngine
from .slim_spacy_nlp_engine import SlimSpacyNlpEngine
from .spacy_nlp_engine import SpacyNlpEngine
from .stanza_nlp_engine import StanzaNlpEngine
from .transformers_nlp_engine import TransformersNlpEngine
@@ -16,7 +15,6 @@ __all__ = [
"NerModelConfiguration",
"NlpArtifacts",
"NlpEngine",
"SlimSpacyNlpEngine",
"SpacyNlpEngine",
"StanzaNlpEngine",
"NlpEngineProvider",

31
presidio-analyzer/presidio_analyzer/nlp_engine/nlp_engine_provider.py
View File

@@ -8,7 +8,6 @@ from presidio_analyzer.input_validation import ConfigurationValidator
from presidio_analyzer.nlp_engine import (
NerModelConfiguration,
NlpEngine,
SlimSpacyNlpEngine,
SpacyNlpEngine,
StanzaNlpEngine,
TransformersNlpEngine,
@@ -41,12 +40,7 @@ class NlpEngineProvider:
nlp_configuration: Optional[Dict] = None,
):
if nlp_engines is None:
nlp_engines = (
SpacyNlpEngine,
StanzaNlpEngine,
TransformersNlpEngine,
SlimSpacyNlpEngine,
)
nlp_engines = (SpacyNlpEngine, StanzaNlpEngine, TransformersNlpEngine)
self.nlp_engines = {
engine.engine_name: engine for engine in nlp_engines if engine.is_available
@@ -82,9 +76,10 @@ class NlpEngineProvider:
with open(conf_file) as file:
return yaml.safe_load(file)
@staticmethod
def _get_full_conf_path(
default_conf_file: Union[Path, str] = "slim_nlp.yaml",
default_conf_file: Union[Path, str] = "default.yaml"
) -> Path:
"""Return a Path to the default conf file."""
return Path(Path(__file__).parent, "../conf", default_conf_file)
@@ -102,21 +97,15 @@ class NlpEngineProvider:
nlp_engine_class = self.nlp_engines[nlp_engine_name]
nlp_models = self.nlp_configuration["models"]
if nlp_engine_name == SlimSpacyNlpEngine.engine_name:
engine = nlp_engine_class(models=nlp_models)
else:
ner_model_configuration = self.nlp_configuration.get(
"ner_model_configuration"
)
if ner_model_configuration:
ner_model_configuration = NerModelConfiguration.from_dict(
ner_model_configuration
)
engine = nlp_engine_class(
models=nlp_models, ner_model_configuration=ner_model_configuration
ner_model_configuration = self.nlp_configuration.get("ner_model_configuration")
if ner_model_configuration:
ner_model_configuration = NerModelConfiguration.from_dict(
ner_model_configuration
)
engine = nlp_engine_class(
models=nlp_models, ner_model_configuration=ner_model_configuration
)
engine.load()
logger.info(
f"Created NLP engine: {engine.engine_name}. "

236
presidio-analyzer/presidio_analyzer/nlp_engine/slim_spacy_nlp_engine.py
View File

@@ -1,236 +0,0 @@
import logging
from pathlib import Path
from typing import Any, Dict, Generator, List, Optional, Tuple, Union
import spacy
from spacy.tokens import Doc
from presidio_analyzer.nlp_engine import NlpArtifacts, NlpEngine, device_detector
logger = logging.getLogger("presidio-analyzer")
# Default small spaCy models per language for tokenization/lemmatization.
# These are lightweight and focused on linguistic features (no NER).
DEFAULT_SLIM_MODELS = {
"en": "en_core_web_sm",
"es": "es_core_news_sm",
"de": "de_core_news_sm",
"fr": "fr_core_news_sm",
"it": "it_core_news_sm",
"pt": "pt_core_news_sm",
"nl": "nl_core_news_sm",
"pl": "pl_core_news_sm",
"ro": "ro_core_news_sm",
"ja": "ja_core_news_sm",
"zh": "zh_core_web_sm",
"da": "da_core_news_sm",
"el": "el_core_news_sm",
"nb": "nb_core_news_sm",
"lt": "lt_core_news_sm",
"mk": "mk_core_news_sm",
"xx": "xx_ent_wiki_sm",
"he": "he_core_news_sm",
}
class SlimSpacyNlpEngine(NlpEngine):
"""A slim NLP engine that provides only tokenization and lemmatization.
This engine loads spaCy models with the NER pipeline component disabled,
reducing memory usage and load time. It is intended for use in Presidio v3
where entity extraction is handled by self-contained recognizers rather
than the shared NLP engine.
The slim engine:
- Provides tokenization, lemmatization, stopword and punctuation checks.
- Does NOT extract named entities (returns empty entity lists).
- Uses small spaCy models by default for fast loading and low memory.
- Supports auto-downloading spaCy models when they are missing.
:param models: List of model configurations per language.
Example: [{"lang_code": "en", "model_name": "en_core_web_sm"}]
If not provided, uses default small models for the given languages.
:param supported_languages: List of language codes to support.
Used only when models is not provided, to select default models.
:param auto_download: Whether to automatically download missing spaCy models.
"""
engine_name = "slim"
is_available = bool(spacy)
def __init__(
self,
models: Optional[List[Dict[str, str]]] = None,
supported_languages: Optional[List[str]] = None,
auto_download: bool = True,
):
if models:
self.models = models
elif supported_languages:
self.models = self._models_from_languages(supported_languages)
else:
self.models = [{"lang_code": "en", "model_name": "en_core_web_sm"}]
self.auto_download = auto_download
self.nlp = None
@staticmethod
def _models_from_languages(languages: List[str]) -> List[Dict[str, str]]:
"""Build model configs from language codes using default small models."""
models = []
for lang in languages:
model_name = DEFAULT_SLIM_MODELS.get(lang)
if not model_name:
raise ValueError(
f"No default slim model for language '{lang}'. "
f"Provide an explicit model via the 'models' parameter. "
f"Supported defaults: {sorted(DEFAULT_SLIM_MODELS.keys())}"
)
models.append({"lang_code": lang, "model_name": model_name})
return models
def _enable_gpu(self) -> None:
"""Enable GPU support for spaCy if available."""
device = device_detector.get_device()
if device != "cpu":
try:
spacy.require_gpu()
except Exception as e:
logger.warning(
f"Failed to enable GPU ({device}), falling back to CPU: {e}"
)
def load(self) -> None:
"""Load spaCy models with NER disabled."""
logger.debug(f"Loading slim SpaCy models: {self.models}")
self._enable_gpu()
self.nlp = {}
for model in self.models:
self._validate_model_params(model)
model_name = model["model_name"]
if self.auto_download:
self._download_spacy_model_if_needed(model_name)
# Disable NER and parser to keep processing slim
self.nlp[model["lang_code"]] = spacy.load(
model_name, disable=["ner", "parser"]
)
logger.info(f"Loaded slim NLP engine with languages: {list(self.nlp.keys())}")
@staticmethod
def _download_spacy_model_if_needed(model_name: str) -> None:
"""Download a spaCy model if it is not already installed."""
if not (spacy.util.is_package(model_name) or Path(model_name).exists()):
logger.warning(f"Model {model_name} is not installed. Downloading...")
spacy.cli.download(model_name)
logger.info(f"Finished downloading model {model_name}")
@staticmethod
def _validate_model_params(model: Dict) -> None:
"""Validate that required model parameters are present."""
if "lang_code" not in model:
raise ValueError("lang_code is missing from model configuration")
if "model_name" not in model:
raise ValueError("model_name is missing from model configuration")
if not isinstance(model["model_name"], str):
raise ValueError("model_name must be a string")
def is_loaded(self) -> bool:
"""Return True if the model is already loaded."""
return self.nlp is not None
def process_text(self, text: str, language: str) -> NlpArtifacts:
"""Execute the slim NLP pipeline on the given text.
Performs tokenization and lemmatization only. No named entities
are extracted.
"""
if not self.nlp:
raise ValueError("NLP engine is not loaded. Consider calling .load()")
if language not in self.nlp:
raise ValueError(
f"Language '{language}' is not supported by this NLP engine. "
f"Supported languages: {list(self.nlp.keys())}"
)
doc = self.nlp[language](text)
return self._doc_to_nlp_artifact(doc, language)
def process_batch(
self,
texts: Union[List[str], List[Tuple[str, object]]],
language: str,
batch_size: int = 1,
n_process: int = 1,
as_tuples: bool = False,
) -> Generator[
Union[Tuple[Any, NlpArtifacts, Any], Tuple[Any, NlpArtifacts]], Any, None
]:
"""Execute the slim NLP pipeline on a batch of texts."""
if not self.nlp:
raise ValueError("NLP engine is not loaded. Consider calling .load()")
if language not in self.nlp:
raise ValueError(
f"Language '{language}' is not supported by this NLP engine. "
f"Supported languages: {list(self.nlp.keys())}"
)
if as_tuples:
if not all(isinstance(item, tuple) and len(item) == 2 for item in texts):
raise ValueError(
"When 'as_tuples' is True, "
"'texts' must be a list of tuples (text, context)."
)
texts = ((str(text), context) for text, context in texts)
else:
texts = (str(text) for text in texts)
batch_output = self.nlp[language].pipe(
texts, as_tuples=as_tuples, batch_size=batch_size, n_process=n_process
)
for output in batch_output:
if as_tuples:
doc, context = output
yield doc.text, self._doc_to_nlp_artifact(doc, language), context
else:
doc = output
yield doc.text, self._doc_to_nlp_artifact(doc, language)
def is_stopword(self, word: str, language: str) -> bool:
"""Return true if the given word is a stop word."""
return self.nlp[language].vocab[word].is_stop
def is_punct(self, word: str, language: str) -> bool:
"""Return true if the given word is a punctuation word."""
return self.nlp[language].vocab[word].is_punct
def get_supported_entities(self) -> List[str]:
"""Return an empty list — the slim engine does not extract entities."""
return []
def get_supported_languages(self) -> List[str]:
"""Return the supported languages for this NLP engine."""
if not self.nlp:
raise ValueError("NLP engine is not loaded. Consider calling .load()")
return list(self.nlp.keys())
def _doc_to_nlp_artifact(self, doc: Doc, language: str) -> NlpArtifacts:
"""Convert a spaCy Doc to NlpArtifacts with no entities."""
lemmas = [token.lemma_ for token in doc]
tokens_indices = [token.idx for token in doc]
return NlpArtifacts(
entities=[],
tokens=doc,
tokens_indices=tokens_indices,
lemmas=lemmas,
nlp_engine=self,
language=language,
scores=[],
)

4
presidio-analyzer/tests/conftest.py
View File

@@ -53,10 +53,6 @@ def nlp_engines(request, nlp_engine_provider) -> Dict[str, NlpEngine]:
}
]
)
elif name == "slim":
available_engines[f"{name}_en"] = engine_cls(
models=[{"lang_code": "en", "model_name": "en_core_web_sm"}]
)
else:
raise ValueError("Unsupported engine for tests")

2
presidio-analyzer/tests/test_analyzer_engine_provider.py
View File

@@ -389,7 +389,7 @@ def test_analyzer_engine_provider_get_full_conf_path():
path = AnalyzerEngineProvider._get_full_conf_path()
assert isinstance(path, Path)
assert path.name == "slim.yaml"
assert path.name == "default_analyzer.yaml"
assert path.exists()

331
presidio-analyzer/tests/test_slim_spacy_nlp_engine.py
View File

@@ -1,331 +0,0 @@
from pathlib import Path
from typing import Iterator
from unittest.mock import MagicMock, patch
import pytest
from presidio_analyzer.nlp_engine import (
NlpEngineProvider,
SlimSpacyNlpEngine,
)
from presidio_analyzer.nlp_engine.slim_spacy_nlp_engine import DEFAULT_SLIM_MODELS
@pytest.fixture(scope="module")
def slim_nlp_engine():
"""Create and load a slim NLP engine for English."""
engine = SlimSpacyNlpEngine(
models=[{"lang_code": "en", "model_name": "en_core_web_sm"}]
)
engine.load()
return engine
class TestSlimSpacyNlpEngineInit:
"""Tests for SlimSpacyNlpEngine initialization."""
def test_when_init_with_default_then_english_model_configured(self):
engine = SlimSpacyNlpEngine()
assert len(engine.models) == 1
assert engine.models[0]["lang_code"] == "en"
assert engine.models[0]["model_name"] == "en_core_web_sm"
def test_when_init_with_explicit_models_then_models_used(self):
models = [{"lang_code": "en", "model_name": "en_core_web_sm"}]
engine = SlimSpacyNlpEngine(models=models)
assert engine.models == models
def test_when_init_with_supported_languages_then_default_models_selected(self):
engine = SlimSpacyNlpEngine(supported_languages=["en", "es"])
assert len(engine.models) == 2
assert engine.models[0] == {"lang_code": "en", "model_name": "en_core_web_sm"}
assert engine.models[1] == {
"lang_code": "es",
"model_name": "es_core_news_sm",
}
def test_when_init_with_unsupported_language_then_raises(self):
with pytest.raises(ValueError, match="No default slim model"):
SlimSpacyNlpEngine(supported_languages=["xx_unsupported"])
def test_when_models_and_languages_both_given_then_models_takes_precedence(self):
models = [{"lang_code": "en", "model_name": "en_core_web_lg"}]
engine = SlimSpacyNlpEngine(
models=models, supported_languages=["es"]
)
# models parameter takes precedence
assert engine.models == models
def test_when_engine_name_then_slim(self):
assert SlimSpacyNlpEngine.engine_name == "slim"
def test_when_is_available_then_true(self):
assert SlimSpacyNlpEngine.is_available is True
def test_when_auto_download_default_then_true(self):
engine = SlimSpacyNlpEngine()
assert engine.auto_download is True
def test_when_auto_download_false_then_false(self):
engine = SlimSpacyNlpEngine(auto_download=False)
assert engine.auto_download is False
class TestSlimSpacyNlpEngineLoad:
"""Tests for loading the slim engine."""
def test_when_load_then_nlp_populated(self, slim_nlp_engine):
assert slim_nlp_engine.nlp is not None
assert "en" in slim_nlp_engine.nlp
def test_when_load_then_ner_disabled(self, slim_nlp_engine):
nlp = slim_nlp_engine.nlp["en"]
pipe_names = nlp.pipe_names
assert "ner" not in pipe_names
def test_when_load_then_parser_disabled(self, slim_nlp_engine):
nlp = slim_nlp_engine.nlp["en"]
pipe_names = nlp.pipe_names
assert "parser" not in pipe_names
def test_when_is_loaded_before_load_then_false(self):
engine = SlimSpacyNlpEngine()
assert engine.is_loaded() is False
def test_when_is_loaded_after_load_then_true(self, slim_nlp_engine):
assert slim_nlp_engine.is_loaded() is True
def test_when_load_with_invalid_model_then_raises(self):
engine = SlimSpacyNlpEngine(
models=[{"lang_code": "en", "model_name": "nonexistent_model_xyz"}],
auto_download=False,
)
with pytest.raises(OSError):
engine.load()
def test_when_validate_model_params_missing_lang_code_then_raises(self):
with pytest.raises(ValueError, match="lang_code"):
SlimSpacyNlpEngine._validate_model_params(
{"model_name": "en_core_web_sm"}
)
def test_when_validate_model_params_missing_model_name_then_raises(self):
with pytest.raises(ValueError, match="model_name"):
SlimSpacyNlpEngine._validate_model_params({"lang_code": "en"})
def test_when_validate_model_params_non_string_model_then_raises(self):
with pytest.raises(ValueError, match="model_name must be a string"):
SlimSpacyNlpEngine._validate_model_params(
{"lang_code": "en", "model_name": 123}
)
class TestSlimSpacyNlpEngineProcessText:
"""Tests for process_text method."""
def test_when_process_text_then_tokens_returned(self, slim_nlp_engine):
artifacts = slim_nlp_engine.process_text("simple text", language="en")
assert len(artifacts.tokens) == 2
def test_when_process_text_then_no_entities(self, slim_nlp_engine):
text = "John Smith lives in New York and works at Microsoft"
artifacts = slim_nlp_engine.process_text(text, language="en")
assert artifacts.entities == []
assert artifacts.scores == []
def test_when_process_text_then_lemmas_returned(self, slim_nlp_engine):
artifacts = slim_nlp_engine.process_text("running quickly", language="en")
assert len(artifacts.lemmas) == 2
# "running" should be lemmatized
assert artifacts.lemmas[0] == "run"
def test_when_process_text_then_tokens_indices_returned(self, slim_nlp_engine):
artifacts = slim_nlp_engine.process_text("hello world", language="en")
assert len(artifacts.tokens_indices) == 2
assert artifacts.tokens_indices[0] == 0
assert artifacts.tokens_indices[1] == 6
def test_when_process_text_then_nlp_engine_set(self, slim_nlp_engine):
artifacts = slim_nlp_engine.process_text("test", language="en")
assert artifacts.nlp_engine is slim_nlp_engine
def test_when_not_loaded_then_process_text_raises(self):
engine = SlimSpacyNlpEngine()
with pytest.raises(ValueError, match="not loaded"):
engine.process_text("test", language="en")
def test_when_unsupported_language_then_process_text_raises(self, slim_nlp_engine):
with pytest.raises(ValueError, match="not supported"):
slim_nlp_engine.process_text("test", language="fr")
def test_when_process_empty_text_then_returns_artifacts(self, slim_nlp_engine):
artifacts = slim_nlp_engine.process_text("", language="en")
assert len(artifacts.tokens) == 0
assert artifacts.entities == []
class TestSlimSpacyNlpEngineProcessBatch:
"""Tests for process_batch method."""
def test_when_process_batch_then_returns_iterator(self, slim_nlp_engine):
results = slim_nlp_engine.process_batch(
["hello", "world"], language="en"
)
assert isinstance(results, Iterator)
def test_when_process_batch_strings_then_text_and_artifacts(self, slim_nlp_engine):
results = list(
slim_nlp_engine.process_batch(["simple text", "another"], language="en")
)
assert len(results) == 2
text, artifacts = results[0]
assert text == "simple text"
assert len(artifacts.tokens) == 2
assert artifacts.entities == []
def test_when_process_batch_as_tuples_then_context_preserved(
self, slim_nlp_engine
):
inputs = [("hello world", {"id": 1}), ("test text", {"id": 2})]
results = list(
slim_nlp_engine.process_batch(inputs, language="en", as_tuples=True)
)
assert len(results) == 2
text, artifacts, context = results[0]
assert text == "hello world"
assert context == {"id": 1}
assert artifacts.entities == []
def test_when_process_batch_not_loaded_then_raises(self):
engine = SlimSpacyNlpEngine()
with pytest.raises(ValueError, match="not loaded"):
list(engine.process_batch(["test"], language="en"))
def test_when_process_batch_invalid_tuples_then_raises(self, slim_nlp_engine):
with pytest.raises(ValueError, match="tuples"):
list(
slim_nlp_engine.process_batch(
["not a tuple"], language="en", as_tuples=True
)
)
class TestSlimSpacyNlpEngineLinguisticUtils:
"""Tests for stopword and punctuation detection."""
def test_when_stopword_then_returns_true(self, slim_nlp_engine):
assert slim_nlp_engine.is_stopword("the", language="en") is True
def test_when_not_stopword_then_returns_false(self, slim_nlp_engine):
assert slim_nlp_engine.is_stopword("microsoft", language="en") is False
def test_when_punct_then_returns_true(self, slim_nlp_engine):
assert slim_nlp_engine.is_punct(".", language="en") is True
def test_when_not_punct_then_returns_false(self, slim_nlp_engine):
assert slim_nlp_engine.is_punct("hello", language="en") is False
class TestSlimSpacyNlpEngineSupportedEntitiesAndLanguages:
"""Tests for supported entities and languages."""
def test_when_get_supported_entities_then_empty(self, slim_nlp_engine):
assert slim_nlp_engine.get_supported_entities() == []
def test_when_get_supported_languages_then_returns_loaded(self, slim_nlp_engine):
assert slim_nlp_engine.get_supported_languages() == ["en"]
def test_when_get_supported_languages_not_loaded_then_raises(self):
engine = SlimSpacyNlpEngine()
with pytest.raises(ValueError, match="not loaded"):
engine.get_supported_languages()
class TestSlimSpacyNlpEngineProvider:
"""Tests for creating slim engine via NlpEngineProvider."""
def test_when_create_slim_engine_via_provider_then_succeeds(self, mocker):
mocker.patch(
"presidio_analyzer.nlp_engine.slim_spacy_nlp_engine."
"SlimSpacyNlpEngine._download_spacy_model_if_needed",
return_value=None,
)
nlp_configuration = {
"nlp_engine_name": "slim",
"models": [{"lang_code": "en", "model_name": "en_core_web_sm"}],
}
provider = NlpEngineProvider(nlp_configuration=nlp_configuration)
engine = provider.create_engine()
assert isinstance(engine, SlimSpacyNlpEngine)
assert engine.nlp is not None
def test_when_create_slim_engine_from_yaml_then_succeeds(self, mocker):
mocker.patch(
"presidio_analyzer.nlp_engine.slim_spacy_nlp_engine."
"SlimSpacyNlpEngine._download_spacy_model_if_needed",
return_value=None,
)
nlp_configuration = {
"nlp_engine_name": "slim",
"models": [{"lang_code": "en", "model_name": "en_core_web_sm"}],
}
provider = NlpEngineProvider(nlp_configuration=nlp_configuration)
engine = provider.create_engine()
assert isinstance(engine, SlimSpacyNlpEngine)
def test_when_slim_engine_in_available_engines(self):
provider = NlpEngineProvider()
assert "slim" in provider.nlp_engines
class TestSlimSpacyNlpEngineGpu:
"""Tests for GPU handling."""
def test_when_gpu_available_then_spacy_gpu_configured(self):
with patch(
"presidio_analyzer.nlp_engine.slim_spacy_nlp_engine.device_detector"
) as mock_detector:
mock_detector.get_device.return_value = "cuda"
with patch(
"presidio_analyzer.nlp_engine.slim_spacy_nlp_engine.spacy"
) as mock_spacy:
mock_spacy.load.return_value = MagicMock()
mock_spacy.util.is_package.return_value = True
engine = SlimSpacyNlpEngine(
models=[{"lang_code": "en", "model_name": "en_core_web_sm"}]
)
engine.load()
mock_spacy.require_gpu.assert_called_once()
def test_when_cpu_device_then_gpu_not_configured(self):
with patch(
"presidio_analyzer.nlp_engine.slim_spacy_nlp_engine.device_detector"
) as mock_detector:
mock_detector.get_device.return_value = "cpu"
with patch(
"presidio_analyzer.nlp_engine.slim_spacy_nlp_engine.spacy"
) as mock_spacy:
mock_spacy.load.return_value = MagicMock()
mock_spacy.util.is_package.return_value = True
engine = SlimSpacyNlpEngine(
models=[{"lang_code": "en", "model_name": "en_core_web_sm"}]
)
engine.load()
mock_spacy.require_gpu.assert_not_called()
class TestDefaultSlimModels:
"""Tests for the default model mapping."""
def test_when_default_models_then_english_present(self):
assert "en" in DEFAULT_SLIM_MODELS
def test_when_default_models_then_all_values_are_sm(self):
for lang, model in DEFAULT_SLIM_MODELS.items():
assert "_sm" in model or "sm" in model, (
f"Default model for {lang} ({model}) should be a small model"
)