bmc_hub/tests/test_telefoni_call_logging.py

import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from starlette.requests import Request

from app.modules.telefoni.backend import router as telefoni_router
from app.modules.telefoni.backend.router import _is_internal_request_target
from app.modules.telefoni.backend.schemas import TelefoniCallLinkUpdate
from app.modules.telefoni.backend.service import TelefoniService


def _telefoni_javascript() -> str:
    return (Path(__file__).resolve().parents[1] / "static/js/telefoni.js").read_text(encoding="utf-8")


def _request_for_host(host: str) -> Request:
    return Request({"type": "http", "method": "POST", "path": "/", "headers": [], "server": (host, 8001)})


def test_terminate_call_creates_placeholder_row_when_missing(monkeypatch):
    calls = []

    def fake_execute_query(query, params=(), *args, **kwargs):
        calls.append((query, params))
        return [{"id": 1}]

    def fake_upsert_call(*, callid, user_id, direction, ekstern_nummer, intern_extension, kontakt_id, raw_payload, started_at):
        return {
            "id": 42,
            "callid": callid,
            "direction": direction,
            "started_at": started_at,
        }

    monkeypatch.setattr("app.modules.telefoni.backend.service.execute_query", fake_execute_query)
    monkeypatch.setattr(TelefoniService, "upsert_call", staticmethod(fake_upsert_call))

    result = TelefoniService.terminate_call("call-123", 120)

    assert result is True
    assert any("INSERT INTO telefoni_opkald" in query for query, _ in calls)


def test_unknown_caller_quick_contact_can_select_and_save_company():
    source = _telefoni_javascript()

    assert 'data-role="quick-company-search"' in source
    assert "async function searchCustomers" in source
    assert "company_id: companyId > 0 ? companyId : null" in source
    assert "body: JSON.stringify({ customer_id: companyId, is_primary: true })" in source


def test_recent_open_case_can_be_linked_to_call_from_popup():
    source = _telefoni_javascript()

    assert 'data-action="link-recent-case"' in source
    assert "await patchCallCase(caseId)" in source
    assert "Link til sag" in source


def test_click_to_call_recognizes_local_and_internal_hub_targets():
    assert _is_internal_request_target(_request_for_host("localhost")) is True
    assert _is_internal_request_target(_request_for_host("127.0.0.1")) is True
    assert _is_internal_request_target(_request_for_host("172.16.31.183")) is True


def test_click_to_call_rejects_external_hub_target_fallback():
    assert _is_internal_request_target(_request_for_host("example.com")) is False
    assert _is_internal_request_target(_request_for_host("172.16.30.183")) is False


def test_legacy_mission_call_is_promoted_when_linked_to_case(monkeypatch):
    executed = []

    def fake_single(query, params=None):
        if "WHERE id = %s" in query:
            return None
        return None

    def fake_query(query, params=None):
        executed.append((query, params))
        if "FROM mission_call_state" in query:
            return [{"id": 77}]
        if "UPDATE telefoni_opkald" in query:
            return [{"id": 77, "sag_id": 42}]
        return []

    monkeypatch.setattr(telefoni_router, "execute_query_single", fake_single)
    monkeypatch.setattr(telefoni_router, "execute_query", fake_query)

    result = asyncio.run(telefoni_router.update_call_links(
        -1,
        TelefoniCallLinkUpdate(sag_id=42, callid="legacy-123", source="legacy_mission"),
    ))

    assert result["sag_id"] == 42
    assert any("INSERT INTO telefoni_opkald" in query for query, _ in executed)
    assert any(params == (42, 77) for query, params in executed if "UPDATE telefoni_opkald" in query)


def test_telefoni_ui_sends_legacy_source_identity_when_linking():
    template = Path("app/modules/telefoni/templates/log.html").read_text(encoding="utf-8")

    assert 'data-callid="{{ call.callid or \'\' }}"' in template
    assert "callid: call?.callid || null" in template
    assert "source: call?.source || null" in template


def test_stale_termination_duration_is_not_derived_from_current_time(monkeypatch):
    queries = []
    monkeypatch.setattr(
        "app.modules.telefoni.backend.service.execute_query",
        lambda query, params=(): queries.append(query) or [{"id": 1}],
    )

    assert TelefoniService.terminate_call("stale-call", None) is True
    assert "INTERVAL '12 hours'" in queries[0]


def test_reused_callids_only_conflict_while_the_call_is_active():
    service = Path("app/modules/telefoni/backend/service.py").read_text(encoding="utf-8")
    migration = Path("migrations/1047_telefoni_reused_callids.sql").read_text(encoding="utf-8")

    assert "ON CONFLICT (callid) WHERE ended_at IS NULL" in service
    assert "WHERE callid = %s AND ended_at IS NULL" in service
    assert "DROP INDEX IF EXISTS ux_telefoni_opkald_callid" in migration
    assert "ux_telefoni_opkald_active_callid" in migration


def test_forwarded_call_uses_yealink_displayed_caller_number(monkeypatch):
    captured = {}

    monkeypatch.setattr(telefoni_router, "_validate_yealink_request", lambda *_: None)
    monkeypatch.setattr(TelefoniService, "find_user_by_extension", staticmethod(lambda _extension: []))
    monkeypatch.setattr(TelefoniService, "find_contact_by_phone", staticmethod(lambda _number: None))
    monkeypatch.setattr(
        TelefoniService,
        "upsert_call",
        staticmethod(lambda **kwargs: captured.update(kwargs) or {"id": 1}),
    )
    monkeypatch.setattr(telefoni_router, "log_system_event", lambda **_kwargs: None)

    request = Request({"type": "http", "method": "GET", "path": "/", "headers": [], "client": ("127.0.0.1", 8001)})
    asyncio.run(telefoni_router.yealink_established(
        request,
        callid="116",
        call_id=None,
        caller=None,
        callee=None,
        remote="sip:206@pbx.sipserver.dk",
        local="sip:210_99773@pbx.sipserver.dk",
        active_user=None,
        called_number=None,
        display_remote="+45 49 13 27 73",
        caller_id=None,
        token=None,
    ))

    assert captured["direction"] == "inbound"
    assert captured["ekstern_nummer"] == "+4549132773"


def test_forwarded_call_does_not_store_forwarding_sip_uri(monkeypatch):
    captured = {}

    monkeypatch.setattr(telefoni_router, "_validate_yealink_request", lambda *_: None)
    monkeypatch.setattr(TelefoniService, "find_user_by_extension", staticmethod(lambda _extension: []))
    monkeypatch.setattr(TelefoniService, "find_contact_by_phone", staticmethod(lambda _number: None))
    monkeypatch.setattr(
        TelefoniService,
        "upsert_call",
        staticmethod(lambda **kwargs: captured.update(kwargs) or {"id": 1}),
    )
    monkeypatch.setattr(telefoni_router, "log_system_event", lambda **_kwargs: None)

    request = Request({"type": "http", "method": "GET", "path": "/", "headers": [], "client": ("127.0.0.1", 8001)})
    asyncio.run(telefoni_router.yealink_established(
        request,
        callid="117",
        call_id=None,
        caller=None,
        callee=None,
        remote="sip:206@pbx.sipserver.dk",
        local="sip:210_99773@pbx.sipserver.dk",
        active_user=None,
        called_number=None,
        display_remote=None,
        caller_id=None,
        token=None,
    ))

    assert captured["ekstern_nummer"] is None


def test_callid_recovery_uses_system_log_without_overwriting_existing_current_calls():
    migration = Path("migrations/1047_telefoni_reused_callids.sql").read_text(encoding="utf-8")

    assert "system_event_log established" in migration
    assert "system_log_recovery" in migration
    assert "current_call.started_at BETWEEN" in migration


def test_case_click_to_call_contract_tracks_case_contact_and_time():
    schema = Path("app/modules/telefoni/backend/schemas.py").read_text(encoding="utf-8")
    router = Path("app/modules/telefoni/backend/router.py").read_text(encoding="utf-8")
    service = Path("app/modules/telefoni/backend/service.py").read_text(encoding="utf-8")

    assert "sag_id: Optional[int]" in schema
    assert "contact_id: Optional[int]" in schema
    assert 'pending_callid = f"click-to-call:' in router
    assert "_register_completed_call_time(resolved_callid)" in router
    assert "INSERT INTO tmodule_times" in router
    # psycopg uses % for placeholders, so the literal SQL LIKE wildcard must
    # be escaped. Otherwise the established Yealink callback returns HTTP 500.
    assert "callid LIKE 'click-to-call:%%'" in service