afterhours-shift-manager/tests/shared/test_three_cx_client.py

"""Tests for shared.three_cx_client — auth flows and XAPI calls (HTTP mocked)."""

import responses

from shared.three_cx_client import ThreeCXClient

BASE = "https://test.3cx.us"


def _stub_oauth():
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-oauth"},
        status=200,
    )


@responses.activate
def test_oauth_authentication_sets_bearer_header():
    _stub_oauth()
    client = ThreeCXClient(
        domain="test.3cx.us",
        auth_mode="oauth",
        client_id="cid",
        client_secret="secret",
    )
    assert client.session.headers["Authorization"] == "Bearer tok-oauth"


@responses.activate
def test_user_authentication_extracts_nested_token():
    responses.add(
        responses.POST,
        f"{BASE}/webclient/api/Login/GetAccessToken",
        json={"Token": {"access_token": "tok-user"}},
        status=200,
    )
    client = ThreeCXClient(domain="test.3cx.us", username="u", password="p")
    assert client.session.headers["Authorization"] == "Bearer tok-user"


@responses.activate
def test_user_authentication_missing_token_raises():
    import pytest

    responses.add(
        responses.POST,
        f"{BASE}/webclient/api/Login/GetAccessToken",
        json={"nope": True},
        status=200,
    )
    with pytest.raises(ValueError):
        ThreeCXClient(domain="test.3cx.us", username="u", password="p")


@responses.activate
def test_get_group_members_resolves_group_then_members():
    _stub_oauth()
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Groups",
        json={"value": [{"Id": 5, "Name": "DEFAULT"}]},
        status=200,
    )
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Groups(5)/Members",
        json={"value": [{"Number": "114", "MemberName": "Alice", "Type": "Extension"}]},
        status=200,
    )
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    members = client.get_group_members("DEFAULT")
    assert members == [{"Number": "114", "MemberName": "Alice", "Type": "Extension"}]


@responses.activate
def test_get_group_members_unknown_group_returns_empty():
    _stub_oauth()
    responses.add(
        responses.GET, f"{BASE}/xapi/v1/Groups", json={"value": []}, status=200
    )
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    assert client.get_group_members("NOPE") == []


@responses.activate
def test_get_queue_and_update_queue_forwarding():
    _stub_oauth()
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='800')",
        json={"Id": 7, "Number": "800"},
        status=200,
    )
    patched = responses.add(responses.PATCH, f"{BASE}/xapi/v1/Queues(7)", status=200)
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    queue = client.get_queue("800")
    assert queue["Id"] == 7
    status = client.update_queue_forwarding(queue_id=7, closed="114", holiday="114")
    assert status == 200
    # The forwarding payload routes both closed and holiday to the extension.
    import json

    body = json.loads(patched.calls[0].request.body)
    assert body["OutOfOfficeRoute"]["Route"]["Number"] == "114"
    assert body["HolidaysRoute"]["Route"]["Number"] == "114"


@responses.activate
def test_set_queue_agents_replaces_membership():
    _stub_oauth()
    patched = responses.add(responses.PATCH, f"{BASE}/xapi/v1/Queues(7)", status=200)
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    status = client.set_queue_agents(7, ["114", "115"])
    assert status == 200

    import json

    body = json.loads(patched.calls[0].request.body)
    assert body["Agents"] == [{"Number": "114"}, {"Number": "115"}]


@responses.activate
def test_set_queue_agents_empty_clears_membership():
    _stub_oauth()
    patched = responses.add(responses.PATCH, f"{BASE}/xapi/v1/Queues(7)", status=200)
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    client.set_queue_agents(7, [])

    import json

    body = json.loads(patched.calls[0].request.body)
    assert body["Agents"] == []


@responses.activate
def test_get_ivr_and_set_ivr_routes():
    _stub_oauth()
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Receptionists/Pbx.GetByNumber(number='800')",
        json={
            "Id": 3,
            "Number": "800",
            "TimeoutForwardDN": "801",
            "TimeoutForwardType": "Queue",
            "TimeoutForwardPeerType": "Queue",
        },
        status=200,
    )
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Receptionists(3)/Forwards",
        json={
            "value": [
                {
                    "Input": "0",
                    "ForwardType": "Queue",
                    "PeerType": "Queue",
                    "ForwardDN": "801",
                    "Id": 16,
                }
            ]
        },
        status=200,
    )
    patched = responses.add(
        responses.PATCH, f"{BASE}/xapi/v1/Receptionists(3)", status=200
    )
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    ivr = client.get_ivr("800")
    assert ivr["Id"] == 3
    assert ivr["Forwards"][0]["ForwardDN"] == "801"

    status = client.set_ivr_routes(3, key0_dn="802", timeout_dn="802")
    assert status == 200

    import json

    body = json.loads(patched.calls[-1].request.body)
    key0 = next(f for f in body["Forwards"] if f["Input"] == "0")
    assert key0["ForwardDN"] == "802"
    assert key0["ForwardType"] == "Queue"
    assert body["TimeoutForwardDN"] == "802"
    assert body["TimeoutForwardType"] == "Queue"


def test_extract_ivr_routes_pulls_key0_and_timeout():
    ivr = {
        "Forwards": [
            {"Input": "1", "ForwardDN": "201"},
            {"Input": "0", "ForwardDN": "101"},
        ],
        "TimeoutForwardDN": "102",
    }
    routes = ThreeCXClient.extract_ivr_routes(ivr)
    assert routes["key0"] == "101"
    assert routes["timeout"] == "102"


def test_extract_ivr_routes_missing_key0_is_none():
    routes = ThreeCXClient.extract_ivr_routes(
        {"Forwards": [], "TimeoutForwardDN": None}
    )
    assert routes == {"key0": None, "timeout": None}