hold-slayer/tests/test_data_layer.py

"""
Data-layer tests.

Alembic migrations produce the schema the ORM models declare (and
adopt a pre-Alembic database); a call gets a durable in_progress row
the moment it starts; transcript entries carry real offsets; the
consolidated response models map from the domain in one place.
"""

import asyncio

import pytest
from sqlalchemy import Boolean, inspect, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

import db.database as dbmod
from core.call_manager import CallManager
from core.event_bus import EventBus
from db.database import Base
from models.call import ActiveCall, CallStatus, CallStatusResponse
from models.device import Device, DeviceType
from services import call_persistence as store

# ================================================================
# Alembic migrations
# ================================================================

def _run_alembic(connection, revision: str) -> None:
    from alembic import command
    from alembic.config import Config

    cfg = Config("alembic.ini")
    cfg.attributes["connection"] = connection
    command.upgrade(cfg, revision)


def _schema_info(sync_conn) -> dict:
    inspector = inspect(sync_conn)
    return {
        "tables": set(inspector.get_table_names()) - {"alembic_version"},
        "call_record_cols": {c["name"] for c in inspector.get_columns("call_records")},
        "is_online_type": next(
            c["type"] for c in inspector.get_columns("devices")
            if c["name"] == "is_online"
        ),
    }


class TestMigrations:
    async def test_upgrade_head_matches_models(self, tmp_path):
        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/mig.db")
        async with engine.begin() as conn:
            await conn.run_sync(dbmod._upgrade_to_head)
        async with engine.connect() as conn:
            info = await conn.run_sync(_schema_info)
        await engine.dispose()

        assert info["tables"] == set(Base.metadata.tables)
        assert "transcript" not in info["call_record_cols"]
        assert isinstance(info["is_online_type"], Boolean)

    async def test_adopts_pre_alembic_schema(self, tmp_path):
        """A create_all-era database (baseline schema, no alembic_version)
        is stamped and migrated forward instead of failing."""
        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/legacy.db")
        async with engine.begin() as conn:
            await conn.run_sync(
                lambda c: _run_alembic(c, dbmod._BASELINE_REVISION)
            )
            await conn.execute(text("DROP TABLE alembic_version"))

        async with engine.begin() as conn:
            await conn.run_sync(dbmod._upgrade_to_head)
        async with engine.connect() as conn:
            info = await conn.run_sync(_schema_info)
        await engine.dispose()

        assert "transcript" not in info["call_record_cols"]
        assert isinstance(info["is_online_type"], Boolean)


# ================================================================
# Durable call rows
# ================================================================

@pytest.fixture
async def mem_db(monkeypatch):
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        poolclass=StaticPool,
        connect_args={"check_same_thread": False},
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    factory = async_sessionmaker(engine, expire_on_commit=False)
    monkeypatch.setattr(dbmod, "_engine", engine)
    monkeypatch.setattr(dbmod, "_session_factory", factory)
    yield factory
    await engine.dispose()


class TestDurableCallRows:
    async def test_in_progress_row_from_the_start(self, mem_db):
        cm = CallManager(
            EventBus(),
            on_call_created=store.persist_call_on_create,
            on_call_ended=store.persist_call_on_end,
        )
        call = await cm.create_call("+15551230000", intent="dispute a charge")

        async with mem_db() as session:
            row = await store.get_record(session, call.id)
        assert row is not None
        assert row.status == "in_progress"
        assert row.ended_at is None
        assert row.intent == "dispute a charge"

        await cm.add_transcript(call.id, "hello, billing please", speaker="caller")
        await cm.end_call(call.id, CallStatus.COMPLETED)

        async with mem_db() as session:
            row = await store.get_record(session, call.id)
            chunks = await store.get_transcript_chunks(session, call.id)
        assert row.status == "completed"
        assert row.ended_at is not None
        assert [(c.seq, c.speaker, c.text) for c in chunks] == [
            (0, "caller", "hello, billing please")
        ]

    async def test_end_without_create_still_writes_row(self, mem_db):
        """If the create-time insert never happened, finalize inserts."""
        cm = CallManager(EventBus(), on_call_ended=store.persist_call_on_end)
        call = await cm.create_call("+15551230001")
        await cm.end_call(call.id, CallStatus.FAILED)

        async with mem_db() as session:
            row = await store.get_record(session, call.id)
        assert row is not None
        assert row.status == "failed"


# ================================================================
# Transcript offsets
# ================================================================

class TestTranscriptOffsets:
    async def test_entries_carry_offset_and_speaker(self):
        cm = CallManager(EventBus())
        call = await cm.create_call("+15551230002")

        await cm.add_transcript(call.id, "one")
        await asyncio.sleep(0.02)
        await cm.add_transcript(call.id, "two", speaker="agent")

        first, second = call.transcript_chunks
        assert first.t_offset_ms >= 0
        assert second.t_offset_ms > first.t_offset_ms
        assert first.speaker == "unknown"
        assert second.speaker == "agent"
        assert call.transcript == "one\ntwo"


# ================================================================
# Consolidated response models
# ================================================================

class TestResponseModels:
    def test_status_response_from_call(self):
        call = ActiveCall(id="call_x", remote_number="+15550000000", intent="pay bill")
        resp = CallStatusResponse.from_call(call)
        assert resp.call_id == "call_x"
        assert resp.status == "initiating"
        assert resp.remote_number == "+15550000000"
        assert resp.intent == "pay bill"

    def test_device_serializes_routability(self):
        device = Device(
            id="dev_1",
            name="Desk Phone",
            type=DeviceType.SIP_PHONE,
            sip_uri="sip:desk@gw",
            is_online=True,
        )
        assert device.model_dump()["can_receive_call"] is True
        device.dnd = True
        assert device.model_dump()["can_receive_call"] is False