gogo-telefon/tests/test_gateway_sms.py

"""M5: GSM gateway SMS provider against a mocked gateway HTTP API + trunk config."""

import httpx
import pytest
from fastapi import FastAPI, Request
from sqlalchemy import select

from gogo.models import PhoneNumber, SmsLog
from gogo.sms.base import send_sms, set_sms_provider
from gogo.sms.gateway import GatewayError, GsmGatewaySmsProvider

# -- fake gateway (GoIP-style HTTP API) ------------------------------------------

fake_gateway = FastAPI()
received: list[dict] = []


@fake_gateway.get("/default/en_US/send.html")
async def send_html(request: Request):
    received.append(dict(request.query_params))
    return "Sending"


@fake_gateway.get("/fail/default/en_US/send.html")
async def send_fail(request: Request):
    return "ERROR: no signal"


@pytest.fixture
def gateway_env(monkeypatch):
    received.clear()
    monkeypatch.setenv("GOGO_GSM_GATEWAY_URL", "http://gw.test")
    monkeypatch.setenv("GOGO_GSM_GATEWAY_USER", "admin")
    monkeypatch.setenv("GOGO_GSM_GATEWAY_PASSWORD", "gwpass")
    from gogo.config import get_settings

    get_settings.cache_clear()

    def factory(timeout):
        return httpx.AsyncClient(
            transport=httpx.ASGITransport(app=fake_gateway),
            base_url="http://gw.test",
            timeout=timeout,
        )

    monkeypatch.setattr(GsmGatewaySmsProvider, "client_factory", staticmethod(factory))
    yield
    get_settings.cache_clear()


async def assign_sim(session, tenant, port=3, msisdn="+38765000111"):
    session.add(PhoneNumber(tenant_id=tenant.id, msisdn=msisdn, gateway_port=port))
    await session.commit()


async def test_sms_sent_through_tenant_line(session, tenant, gateway_env):
    await assign_sim(session, tenant, port=3)
    provider = GsmGatewaySmsProvider()
    await provider.send(tenant.id, "+38761222333", "Potvrđen termin: sutra u 17h")

    assert len(received) == 1
    params = received[0]
    assert params["u"] == "admin"
    assert params["p"] == "gwpass"
    assert params["l"] == "3"  # the tenant's SIM line
    assert params["n"] == "38761222333"  # no leading +
    assert "Potvrđen termin" in params["m"]


async def test_no_sim_assigned_raises(session, tenant, gateway_env):
    provider = GsmGatewaySmsProvider()
    with pytest.raises(GatewayError, match="no SIM"):
        await provider.send(tenant.id, "+38761222333", "test")


async def test_gateway_error_body_raises(session, tenant, gateway_env, monkeypatch):
    await assign_sim(session, tenant)

    def factory(timeout):
        return httpx.AsyncClient(
            transport=httpx.ASGITransport(app=fake_gateway),
            base_url="http://gw.test/fail",
            timeout=timeout,
        )

    monkeypatch.setenv("GOGO_GSM_GATEWAY_URL", "http://gw.test/fail")
    from gogo.config import get_settings

    get_settings.cache_clear()
    monkeypatch.setattr(GsmGatewaySmsProvider, "client_factory", staticmethod(factory))

    provider = GsmGatewaySmsProvider()
    with pytest.raises(GatewayError, match="rejected"):
        await provider.send(tenant.id, "+38761222333", "test")


async def test_send_sms_wrapper_logs_failure_without_raising(
    session, tenant, gateway_env
):
    """SMS failure must never break the calling flow (§9) — logged as failed."""
    set_sms_provider(GsmGatewaySmsProvider())  # no SIM assigned → will fail
    try:
        ok = await send_sms(session, tenant.id, "+38761222333", "test", kind="confirmation")
        await session.commit()
    finally:
        set_sms_provider(None)
    assert ok is False
    row = (await session.execute(select(SmsLog))).scalar_one()
    assert row.status == "failed"
    assert "no SIM" in row.error


async def test_asterisk_config_includes_gateway_line(session, tenant, tmp_path):
    from gogo.telephony.asterisk import gateway_line_password, generate

    await assign_sim(session, tenant, port=2, msisdn="+38765000222")
    await generate(tmp_path)
    pjsip = (tmp_path / "pjsip.conf").read_text()
    assert "[gw-line-2]" in pjsip
    assert "context=gogo-inbound-salon-merima" in pjsip
    assert f"password={gateway_line_password('+38765000222')}" in pjsip