tatlock/tests/core/test_tenant_guard.py

"""
Tests for the tenant isolation guard.

Isolation is tenant-based: the production tenant ("jpmschweitzer") owns
real data in the shared services, and every non-production environment
must run under the reserved test tenant ("llm_tester") or an explicit
"test_"-prefixed namespace.

Guard matrix covered here: dev/test/prod x default/explicit user, at
both config level (effective_default_user) and request-context
resolution (get_user).
"""

import pytest
from pydantic import ValidationError

from src.core.config import (
    PRODUCTION_TENANT,
    TEST_TENANT,
    Config,
    Environment,
)
from src.core.context import RequestContext, get_user


def make_config(**overrides) -> Config:
    """Build a Config isolated from the local .env file."""
    return Config(_env_file=None, **overrides)


@pytest.mark.unit
class TestEffectiveDefaultUserMatrix:
    """Config-level guard: effective_default_user per environment."""

    # --- development ---

    def test_dev_without_default_user_forces_test_tenant(self):
        config = make_config(ENVIRONMENT=Environment.DEVELOPMENT)
        assert config.effective_default_user == TEST_TENANT
        assert config.tenant_forced is False

    def test_dev_with_test_tenant_is_kept(self):
        config = make_config(ENVIRONMENT=Environment.DEVELOPMENT, DEFAULT_USER=TEST_TENANT)
        assert config.effective_default_user == TEST_TENANT
        assert config.tenant_forced is False

    def test_dev_with_test_prefixed_override_is_kept(self):
        config = make_config(ENVIRONMENT=Environment.DEVELOPMENT, DEFAULT_USER="test_phase_b")
        assert config.effective_default_user == "test_phase_b"
        assert config.tenant_forced is False

    def test_dev_with_misconfigured_user_is_forced_to_test_tenant(self):
        config = make_config(ENVIRONMENT=Environment.DEVELOPMENT, DEFAULT_USER="alice")
        assert config.effective_default_user == TEST_TENANT
        assert config.tenant_forced is True

    def test_dev_with_production_tenant_refuses_startup(self):
        with pytest.raises(ValidationError) as exc_info:
            make_config(
                ENVIRONMENT=Environment.DEVELOPMENT,
                DEFAULT_USER=PRODUCTION_TENANT,
            )
        assert "Refusing to start" in str(exc_info.value)
        assert PRODUCTION_TENANT in str(exc_info.value)

    # --- testing ---

    def test_testing_without_default_user_forces_test_tenant(self):
        config = make_config(ENVIRONMENT=Environment.TESTING)
        assert config.effective_default_user == TEST_TENANT

    def test_testing_with_misconfigured_user_is_forced_to_test_tenant(self):
        config = make_config(ENVIRONMENT=Environment.TESTING, DEFAULT_USER="bob")
        assert config.effective_default_user == TEST_TENANT
        assert config.tenant_forced is True

    def test_testing_with_production_tenant_refuses_startup(self):
        with pytest.raises(ValidationError, match="Refusing to start"):
            make_config(
                ENVIRONMENT=Environment.TESTING,
                DEFAULT_USER=PRODUCTION_TENANT,
            )

    def test_testing_with_test_prefixed_override_is_kept(self):
        config = make_config(ENVIRONMENT=Environment.TESTING, DEFAULT_USER="test_ci_run")
        assert config.effective_default_user == "test_ci_run"

    # --- production ---

    def test_prod_without_default_user_uses_production_tenant(self):
        config = make_config(ENVIRONMENT=Environment.PRODUCTION)
        assert config.effective_default_user == PRODUCTION_TENANT
        assert config.tenant_forced is False

    def test_prod_with_explicit_production_tenant_is_kept(self):
        config = make_config(ENVIRONMENT=Environment.PRODUCTION, DEFAULT_USER=PRODUCTION_TENANT)
        assert config.effective_default_user == PRODUCTION_TENANT

    def test_prod_with_explicit_other_user_is_kept(self):
        config = make_config(ENVIRONMENT=Environment.PRODUCTION, DEFAULT_USER="household_guest")
        assert config.effective_default_user == "household_guest"
        assert config.tenant_forced is False


@pytest.mark.unit
class TestRequestContextGuard:
    """Request-context resolution guard: get_user() per environment."""

    def _patch_environment(self, monkeypatch, environment: Environment):
        from src.core import config as config_module

        monkeypatch.setattr(config_module.config, "ENVIRONMENT", environment)

    def test_dev_default_resolution_is_test_tenant(self, monkeypatch):
        self._patch_environment(monkeypatch, Environment.DEVELOPMENT)
        assert get_user() == TEST_TENANT

    def test_dev_explicit_production_tenant_is_forced(self, monkeypatch):
        self._patch_environment(monkeypatch, Environment.DEVELOPMENT)
        with RequestContext(user=PRODUCTION_TENANT):
            assert get_user() == TEST_TENANT

    def test_testing_explicit_production_tenant_is_forced(self, monkeypatch):
        self._patch_environment(monkeypatch, Environment.TESTING)
        with RequestContext(user=PRODUCTION_TENANT):
            assert get_user() == TEST_TENANT

    def test_dev_explicit_other_user_passes_through(self, monkeypatch):
        self._patch_environment(monkeypatch, Environment.DEVELOPMENT)
        with RequestContext(user="testuser"):
            assert get_user() == "testuser"

    def test_prod_explicit_production_tenant_passes_through(self, monkeypatch):
        self._patch_environment(monkeypatch, Environment.PRODUCTION)
        with RequestContext(user=PRODUCTION_TENANT):
            assert get_user() == PRODUCTION_TENANT

    def test_prod_explicit_other_user_passes_through(self, monkeypatch):
        self._patch_environment(monkeypatch, Environment.PRODUCTION)
        with RequestContext(user="alice"):
            assert get_user() == "alice"


@pytest.mark.unit
class TestSuiteRunsUnderTestTenant:
    """
    The live test session itself must resolve to the test tenant.

    The session guard in tests/conftest.py hard-fails the suite when the
    effective tenant is the production tenant; these tests assert the
    namespaces every shared-service touch would use (Qdrant memories
    collection, Redis session keys) are the llm_tester ones.
    """

    def test_effective_tenant_is_not_production(self):
        from src.core.context import get_default_user

        assert get_default_user() != PRODUCTION_TENANT

    def test_effective_tenant_is_the_reserved_test_tenant(self):
        from src.core.context import get_default_user

        assert get_default_user() == TEST_TENANT

    def test_memories_collection_namespace_is_test_tenant(self):
        from src.core.context import get_default_user
        from src.core.multi_tenancy import get_memory_collection_name

        assert (
            get_memory_collection_name(get_default_user())
            == f"memories_{TEST_TENANT}"
        )

    def test_redis_session_namespace_is_test_tenant(self):
        from src.core.context import get_default_user
        from src.core.multi_tenancy import get_session_key

        key = get_session_key(get_default_user(), "conv_test")
        assert key.startswith(f"session:{TEST_TENANT}:")


@pytest.mark.unit
class TestStartupTenantGuardLog:
    """One loud startup log line states the effective tenant."""

    def test_non_production_logs_forced_tenant(self, monkeypatch):
        from src.core import startup as startup_module

        events = []

        class _Recorder:
            def warning(self, event, **kw):
                events.append((event, kw))

            def info(self, event, **kw):
                events.append((event, kw))

        monkeypatch.setattr(startup_module, "logger", _Recorder())
        monkeypatch.setattr(startup_module.config, "ENVIRONMENT", Environment.DEVELOPMENT)

        startup_module.log_tenant_guard()

        assert events == [
            (
                "tenant_guard_active",
                {
                    "environment": "development",
                    "forced_tenant": TEST_TENANT,
                    "default_user_overridden": startup_module.config.tenant_forced,
                    "configured_default_user": startup_module.config.DEFAULT_USER,
                },
            )
        ]

    def test_production_logs_production_tenant(self, monkeypatch):
        from src.core import startup as startup_module

        events = []

        class _Recorder:
            def warning(self, event, **kw):
                events.append(("warning", event, kw))

            def info(self, event, **kw):
                events.append(("info", event, kw))

        monkeypatch.setattr(startup_module, "logger", _Recorder())
        monkeypatch.setattr(startup_module.config, "ENVIRONMENT", Environment.PRODUCTION)

        startup_module.log_tenant_guard()

        assert events == [
            (
                "info",
                "tenant_guard_production",
                {"environment": "production", "tenant": PRODUCTION_TENANT},
            )
        ]