call-center/tests/test_postgres_readiness.py

import importlib
import os
import time

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from services.shared.db import get_session
from services.shared.sql_models import Interaction, VoiceAISessionRow, VoiceEventRow, WhatsAppMessageRow, WhatsAppThreadRow


pytestmark = pytest.mark.skipif(
    not os.getenv("DATABASE_URL", "").startswith("postgresql"),
    reason="PostgreSQL readiness smoke requires PostgreSQL",
)


def _admin_headers() -> dict[str, str]:
    return {"X-User": "admin", "X-Role": "admin"}


def test_postgres_readiness_smoke(monkeypatch):
    interaction_module = importlib.import_module("services.interaction_service.app")
    whatsapp_module = importlib.import_module("services.whatsapp_adapter_service.app")
    voice_module = importlib.import_module("services.voice_adapter_service.app")
    runtime_module = importlib.import_module("services.ai_voice_runtime_service.app")

    monkeypatch.setattr(
        runtime_module,
        "_orchestrator_request",
        lambda method, path, *, payload=None, timeout=10.0: {
            "session_id": "ais_pg_smoke",
            "language": "ru",
            "greeting_text": "Здравствуйте, это AI-оператор.",
            "disclosure_required": True,
        },
    )

    interaction_client = TestClient(interaction_module.app)
    whatsapp_client = TestClient(whatsapp_module.app)
    voice_client = TestClient(voice_module.app)
    runtime_client = TestClient(runtime_module.app)

    assert interaction_client.get("/health").status_code == 200
    assert whatsapp_client.get("/health").status_code == 200
    assert voice_client.get("/health").status_code == 200
    assert runtime_client.get("/health").status_code == 200

    created_interaction = interaction_client.post(
        "/interactions",
        json={
            "channel": "voice",
            "subject": "Postgres smoke interaction",
            "customer_id": None,
            "queue_id": "q_pg_smoke",
            "priority": 2,
        },
        headers=_admin_headers(),
    )
    assert created_interaction.status_code == 200
    interaction_id = created_interaction.json()["interaction_id"]

    fetched_interaction = interaction_client.get(f"/interactions/{interaction_id}")
    assert fetched_interaction.status_code == 200
    assert fetched_interaction.json()["interaction_id"] == interaction_id

    created_whatsapp = whatsapp_client.post(
        "/integrations/whatsapp/webhook",
        json={
            "chat_id": "pg_smoke_chat",
            "text": "Привет из PostgreSQL smoke",
            "external_message_id": "pg_smoke_ext_1",
            "whatsapp_user_id": "wa_pg_smoke_user",
            "phone_number": "+77015550077",
            "display_name": "PG Smoke",
            "payload": {"provider": "manual"},
        },
    )
    assert created_whatsapp.status_code == 200
    thread_id = created_whatsapp.json()["thread_id"]

    listed_threads = whatsapp_client.get("/integrations/whatsapp/threads", headers=_admin_headers())
    assert listed_threads.status_code == 200
    assert any(item["thread_id"] == thread_id for item in listed_threads.json())

    created_voice_event = voice_client.post(
        "/integrations/voice/events",
        headers=_admin_headers(),
        json={
            "event_type": "call.started",
            "call_id": "pg_smoke_call",
            "interaction_id": interaction_id,
            "source_event_id": "pg_smoke_call_started",
            "payload": {"source": "postgres-smoke"},
        },
    )
    assert created_voice_event.status_code == 200

    listed_voice_events = voice_client.get("/integrations/voice/events", headers=_admin_headers())
    assert listed_voice_events.status_code == 200
    assert any(item["source_event_id"] == "pg_smoke_call_started" for item in listed_voice_events.json())

    created_voice_session = runtime_client.post(
        "/internal/voice-ai/sessions",
        headers=_admin_headers(),
        json={
            "call_id": "pg_smoke_voice_call",
            "linked_id": "pg_smoke_linked",
            "interaction_id": interaction_id,
            "queue_id": "q_pg_smoke",
            "caller_number": "+77010000077",
            "caller_name": "PG Smoke Caller",
            "agent_profile": "voice_support",
            "language_hint": "ru",
            "handoff_queue_id": "q_pg_smoke",
            "metadata": {"queue_code": "pg_smoke"},
        },
    )
    assert created_voice_session.status_code == 200
    voice_session_id = created_voice_session.json()["voice_session_id"]

    deadline = time.time() + 2.0
    while time.time() < deadline:
        session = get_session()
        try:
            voice_session = session.execute(
                select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == voice_session_id)
            ).scalar_one_or_none()
            if voice_session and voice_session.ai_session_id == "ais_pg_smoke":
                break
        finally:
            session.close()
        time.sleep(0.05)
    else:
        raise AssertionError("voice AI session did not finish async startup")

    session = get_session()
    try:
        stored_interaction = session.execute(
            select(Interaction).where(Interaction.interaction_id == interaction_id)
        ).scalar_one_or_none()
        stored_thread = session.execute(
            select(WhatsAppThreadRow).where(WhatsAppThreadRow.thread_id == thread_id)
        ).scalar_one_or_none()
        stored_message = session.execute(
            select(WhatsAppMessageRow).where(WhatsAppMessageRow.thread_id == thread_id)
        ).scalar_one_or_none()
        stored_event = session.execute(
            select(VoiceEventRow).where(VoiceEventRow.source_event_id == "pg_smoke_call_started")
        ).scalar_one_or_none()
        stored_voice_session = session.execute(
            select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == voice_session_id)
        ).scalar_one_or_none()
        assert stored_interaction is not None
        assert stored_thread is not None
        assert stored_message is not None
        assert stored_event is not None
        assert stored_voice_session is not None
        assert stored_voice_session.ai_session_id == "ais_pg_smoke"
    finally:
        session.close()