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]