omnigent-ai--omnigent/tests/testing/test_guardrails.py

"""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 now used by the pytest session
guardrail hook.
"""

from __future__ import annotations

import logging

import pytest

from omnigent.testing.guardrails import (
    DEV_PORTS,
    TestGuardrailError,
    base_url_violation,
    check_test_environment,
    looks_like_pytest,
    looks_like_test_db,
)

# An env that asserts a test run without relying on the ambient
# PYTEST_CURRENT_TEST (which pytest only sets per-test).
_TEST_ENV = {"OMNIGENT_TEST_MODE": "1"}
_TMP_DB = "sqlite:////tmp/pytest-abc/test.db"


def _guardrail_warnings(records: list[logging.LogRecord]) -> list[str]:
    """Return the messages of records emitted by the guardrail logger."""
    return [
        r.getMessage()
        for r in records
        if r.name == "omnigent.testing.guardrails" and "TEST GUARDRAIL:" in r.getMessage()
    ]


# ── pass case ────────────────────────────────────────────


def test_clean_environment_emits_no_warning(caplog: pytest.LogCaptureFixture) -> None:
    """A test process + tmp DB + ephemeral base URL is clean."""
    with caplog.at_level(logging.WARNING):
        violations = check_test_environment(
            env=_TEST_ENV,
            db_uri=_TMP_DB,
            base_url="http://127.0.0.1:54321",
            warn_only=True,
        )
    assert violations == []
    assert _guardrail_warnings(caplog.records) == []


@pytest.mark.parametrize(
    "db_uri",
    [
        "sqlite://",
        "sqlite:///:memory:",
        "sqlite:///file::memory:?cache=shared",
        "sqlite:////tmp/test.db",
        "sqlite:////tmp/foo/test.db",
        "sqlite:///foo_test.db",
        "sqlite:////home/user/project/tests/session.db",
        "sqlite:///tests/session.db",
        "sqlite:////var/data/my_test_store.db",
    ],
)
def test_looks_like_test_db_accepts_throwaway_uris(db_uri: str) -> None:
    assert looks_like_test_db(db_uri) is True


@pytest.mark.parametrize(
    "db_uri",
    [
        "",
        "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


# ── each violation emits a warning ───────────────────────


def test_non_test_process_warns(
    caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A process with no test signal emits the process-check warning.

    ``looks_like_pytest`` consults ``sys.modules`` (pytest is always
    imported while these tests run), so we stub the module-set helper to
    simulate a non-test process, then confirm the violation is logged and
    warn mode still returns rather than raising.
    """
    from omnigent.testing import guardrails

    monkeypatch.setattr(guardrails, "_imported_modules", frozenset)
    with caplog.at_level(logging.WARNING):
        violations = check_test_environment(
            env={},
            db_uri=_TMP_DB,
            warn_only=True,
        )
    warnings = _guardrail_warnings(caplog.records)
    assert any("does not look like a pytest run" in w for w in warnings)
    assert any("pytest run" in v for v in violations)


def test_real_db_warns(caplog: pytest.LogCaptureFixture) -> None:
    with caplog.at_level(logging.WARNING):
        check_test_environment(
            env=_TEST_ENV,
            db_uri="sqlite:////home/alice/.omnigent/chat.db",
            warn_only=True,
        )
    warnings = _guardrail_warnings(caplog.records)
    assert any("does not look like a test DB" in w for w in warnings)


@pytest.mark.parametrize("port", sorted(DEV_PORTS))
def test_dev_port_base_url_warns(port: int, caplog: pytest.LogCaptureFixture) -> None:
    with caplog.at_level(logging.WARNING):
        violations = check_test_environment(
            env=_TEST_ENV,
            db_uri=_TMP_DB,
            base_url=f"http://localhost:{port}",
            warn_only=True,
        )
    warnings = _guardrail_warnings(caplog.records)
    assert any(f"port {port}" in w for w in warnings)
    assert any(str(port) in v for v in violations)


def test_dev_host_no_port_warns() -> None:
    assert base_url_violation("http://localhost") is not None
    assert base_url_violation("http://host.docker.internal") is not None


def test_ephemeral_base_url_is_clean() -> None:
    # Random free port on loopback -> the fixture-server shape. Clean.
    assert base_url_violation("http://127.0.0.1:49152") is None


def test_multiple_violations_each_warn(caplog: pytest.LogCaptureFixture) -> None:
    with caplog.at_level(logging.WARNING):
        violations = check_test_environment(
            env=_TEST_ENV,
            db_uri="sqlite:////home/alice/.omnigent/chat.db",
            base_url="http://localhost:8000",
            warn_only=True,
        )
    warnings = _guardrail_warnings(caplog.records)
    assert len(violations) == 2
    assert len(warnings) == 2


# ── warn_only=True never raises ──────────────────────────


def test_warn_only_never_raises() -> None:
    """Every violation at once must not raise when warn_only=True."""
    # Strip both the flag and PYTEST_CURRENT_TEST so even the process
    # check has its best shot at a violation; warn mode must still return.
    violations = check_test_environment(
        env={},
        db_uri="sqlite:////home/alice/.omnigent/chat.db",
        base_url="http://localhost:6767",
        warn_only=True,
    )
    assert isinstance(violations, list)
    # DB + base_url violations are deterministic regardless of sys.modules.
    assert any("test DB" in v for v in violations)
    assert any("6767" in v for v in violations)


# ── hard-fail mode (warn_only=False) ─────────────────────


def test_hard_fail_raises_on_violation() -> None:
    with pytest.raises(TestGuardrailError) as exc:
        check_test_environment(
            env=_TEST_ENV,
            db_uri="sqlite:////home/alice/.omnigent/chat.db",
            warn_only=False,
        )
    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 (
        check_test_environment(
            env=_TEST_ENV,
            db_uri=_TMP_DB,
            base_url="http://127.0.0.1:49152",
            warn_only=False,
        )
        == []
    )