Address guardrail review feedback

omnigent/testing/guardrails.py

@@ -1,15 +1,15 @@
-"""Warn-mode test-environment guardrails.
+"""Test-environment guardrails.
 
 A small, additive safety net that checks a test run is pointed at
 throwaway resources — running under pytest, against a tmp/in-memory
 SQLite DB, and not aimed at a known dev/prod host or port — *before*
 the suite starts mutating state.
 
-Today every check is **warn-only**: a violation logs a clear
-``TEST GUARDRAIL:`` ``WARNING`` and the run continues. The design keeps
-a single ``warn_only`` switch so a future PR can flip the default to
-``False`` and have the same checks hard-fail (raise
-:class:`TestGuardrailError`) with no other code change.
+When checks hard-fail, a violation raises :class:`TestGuardrailError`
+before the suite can mutate real resources. Set
+``OMNIGENT_DISABLE_TEST_GUARDRAILS`` to a truthy value (``1``, ``true``,
+``yes``, or ``on``) to temporarily downgrade violations to warn-only for
+deliberate integration runs that target non-test resources.
 
 Entry point: :func:`check_test_environment`.
 """
@@ -18,14 +18,15 @@ from __future__ import annotations
 
 import logging
 import os
+import re
 import tempfile
 from collections.abc import Mapping
 from pathlib import Path
 
 _logger = logging.getLogger(__name__)
 
-# Prefix on every guardrail log line so violations are greppable and so a
-# future hard-fail mode can reuse the identical message.
+# Prefix on every guardrail log line so violations are greppable and
+# hard-fail exceptions reuse the identical message.
 _WARN_PREFIX = "TEST GUARDRAIL:"
 
 # Ports we never want a test to drive: the local server default (6767),
@@ -53,15 +54,16 @@ DEV_HOSTS: frozenset[str] = frozenset(
 # set per-test, not at session configure time). OMNIGENT_TEST_MODE is
 # introduced by this module as the canonical, settable flag.
 _TEST_MODE_ENV_VARS = ("OMNIGENT_TEST_MODE", "OMNIGENT_ENV")
-_TEST_MODE_ENV_VALUES = frozenset({"1", "true", "test", "testing", "yes", "on"})
+_TRUTHY_ENV_VALUES = frozenset({"1", "true", "yes", "on"})
+_TEST_MODE_ENV_VALUES = _TRUTHY_ENV_VALUES | {"test", "testing"}
+_DISABLE_GUARDRAILS_ENV_VAR = "OMNIGENT_DISABLE_TEST_GUARDRAILS"
 
 
 class TestGuardrailError(AssertionError):
     """Raised by :func:`check_test_environment` when ``warn_only=False``.
 
-    Subclasses :class:`AssertionError` so a future hard-fail mode reads
-    naturally as a failed test precondition. Unused while the default is
-    warn-only, but defined now so the hard-fail flip is a one-line change.
+    Subclasses :class:`AssertionError` so hard-fail mode reads naturally
+    as a failed test precondition.
     """
 
     # Tell pytest this is not a test class despite the ``Test`` prefix.
@@ -102,10 +104,10 @@ def _imported_modules() -> frozenset[str]:
 def looks_like_test_db(db_uri: str) -> bool:
     """Return whether *db_uri* looks like a throwaway test database.
 
-    Accepts in-memory SQLite, file SQLite under a system temp dir, or any
-    URI whose path/name contains ``test``. Everything else (a real
-    ``~/.omnigent/chat.db``, a Postgres ``DATABASE_URL``) is treated as a
-    non-test DB.
+    Accepts in-memory SQLite, file SQLite under a system temp dir, or a
+    file-backed SQLite path with ``test`` or ``tests`` as a delimited
+    path/name token. Everything else (a real ``~/.omnigent/chat.db``, a Postgres
+    ``DATABASE_URL``) is treated as a non-test DB.
 
     :param db_uri: A SQLAlchemy-style URI, e.g. ``sqlite:///…`` .
     :returns: ``True`` if the URI looks like a test DB.
@@ -120,10 +122,13 @@ def looks_like_test_db(db_uri: str) -> bool:
     if lowered in ("sqlite://", "sqlite:///"):
         return True
 
-    if "test" in lowered:
-        return True
-
     path = _sqlite_path(db_uri)
+    # Only treat a ``test`` token as proof for file-backed SQLite paths.
+    # Non-SQLite authorities such as ``postgresql://prod-test-cluster/app``
+    # may contain ``test`` in a real host name and must not be silently
+    # accepted as throwaway DBs.
+    if path is not None and _sqlite_path_has_test_token(path):
+        return True
     if path is not None and _under_temp_dir(path):
         return True
 
@@ -147,6 +152,16 @@ def _sqlite_path(db_uri: str) -> Path | None:
     return Path(raw)
 
 
+def _sqlite_path_has_test_token(path: Path) -> bool:
+    """Return whether a SQLite path has ``test``/``tests`` as a delimited token."""
+    return any(_has_test_token(part) for part in path.parts)
+
+
+def _has_test_token(value: str) -> bool:
+    """Return whether ``test``/``tests`` appears as a delimited token."""
+    return re.search(r"(?<![a-z0-9])tests?(?![a-z0-9])", value.lower()) is not None
+
+
 def _under_temp_dir(path: Path) -> bool:
     """Return whether *path* lives under a system temp directory.
 
@@ -195,6 +210,10 @@ def base_url_violation(base_url: str) -> str | None:
         # A named dev host with no explicit port (default 80/443) still
         # smells like a real instance rather than an ephemeral fixture.
         return f"base_url {base_url!r} targets dev host {host!r}"
+    # A dev host with an explicit random/non-dev port is intentionally
+    # clean: fixture servers legitimately bind localhost on ephemeral
+    # free ports. Only known dev/prod ports, or named dev hosts with no
+    # explicit port, are flagged.
     return None
 
 
@@ -214,14 +233,19 @@ def check_test_environment(
     c. *base_url*, when given, does not target a dev/prod host or port
        (:func:`base_url_violation`).
 
-    When ``warn_only`` is ``True`` (the default today) each violation is
-    logged as a ``WARNING`` prefixed with ``TEST GUARDRAIL:`` and the
-    function returns the list of reasons. When ``warn_only`` is ``False``
-    the same set of violations raises :class:`TestGuardrailError` — this
-    is the future hard-fail mode and is not used by the suite yet.
+    When ``warn_only`` is ``True`` each violation is logged as a
+    ``WARNING`` prefixed with ``TEST GUARDRAIL:`` and the function
+    returns the list of reasons. When ``warn_only`` is ``False`` the same
+    set of violations raises :class:`TestGuardrailError`, unless
+    ``OMNIGENT_DISABLE_TEST_GUARDRAILS`` is truthy, in which case
+    violations are logged and returned instead.
+
+    The pytest session hook passes ``warn_only=False``; the ``True``
+    default is retained for ad-hoc/library callers.
 
     :param env: Environment mapping; defaults to ``os.environ``.
-    :param db_uri: The resolved store DB URI for this run.
+    :param db_uri: The resolved store DB URI for this run; empty values
+        skip the DB check.
     :param base_url: Optional base URL the test will drive.
     :param warn_only: Log instead of raise on violation (default ``True``).
     :returns: The list of violation reason strings (empty when clean).
@@ -238,11 +262,14 @@ def check_test_environment(
             "(no PYTEST_CURRENT_TEST / OMNIGENT_TEST_MODE and pytest not imported)"
         )
 
-    if not looks_like_test_db(db_uri):
-        violations.append(
-            f"db_uri {db_uri!r} does not look like a test DB "
-            "(expected an in-memory/tmp SQLite or a URI containing 'test')"
-        )
+    if db_uri.strip():
+        if not looks_like_test_db(db_uri):
+            violations.append(
+                f"db_uri {db_uri!r} does not look like a test DB "
+                "(expected an in-memory/tmp SQLite or a SQLite path with 'test'/'tests')"
+            )
+    else:
+        _logger.debug("%s db_uri is blank; skipping DB check", _WARN_PREFIX)
 
     if base_url is not None:
         reason = base_url_violation(base_url)
@@ -252,9 +279,21 @@ def check_test_environment(
     if not violations:
         return violations
 
-    if warn_only:
+    guardrails_disabled = _guardrails_disabled(env)
+    if warn_only or guardrails_disabled:
+        if not warn_only and guardrails_disabled:
+            _logger.warning(
+                "%s escape hatch active (%s) — hard-fail suppressed",
+                _WARN_PREFIX,
+                _DISABLE_GUARDRAILS_ENV_VAR,
+            )
         for reason in violations:
             _logger.warning("%s %s", _WARN_PREFIX, reason)
         return violations
 
     raise TestGuardrailError(f"{_WARN_PREFIX} " + "; ".join(violations))
+
+
+def _guardrails_disabled(env: Mapping[str, str]) -> bool:
+    """Return whether the global test-guardrail escape hatch is enabled."""
+    return env.get(_DISABLE_GUARDRAILS_ENV_VAR, "").strip().lower() in _TRUTHY_ENV_VALUES

tests/conftest.py

@@ -205,13 +205,12 @@ def pytest_configure(config: pytest.Config) -> None:
 
 
 def _run_test_environment_guardrails(config: pytest.Config) -> None:
-    """Surface test-environment guardrail warnings at session start.
+    """Enforce test-environment guardrails at session start.
 
-    Warn-only: :func:`check_test_environment` logs ``TEST GUARDRAIL:``
-    warnings for anything that looks like a real (non-test) DB or a base
-    URL aimed at a dev/prod host or port, and never raises in this mode.
-    A future PR can flip ``warn_only=False`` to make these hard
-    preconditions — this call site needs no change for that.
+    Hard-fail: :func:`check_test_environment` raises on anything that
+    looks like a real (non-test) DB or a base URL aimed at a dev/prod host
+    or port. Set ``OMNIGENT_DISABLE_TEST_GUARDRAILS=1`` to temporarily
+    downgrade violations to warn-only for deliberate integration runs.
 
     The resolved DB URI mirrors how a run would pick one: an explicit
     ``OMNIGENT_DATABASE_URI`` wins (so pointing the suite at a real DB
@@ -222,7 +221,7 @@ def _run_test_environment_guardrails(config: pytest.Config) -> None:
 
     db_uri = os.environ.get("OMNIGENT_DATABASE_URI") or os.environ.get("MLFLOW_TRACKING_URI", "")
     base_url = config.getoption("--omnigent-server-url", default=None)
-    check_test_environment(db_uri=db_uri, base_url=base_url, warn_only=True)
+    check_test_environment(db_uri=db_uri, base_url=base_url, warn_only=False)
 
 
 def pytest_unconfigure(config: pytest.Config) -> None:

tests/testing/test_guardrails.py

@@ -1,9 +1,9 @@
-"""Unit tests for the warn-mode test-environment guardrails.
+"""Unit tests for the test-environment guardrails.
 
 In-process only: no server, no browser. Exercises the pass case, each
 violation's warning, and the contract that ``warn_only=True`` never
-raises. Also covers the hard-fail path so a future flip of the default
-is already verified.
+raises. Also covers the hard-fail path now used by the pytest session
+guardrail hook.
 """
 
 from __future__ import annotations
@@ -58,8 +58,11 @@ def test_clean_environment_emits_no_warning(caplog: pytest.LogCaptureFixture) ->
         "sqlite://",
         "sqlite:///:memory:",
         "sqlite:///file::memory:?cache=shared",
+        "sqlite:////tmp/test.db",
         "sqlite:////tmp/foo/test.db",
-        "postgresql://localhost/test_db",
+        "sqlite:///foo_test.db",
+        "sqlite:////home/user/project/tests/session.db",
+        "sqlite:///tests/session.db",
         "sqlite:////var/data/my_test_store.db",
     ],
 )
@@ -72,13 +75,29 @@ def test_looks_like_test_db_accepts_throwaway_uris(db_uri: str) -> None:
     [
         "",
         "sqlite:////home/alice/.omnigent/chat.db",
+        "sqlite:///testing.db",
+        "sqlite:///test123.db",
+        "sqlite:///contest.db",
+        "sqlite:///latest.db",
         "postgresql://prod-host:5432/omnigent",
+        "postgresql://prod-test-cluster/app",
+        "postgres://h/latest",
     ],
 )
 def test_looks_like_test_db_rejects_real_uris(db_uri: str) -> None:
     assert looks_like_test_db(db_uri) is False
 
 
+@pytest.mark.parametrize("db_uri", ["", "   "])
+def test_empty_db_uri_skips_db_check(
+    db_uri: str,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.DEBUG):
+        assert check_test_environment(env=_TEST_ENV, db_uri=db_uri, warn_only=False) == []
+    assert any("db_uri is blank; skipping DB check" in r.getMessage() for r in caplog.records)
+
+
 def test_looks_like_pytest_via_flag() -> None:
     assert looks_like_pytest({"OMNIGENT_TEST_MODE": "1"}) is True
     assert looks_like_pytest({"PYTEST_CURRENT_TEST": "x::y (call)"}) is True
@@ -178,7 +197,7 @@ def test_warn_only_never_raises() -> None:
     assert any("6767" in v for v in violations)
 
 
-# ── future hard-fail mode (warn_only=False) ──────────────
+# ── hard-fail mode (warn_only=False) ─────────────────────
 
 
 def test_hard_fail_raises_on_violation() -> None:
@@ -191,6 +210,59 @@ def test_hard_fail_raises_on_violation() -> None:
     assert "TEST GUARDRAIL:" in str(exc.value)
 
 
+@pytest.mark.parametrize(
+    "db_uri",
+    [
+        "postgresql://prod-test-cluster/app",
+        "postgres://h/latest",
+    ],
+)
+def test_hard_fail_rejects_non_sqlite_test_substrings(db_uri: str) -> None:
+    with pytest.raises(TestGuardrailError) as exc:
+        check_test_environment(env=_TEST_ENV, db_uri=db_uri, warn_only=False)
+    assert "does not look like a test DB" in str(exc.value)
+
+
+def test_escape_hatch_downgrades_hard_fail_to_warning(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    env = {**_TEST_ENV, "OMNIGENT_DISABLE_TEST_GUARDRAILS": "yes"}
+    with caplog.at_level(logging.WARNING):
+        violations = check_test_environment(
+            env=env,
+            db_uri="sqlite:////home/alice/.omnigent/chat.db",
+            warn_only=False,
+        )
+    assert any("does not look like a test DB" in v for v in violations)
+    assert any("escape hatch active" in w for w in _guardrail_warnings(caplog.records))
+    assert any("does not look like a test DB" in w for w in _guardrail_warnings(caplog.records))
+
+
+def test_escape_hatch_message_only_for_hard_fail_suppression(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    env = {**_TEST_ENV, "OMNIGENT_DISABLE_TEST_GUARDRAILS": "yes"}
+    with caplog.at_level(logging.WARNING):
+        check_test_environment(
+            env=env,
+            db_uri="sqlite:////home/alice/.omnigent/chat.db",
+            warn_only=True,
+        )
+        check_test_environment(env=env, db_uri=_TMP_DB, warn_only=False)
+    assert not any("escape hatch active" in r.getMessage() for r in caplog.records)
+
+
+def test_dev_port_base_url_hard_fails() -> None:
+    with pytest.raises(TestGuardrailError) as exc:
+        check_test_environment(
+            env=_TEST_ENV,
+            db_uri=_TMP_DB,
+            base_url="http://localhost:6767",
+            warn_only=False,
+        )
+    assert "port 6767" in str(exc.value)
+
+
 def test_hard_fail_passes_when_clean() -> None:
     # No exception, returns empty list.
     assert (