mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 14:53:12 +00:00
Adam Moussa
dccf366dcf
A slower caller still holding the pre-rotation secret could write it back over the new one. The swap now happens under the auth lock, and a retired secret is dropped.
323 lines
9.4 KiB
Python
323 lines
9.4 KiB
Python
"""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")
|
|
stale = tcx.oauth_client("test.3cx.us", "cid", "old-secret")
|
|
assert first is second is stale
|
|
assert stale.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_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
|