Add unit tests (#17)

Author: steven10a

pyproject.toml

@@ -58,6 +58,7 @@ dev = [
   "pymdown-extensions>=10.0.0",
   "coverage>=7.8.0",
   "hypothesis>=6.131.20",
+  "pytest-cov>=6.3.0",
 ]
 
 [tool.uv.workspace]
@@ -103,8 +104,24 @@ convention = "google"
 [tool.ruff.format]
 docstring-code-format = true
 
+[tool.coverage.run]
+source = ["guardrails"]
+omit = [
+  "src/guardrails/evals/*",
+]
+
 [tool.mypy]
 strict = true
 disallow_incomplete_defs = false
 disallow_untyped_defs = false
 disallow_untyped_calls = false
+exclude = [
+  "examples",
+  "src/guardrails/evals",
+]
+
+[tool.pyright]
+ignore = [
+  "examples",
+  "src/guardrails/evals",
+]

tests/conftest.py

@@ -0,0 +1,129 @@
+"""Shared pytest fixtures for guardrails tests.
+
+These fixtures provide deterministic test environments by stubbing the OpenAI
+client library, seeding environment variables, and preventing accidental live
+network activity during the suite.
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+import types
+from collections.abc import Iterator
+from dataclasses import dataclass
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+
+class _StubOpenAIBase:
+    """Base stub with attribute bag behaviour for OpenAI client classes."""
+
+    def __init__(self, **kwargs: Any) -> None:
+        self._client_kwargs = kwargs
+        self.chat = SimpleNamespace()
+        self.responses = SimpleNamespace()
+        self.api_key = kwargs.get("api_key", "test-key")
+        self.base_url = kwargs.get("base_url")
+        self.organization = kwargs.get("organization")
+        self.timeout = kwargs.get("timeout")
+        self.max_retries = kwargs.get("max_retries")
+
+    def __getattr__(self, item: str) -> Any:
+        """Return None for unknown attributes to emulate real client laziness."""
+        return None
+
+
+class _StubAsyncOpenAI(_StubOpenAIBase):
+    """Stub asynchronous OpenAI client."""
+
+
+class _StubSyncOpenAI(_StubOpenAIBase):
+    """Stub synchronous OpenAI client."""
+
+
+@dataclass(frozen=True, slots=True)
+class _DummyResponse:
+    """Minimal response type with choices and output."""
+
+    choices: list[Any] | None = None
+    output: list[Any] | None = None
+    output_text: str | None = None
+    type: str | None = None
+    delta: str | None = None
+
+
+_STUB_OPENAI_MODULE = types.ModuleType("openai")
+_STUB_OPENAI_MODULE.AsyncOpenAI = _StubAsyncOpenAI
+_STUB_OPENAI_MODULE.OpenAI = _StubSyncOpenAI
+_STUB_OPENAI_MODULE.AsyncAzureOpenAI = _StubAsyncOpenAI
+_STUB_OPENAI_MODULE.AzureOpenAI = _StubSyncOpenAI
+_STUB_OPENAI_MODULE.NOT_GIVEN = object()
+
+
+class APITimeoutError(Exception):
+    """Stub API timeout error."""
+
+
+_STUB_OPENAI_MODULE.APITimeoutError = APITimeoutError
+
+_OPENAI_TYPES_MODULE = types.ModuleType("openai.types")
+_OPENAI_TYPES_MODULE.Completion = _DummyResponse
+_OPENAI_TYPES_MODULE.Response = _DummyResponse
+
+_OPENAI_CHAT_MODULE = types.ModuleType("openai.types.chat")
+_OPENAI_CHAT_MODULE.ChatCompletion = _DummyResponse
+_OPENAI_CHAT_MODULE.ChatCompletionChunk = _DummyResponse
+
+_OPENAI_RESPONSES_MODULE = types.ModuleType("openai.types.responses")
+_OPENAI_RESPONSES_MODULE.Response = _DummyResponse
+_OPENAI_RESPONSES_MODULE.ResponseInputItemParam = dict  # type: ignore[attr-defined]
+_OPENAI_RESPONSES_MODULE.ResponseOutputItem = dict  # type: ignore[attr-defined]
+_OPENAI_RESPONSES_MODULE.ResponseStreamEvent = dict  # type: ignore[attr-defined]
+
+
+_OPENAI_RESPONSES_RESPONSE_MODULE = types.ModuleType("openai.types.responses.response")
+_OPENAI_RESPONSES_RESPONSE_MODULE.Response = _DummyResponse
+
+
+class _ResponseTextConfigParam(dict):
+    """Stub config param used for response formatting."""
+
+
+_OPENAI_RESPONSES_MODULE.ResponseTextConfigParam = _ResponseTextConfigParam
+
+sys.modules["openai"] = _STUB_OPENAI_MODULE
+sys.modules["openai.types"] = _OPENAI_TYPES_MODULE
+sys.modules["openai.types.chat"] = _OPENAI_CHAT_MODULE
+sys.modules["openai.types.responses"] = _OPENAI_RESPONSES_MODULE
+sys.modules["openai.types.responses.response"] = _OPENAI_RESPONSES_RESPONSE_MODULE
+
+
+@pytest.fixture(autouse=True)
+def stub_openai_module(monkeypatch: pytest.MonkeyPatch) -> Iterator[types.ModuleType]:
+    """Provide stub OpenAI module so tests avoid real network-bound clients."""
+    # Patch imported symbols in guardrails modules
+    from guardrails import _base_client, client, types as guardrail_types  # type: ignore
+
+    monkeypatch.setattr(_base_client, "AsyncOpenAI", _StubAsyncOpenAI, raising=False)
+    monkeypatch.setattr(_base_client, "OpenAI", _StubSyncOpenAI, raising=False)
+    monkeypatch.setattr(client, "AsyncOpenAI", _StubAsyncOpenAI, raising=False)
+    monkeypatch.setattr(client, "OpenAI", _StubSyncOpenAI, raising=False)
+    monkeypatch.setattr(client, "AsyncAzureOpenAI", _StubAsyncOpenAI, raising=False)
+    monkeypatch.setattr(client, "AzureOpenAI", _StubSyncOpenAI, raising=False)
+    monkeypatch.setattr(guardrail_types, "AsyncOpenAI", _StubAsyncOpenAI, raising=False)
+    monkeypatch.setattr(guardrail_types, "OpenAI", _StubSyncOpenAI, raising=False)
+    monkeypatch.setattr(guardrail_types, "AsyncAzureOpenAI", _StubAsyncOpenAI, raising=False)
+    monkeypatch.setattr(guardrail_types, "AzureOpenAI", _StubSyncOpenAI, raising=False)
+
+    monkeypatch.setenv("OPENAI_API_KEY", "test-key")
+
+    yield _STUB_OPENAI_MODULE
+
+
+@pytest.fixture(autouse=True)
+def configure_logging() -> None:
+    """Ensure logging defaults to DEBUG for deterministic assertions."""
+    logging.basicConfig(level=logging.DEBUG)

tests/integration/test_suite.py

@@ -378,11 +378,7 @@ async def run_test(
             else:
                 # Find the triggered result
                 triggered_result = next(
-                    (
-                        r
-                        for r in response.guardrail_results.all_results
-                        if r.tripwire_triggered
-                    ),
+                    (r for r in response.guardrail_results.all_results if r.tripwire_triggered),
                     None,
                 )
                 info = triggered_result.info if triggered_result else None
@@ -394,9 +390,7 @@ async def run_test(
                         "details": {"result": info},
                     },
                 )
-                print(
-                    f"❌ {test.name} - Passing case {idx} triggered when it shouldn't"
-                )
+                print(f"❌ {test.name} - Passing case {idx} triggered when it shouldn't")
                 if info:
                     print(f"  Info: {info}")
 
@@ -427,11 +421,7 @@ async def run_test(
             if tripwire_triggered:
                 # Find the triggered result
                 triggered_result = next(
-                    (
-                        r
-                        for r in response.guardrail_results.all_results
-                        if r.tripwire_triggered
-                    ),
+                    (r for r in response.guardrail_results.all_results if r.tripwire_triggered),
                     None,
                 )
                 info = triggered_result.info if triggered_result else None
@@ -517,17 +507,9 @@ async def run_test_suite(
         results["tests"].append(outcome)
 
         # Calculate test status
-        passing_fails = sum(
-            1 for c in outcome["passing_cases"] if c["status"] == "FAIL"
-        )
-        failing_fails = sum(
-            1 for c in outcome["failing_cases"] if c["status"] == "FAIL"
-        )
-        errors = sum(
-            1
-            for c in outcome["passing_cases"] + outcome["failing_cases"]
-            if c["status"] == "ERROR"
-        )
+        passing_fails = sum(1 for c in outcome["passing_cases"] if c["status"] == "FAIL")
+        failing_fails = sum(1 for c in outcome["failing_cases"] if c["status"] == "FAIL")
+        errors = sum(1 for c in outcome["passing_cases"] + outcome["failing_cases"] if c["status"] == "ERROR")
 
         if errors > 0:
             results["summary"]["error_tests"] += 1
@@ -538,16 +520,8 @@ async def run_test_suite(
 
         # Count case results
         total_cases = len(outcome["passing_cases"]) + len(outcome["failing_cases"])
-        passed_cases = sum(
-            1
-            for c in outcome["passing_cases"] + outcome["failing_cases"]
-            if c["status"] == "PASS"
-        )
-        failed_cases = sum(
-            1
-            for c in outcome["passing_cases"] + outcome["failing_cases"]
-            if c["status"] == "FAIL"
-        )
+        passed_cases = sum(1 for c in outcome["passing_cases"] + outcome["failing_cases"] if c["status"] == "PASS")
+        failed_cases = sum(1 for c in outcome["passing_cases"] + outcome["failing_cases"] if c["status"] == "FAIL")
         error_cases = errors
 
         results["summary"]["total_cases"] += total_cases
@@ -564,9 +538,7 @@ def print_summary(results: dict[str, Any]) -> None:
     print("GUARDRAILS TEST SUMMARY")
     print("=" * 50)
     print(
-        f"Tests: {summary['passed_tests']} passed, "
-        f"{summary['failed_tests']} failed, "
-        f"{summary['error_tests']} errors",
+        f"Tests: {summary['passed_tests']} passed, " f"{summary['failed_tests']} failed, " f"{summary['error_tests']} errors",
     )
     print(
         f"Cases: {summary['total_cases']} total, "

tests/unit/checks/test_keywords.py

@@ -0,0 +1,68 @@
+"""Tests for keyword-based guardrail helpers."""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from guardrails.checks.text.competitors import CompetitorCfg, competitors
+from guardrails.checks.text.keywords import KeywordCfg, keywords, match_keywords
+from guardrails.types import GuardrailResult
+
+
+def test_match_keywords_sanitizes_trailing_punctuation() -> None:
+    """Ensure keyword sanitization strips trailing punctuation before matching."""
+    config = KeywordCfg(keywords=["token.", "secret!", "KEY?"])
+    result = match_keywords("Leaked token appears here.", config, guardrail_name="Test Guardrail")
+
+    assert result.tripwire_triggered is True  # noqa: S101
+    assert result.info["sanitized_keywords"] == ["token", "secret", "KEY"]  # noqa: S101
+    assert result.info["matched"] == ["token"]  # noqa: S101
+    assert result.info["guardrail_name"] == "Test Guardrail"  # noqa: S101
+    assert result.info["checked_text"] == "Leaked token appears here."  # noqa: S101
+
+
+def test_match_keywords_deduplicates_case_insensitive_matches() -> None:
+    """Repeated matches differing by case should be deduplicated."""
+    config = KeywordCfg(keywords=["Alert"])
+    result = match_keywords("alert ALERT Alert", config, guardrail_name="Keyword Filter")
+
+    assert result.tripwire_triggered is True  # noqa: S101
+    assert result.info["matched"] == ["alert"]  # noqa: S101
+
+
+@pytest.mark.asyncio
+async def test_keywords_guardrail_wraps_match_keywords() -> None:
+    """Async guardrail should mirror match_keywords behaviour."""
+    config = KeywordCfg(keywords=["breach"])
+    result = await keywords(ctx=None, data="Potential breach detected", config=config)
+
+    assert isinstance(result, GuardrailResult)  # noqa: S101
+    assert result.tripwire_triggered is True  # noqa: S101
+    assert result.info["guardrail_name"] == "Keyword Filter"  # noqa: S101
+
+
+@pytest.mark.asyncio
+async def test_competitors_uses_keyword_matching() -> None:
+    """Competitors guardrail delegates to keyword matching with distinct name."""
+    config = CompetitorCfg(keywords=["ACME Corp"])
+    result = await competitors(ctx=None, data="Comparing against ACME Corp today", config=config)
+
+    assert result.tripwire_triggered is True  # noqa: S101
+    assert result.info["guardrail_name"] == "Competitors"  # noqa: S101
+    assert result.info["matched"] == ["ACME Corp"]  # noqa: S101
+
+
+def test_keyword_cfg_requires_non_empty_keywords() -> None:
+    """KeywordCfg should enforce at least one keyword."""
+    with pytest.raises(ValidationError):
+        KeywordCfg(keywords=[])
+
+
+@pytest.mark.asyncio
+async def test_keywords_does_not_trigger_on_benign_text() -> None:
+    """Guardrail should not trigger when no keywords are present."""
+    config = KeywordCfg(keywords=["restricted"])
+    result = await keywords(ctx=None, data="Safe content", config=config)
+
+    assert result.tripwire_triggered is False  # noqa: S101