Senad Uka
de5a2abdeb
- GsmGatewaySmsProvider: GoIP-style HTTP SMS API, per-tenant SIM line from phone_numbers.gateway_port; model-specific bits isolated for easy adaptation (§17.3); failure never breaks the calling flow (logged to sms_log) - Asterisk config generator now emits [gw-line-N] SIP trunk endpoints per assigned SIM, routed straight into the tenant's inbound context (1 SIM = 1 salon §5.1); deterministic per-SIM SIP secrets - docs/HARDWARE_TESTING.md: step-by-step live runbook for the joint session (gateway registration, inbound path, SMS via gateway, operator forwarding codes per m:tel/BH Telecom/HT Eronet, e2e acceptance, rollback) - 5 tests against a mocked gateway API (forwarding-code generator and SIM/port admin UI shipped in M3) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
"""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
|