tests/test_live_wiring.py

"""Guard the env-gating contract for the live wiring (no network, no creds).

The app has no offline product mode: it requires live inference and a real event
store and refuses to run without them.  These tests pin that *selection* behaviour
so it cannot silently regress:

  * Models  — :func:`~src.models.openai_compat.has_live_credentials` is False unless a
    Modal/HF binding is set, and :meth:`ModelRouter.from_env` always builds the live
    path (the deterministic stub is reachable only via ``ModelRouter(offline=True)``).
  * Ledger  — :func:`~src.core.ledger_factory.make_ledger` raises with no ``DATABASE_URL``
    and returns the ``SqlAlchemyLedger`` for a URL.
  * Memory  — :func:`~src.core.memory_index.memory_index_from_env` is ``None`` when
    the gate is unset and a cloud index when ``MEMORY_INDEX=cloud``.

Everything runs on ``monkeypatch`` env edits (auto-reverted) and an in-memory SQLite
URL so the suite never touches a real network, a database server, or live credentials.
These cases assert the real wiring, so they opt out of the mock-infra fixture.
"""

from __future__ import annotations

import pytest

from src.core.ledger_factory import make_ledger
from src.core.memory_index import memory_index_from_env
from src.models.openai_compat import has_live_credentials
from src.models.provider import DeterministicTinyModel
from src.models.router import ModelRouter

pytestmark = pytest.mark.real_infra

# Env vars these tests touch; cleared before each case so the host environment and
# sibling tests never leak into (or out of) a case.
_LIVE_ENV = (
    "MODAL_WORKSPACE",
    "MODAL_LLM_BASE_URL",
    "DATABASE_URL",
    "MEMORY_INDEX",
    "MEMORY_INDEX_BACKEND",
    "MEM0_API_KEY",
)


@pytest.fixture(autouse=True)
def _clean_live_env(monkeypatch):
    """Start each case from a known-clean slate (monkeypatch auto-reverts after)."""
    for name in _LIVE_ENV:
        monkeypatch.delenv(name, raising=False)


# ── models: always live; the stub is reachable only by explicit opt-in ─────────


class TestCredentialDetection:
    def test_has_live_credentials_false_when_unset(self):
        assert has_live_credentials() is False

    def test_workspace_marks_live_credentials(self, monkeypatch):
        monkeypatch.setenv("MODAL_WORKSPACE", "demo-workspace")
        assert has_live_credentials() is True

    def test_base_url_marks_live_credentials(self, monkeypatch):
        monkeypatch.setenv("MODAL_LLM_BASE_URL", "https://demo.modal.run/v1")
        assert has_live_credentials() is True

    def test_blank_workspace_is_not_a_binding(self, monkeypatch):
        # An empty/whitespace value must NOT count as a live binding.
        monkeypatch.setenv("MODAL_WORKSPACE", "   ")
        assert has_live_credentials() is False


class TestRouterAlwaysLive:
    def test_from_env_is_live_when_unset(self):
        # No offline auto-detection: from_env always builds the live path.
        router = ModelRouter.from_env()
        assert router.offline is False
        assert not isinstance(router.for_profile("fast"), DeterministicTinyModel)

    def test_from_env_is_live_when_bound(self, monkeypatch):
        monkeypatch.setenv("MODAL_WORKSPACE", "demo-workspace")
        assert ModelRouter.from_env().offline is False

    def test_stub_only_via_explicit_offline(self):
        # The deterministic stub is the test seam, never selected automatically.
        assert isinstance(ModelRouter(offline=True).for_profile("fast"), DeterministicTinyModel)


# ── ledger: required — raises with no URL, SqlAlchemy-backed for a URL ──────────


class TestLedgerSelection:
    def test_no_url_raises(self):
        with pytest.raises(RuntimeError, match="DATABASE_URL is required"):
            make_ledger()

    def test_sqlite_url_returns_sqlalchemy_ledger(self):
        from src.core.sqlalchemy_ledger import SqlAlchemyLedger

        ledger = make_ledger("sqlite:///:memory:")
        assert isinstance(ledger, SqlAlchemyLedger)

    def test_sqlite_ledger_append_read_roundtrip(self):
        from src.core.events import Event

        ledger = make_ledger("sqlite:///:memory:")
        event = Event(run_id="r1", turn=0, kind="run.started", actor="engine", payload={"text": "hi"})
        ledger.append(event)
        assert [e.id for e in ledger.events] == [event.id]


# ── memory: None when the gate is unset, cloud index when MEMORY_INDEX=cloud ───


class TestMemoryIndexSelection:
    def test_none_when_gate_unset(self):
        assert memory_index_from_env() is None

    def test_cloud_index_when_gate_is_cloud(self, monkeypatch):
        pytest.importorskip("mem0")
        from src.core.memory_index import Mem0CloudIndex

        monkeypatch.setenv("MEMORY_INDEX", "cloud")
        monkeypatch.setenv("MEM0_API_KEY", "dummy-key")
        index = memory_index_from_env()
        assert isinstance(index, Mem0CloudIndex)