""" 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