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/IVRs/Pbx.GetByNumber(number='800')",
        json={
            "Id": 3,
            "Number": "800",
            "Options": [{"Digit": "0", "Route": {"To": "Extension", "Number": "101"}}],
            "TimeoutForward": {"To": "Extension", "Number": "102"},
        },
        status=200,
    )
    patched = responses.add(responses.PATCH, f"{BASE}/xapi/v1/IVRs(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

    key0 = {"To": "Queue", "Number": "802"}
    status = client.set_ivr_routes(3, key0_route=key0, timeout_route=key0)
    assert status == 200

    import json

    body = json.loads(patched.calls[0].request.body)
    assert body["Options"][0]["Digit"] == "0"
    assert body["Options"][0]["Route"]["Number"] == "802"
    assert body["TimeoutForward"]["Number"] == "802"


def test_extract_ivr_routes_pulls_key0_and_timeout():
    ivr = {
        "Options": [
            {"Digit": "1", "Route": {"Number": "201"}},
            {"Digit": "0", "Route": {"Number": "101"}},
        ],
        "TimeoutForward": {"Number": "102"},
    }
    routes = ThreeCXClient.extract_ivr_routes(ivr)
    assert routes["key0"] == {"Number": "101"}
    assert routes["timeout"] == {"Number": "102"}


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


def test_route_to_extension():
    assert ThreeCXClient.route_to_extension({"Number": "802"}) == "802"
    assert ThreeCXClient.route_to_extension(None) is None
    assert ThreeCXClient.route_to_extension({}) is None