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}


def _token_posts():
    return [c for c in responses.calls if c.request.url.endswith("/connect/token")]


@responses.activate
def test_oauth_client_reuses_process_cache():
    from shared import three_cx_client as tcx

    tcx._oauth_clients.clear()
    _stub_oauth()
    first = tcx.oauth_client("test.3cx.us", "cid", "secret")
    second = tcx.oauth_client("test.3cx.us", "cid", "secret")
    assert first is second
    assert len(_token_posts()) == 1
    tcx._oauth_clients.clear()


@responses.activate
def test_oauth_client_refreshes_expired_token():
    from shared import three_cx_client as tcx

    tcx._oauth_clients.clear()
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-1", "expires_in": 3600},
        status=200,
    )
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-2", "expires_in": 3600},
        status=200,
    )
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='801')",
        json={"Id": 83},
        status=200,
    )
    client = tcx.oauth_client("test.3cx.us", "cid", "secret")
    client._token_expires_at = 0
    queue = client.get_queue("801")
    assert queue["Id"] == 83
    assert client.session.headers["Authorization"] == "Bearer tok-2"
    assert len(_token_posts()) == 2
    tcx._oauth_clients.clear()


@responses.activate
def test_oauth_client_reauths_when_client_secret_changes():
    from shared import three_cx_client as tcx

    tcx._oauth_clients.clear()
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-old", "expires_in": 3600},
        status=200,
    )
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-new", "expires_in": 3600},
        status=200,
    )
    first = tcx.oauth_client("test.3cx.us", "cid", "old-secret")
    second = tcx.oauth_client("test.3cx.us", "cid", "new-secret")
    assert first is second
    assert second.session.headers["Authorization"] == "Bearer tok-new"
    posts = _token_posts()
    assert len(posts) == 2
    assert "client_secret=new-secret" in posts[1].request.body
    tcx._oauth_clients.clear()


@responses.activate
def test_oauth_client_accepts_a_reverted_client_secret():
    from shared import three_cx_client as tcx

    tcx._oauth_clients.clear()
    for token in ("tok-a", "tok-b", "tok-a-again"):
        responses.add(
            responses.POST,
            f"{BASE}/connect/token",
            json={"access_token": token, "expires_in": 3600},
            status=200,
        )
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='801')",
        json={"Id": 83},
        status=200,
    )
    client = tcx.oauth_client("test.3cx.us", "cid", "secret-a")
    tcx.oauth_client("test.3cx.us", "cid", "secret-b")
    reverted = tcx.oauth_client("test.3cx.us", "cid", "secret-a")
    assert reverted is client
    assert reverted.session.headers["Authorization"] == "Bearer tok-a-again"
    assert reverted.get_queue("801")["Id"] == 83
    posts = _token_posts()
    assert len(posts) == 3
    assert "client_secret=secret-a" in posts[2].request.body
    tcx._oauth_clients.clear()


@responses.activate
def test_oauth_client_keeps_current_secret_when_candidate_login_fails():
    from shared import three_cx_client as tcx

    tcx._oauth_clients.clear()
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-current", "expires_in": 3600},
        status=200,
    )
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-new", "expires_in": 3600},
        status=200,
    )
    responses.add(responses.POST, f"{BASE}/connect/token", status=401)
    client = tcx.oauth_client("test.3cx.us", "cid", "current-secret")
    tcx.oauth_client("test.3cx.us", "cid", "new-secret")
    kept = tcx.oauth_client("test.3cx.us", "cid", "revoked-secret")
    assert kept is client
    assert kept.session.headers["Authorization"] == "Bearer tok-new"
    assert kept._client_secret == "new-secret"
    tcx._oauth_clients.clear()


@responses.activate
def test_oauth_request_retries_once_after_401():
    _stub_oauth()
    responses.add(
        responses.POST,
        f"{BASE}/connect/token",
        json={"access_token": "tok-refreshed", "expires_in": 3600},
        status=200,
    )
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='801')",
        status=401,
    )
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='801')",
        json={"Id": 83},
        status=200,
    )
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    assert client.get_queue("801")["Id"] == 83
    assert client.session.headers["Authorization"] == "Bearer tok-refreshed"
    assert len(_token_posts()) == 2


@responses.activate
def test_oauth_401_returns_original_response_when_relogin_fails():
    import pytest
    from requests import HTTPError

    _stub_oauth()
    responses.add(responses.POST, f"{BASE}/connect/token", status=401)
    responses.add(
        responses.GET,
        f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='801')",
        status=401,
    )
    client = ThreeCXClient(
        domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
    )
    with pytest.raises(HTTPError) as raised:
        client.get_queue("801")
    assert "Queues/Pbx.GetByNumber" in str(raised.value)
    assert raised.value.response.status_code == 401