hold-slayer/tests/test_structure.py

"""
Composition and API-surface tests.

Covers the gateway composed the way main.py's lifespan composes it
(mode handlers, on_call_ended hook, receptionist-owned inbound
policy) and the REST routes running against a real (SQLite) database
through the shared data layer in services/call_persistence.py.
"""

import httpx
import pytest
from pydantic import SecretStr
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool

import main
from config import ReceptionistSettings, Settings, get_settings
from core.gateway import AIPSTNGateway
from db.database import Base, CallRecord, get_db
from models.call import CallMode, CallStatus
from models.routing import RoutingAction, RoutingActionType, RoutingDecision
from services.receptionist import ReceptionistService

# ================================================================
# Gateway composition
# ================================================================

class TestGatewayComposition:
    async def test_mode_handler_launches_per_call(self):
        gateway = AIPSTNGateway(settings=Settings(max_concurrent_calls=4))
        launched: list[tuple] = []
        gateway.register_mode_handler(
            CallMode.HOLD_SLAYER,
            lambda call, leg_id, flow_id: launched.append((call.id, leg_id, flow_id)),
        )

        call = await gateway.make_call("+15551234567", mode=CallMode.HOLD_SLAYER,
                                       call_flow_id="acme-main")

        assert launched == [(call.id, gateway.call_manager.legs_for_call(call.id)[0], "acme-main")]

    async def test_direct_mode_needs_no_handler(self):
        gateway = AIPSTNGateway(settings=Settings(max_concurrent_calls=4))
        call = await gateway.make_call("+15551234567")
        assert call.status == CallStatus.RINGING

    async def test_on_call_ended_hook_from_constructor(self):
        ended: list[tuple] = []

        async def hook(call, status):
            ended.append((call.id, status))

        gateway = AIPSTNGateway(
            settings=Settings(max_concurrent_calls=4), on_call_ended=hook
        )
        call = await gateway.make_call("+15551234567")
        await gateway.hangup_call(call.id)

        assert ended == [(call.id, CallStatus.COMPLETED)]


# ================================================================
# Receptionist-owned inbound policy
# ================================================================

class _StubRouting:
    def __init__(self, decision):
        self._decision = decision

    async def evaluate(self, caller_number, dnis):
        return self._decision


class TestInboundPolicy:
    def _gateway(self) -> AIPSTNGateway:
        settings = Settings(max_concurrent_calls=4)
        settings.receptionist = ReceptionistSettings(enabled=False)
        return AIPSTNGateway(settings=settings)

    async def test_inbound_call_answered_and_tracked(self):
        gateway = self._gateway()
        receptionist = ReceptionistService(gateway)

        await receptionist.on_inbound_call(
            "sip:+16135550100@pstn", "sip:+15551234567@gw", "leg_in1"
        )

        calls = list(gateway.call_manager.active_calls.values())
        assert len(calls) == 1
        call = calls[0]
        assert call.direction == "inbound"
        assert call.remote_number == "+16135550100"
        assert call.status == CallStatus.CONNECTED
        assert gateway.call_manager.legs_for_call(call.id) == ["leg_in1"]

    async def test_reject_rule_declines_before_answer(self):
        gateway = self._gateway()
        decision = RoutingDecision(
            action=RoutingAction(type=RoutingActionType.REJECT),
            matched_rule_id="rule_x",
            matched_rule_name="block",
            reason="matched rule 'block'",
        )
        receptionist = ReceptionistService(gateway, routing=_StubRouting(decision))

        await receptionist.on_inbound_call(
            "sip:+18005550100@pstn", "sip:+15551234567@gw", "leg_in2"
        )

        assert gateway.call_manager.active_calls == {}


# ================================================================
# REST routes on the shared data layer (real SQLite)
# ================================================================

@pytest.fixture
async def client(monkeypatch):
    monkeypatch.setattr(get_settings(), "api_token", SecretStr(""))

    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)

    async def _get_db():
        async with factory() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise

    main.app.dependency_overrides[get_db] = _get_db
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        c.db_factory = factory
        yield c
    main.app.dependency_overrides.pop(get_db, None)
    await engine.dispose()


FLOW_PAYLOAD = {
    "name": "Acme Main Line",
    "phone_number": "+18005551234",
    "description": "Main IVR",
    "steps": [
        {
            "id": "step1",
            "description": "Press 2 for billing",
            "action": "dtmf",
            "action_value": "2",
        }
    ],
    "tags": ["test"],
}


class TestCallFlowRoutes:
    async def test_crud_round_trip(self, client):
        resp = await client.post("/api/call-flows/", json=FLOW_PAYLOAD)
        assert resp.status_code == 200, resp.text
        flow_id = resp.json()["id"]
        assert flow_id == "acme-main-line"

        resp = await client.post("/api/call-flows/", json=FLOW_PAYLOAD)
        assert resp.status_code == 409

        resp = await client.get("/api/call-flows/")
        assert [f["id"] for f in resp.json()] == [flow_id]

        resp = await client.get(f"/api/call-flows/{flow_id}")
        assert resp.json()["steps"][0]["action_value"] == "2"

        resp = await client.get("/api/call-flows/by-number/+18005551234")
        assert resp.json()["id"] == flow_id

        resp = await client.put(
            f"/api/call-flows/{flow_id}", json={"notes": "updated"}
        )
        assert resp.json()["notes"] == "updated"

        resp = await client.delete(f"/api/call-flows/{flow_id}")
        assert resp.json()["status"] == "deleted"

        resp = await client.get(f"/api/call-flows/{flow_id}")
        assert resp.status_code == 404


class TestCallHistoryRoutes:
    async def test_history_and_record(self, client):
        resp = await client.get("/api/calls/history")
        assert resp.status_code == 200
        assert resp.json() == []

        async with client.db_factory() as session:
            session.add(CallRecord(
                id="call_hist1",
                direction="outbound",
                remote_number="+18005551234",
                status="completed",
                mode="hold_slayer",
                intent="dispute charge",
                duration=120,
                hold_time=90,
            ))
            await session.commit()

        resp = await client.get("/api/calls/history")
        assert [r["id"] for r in resp.json()] == ["call_hist1"]

        resp = await client.get("/api/calls/history?number=%2B18005551234")
        assert len(resp.json()) == 1

        resp = await client.get("/api/calls/call_hist1/record")
        assert resp.json()["intent"] == "dispute charge"

        resp = await client.get("/api/calls/call_missing/record")
        assert resp.status_code == 404

        resp = await client.get("/api/calls/call_hist1/transcript")
        assert resp.json() == []