mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 06:43:12 +00:00
5981776178
* fix(3cx): refresh the OAuth token before it expires The worker kept one 3CX access token for the life of the process, so the 8am queue update failed with 401 after the one-hour token lifetime. * fix(3cx): ignore a client secret this process already replaced 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. * fix(3cx): adopt a new client secret only after login succeeds A candidate secret is tried before it replaces the current one, so a revoked secret cannot stick and a later revert to a working secret still takes effect. A failed re-login after 401 returns the original API response.
400 lines
12 KiB
Python
400 lines
12 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")
|
|
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
|