mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 03:23:12 +00:00
Fix holiday router 3CX IVR calls (Receptionists entity, not IVRs) (#124)
get_ivr/set_ivr_routes/extract_ivr_routes targeted a nonexistent IVRs entity set with an Options[].Route/TimeoutForward shape. The live 3CX IVR is the Receptionists entity: the no-input/timeout route is the scalar TimeoutForwardDN, and the key-0 route is a child of the Forwards collection (matched by Input=='0'), written via a parent deep-PATCH. Routes are now destination numbers. Caught by the live prod round-trip (get_ivr returned 405) before any holiday ran; rewritten and re-verified against the live PBX. v1.11.1.
This commit is contained in:
Adam Moussa
GitHub
parent
88e782c205
commit
54b585776b
6 changed files with 188 additions and 107 deletions
|
|
@ -10,6 +10,15 @@ fine and still supported.
|
|||
|
||||
---
|
||||
|
||||
## v1.11.1 — June 17, 2026
|
||||
|
||||
**Fix holiday call routing to 3CX.** The holiday router was calling the wrong 3CX
|
||||
API path for the auto-attendant (IVR 800), so activating a holiday wouldn't have
|
||||
rerouted calls to the holiday queue. Corrected to the right endpoint and verified
|
||||
end-to-end against the live phone system. No change to how you use the bot.
|
||||
|
||||
---
|
||||
|
||||
## v1.11.0 — June 15, 2026
|
||||
|
||||
**Holiday coverage and last-minute pickups.** Two related additions:
|
||||
|
|
|
|||
|
|
@ -52,11 +52,6 @@ def _holiday_extensions(holiday: dict) -> list[str]:
|
|||
return extensions or [FALLBACK_EXTENSION]
|
||||
|
||||
|
||||
def _queue_route(queue_number: str) -> dict:
|
||||
"""Build a 3CX Route object that forwards to the holiday queue extension."""
|
||||
return {"To": "Queue", "Number": str(queue_number), "External": ""}
|
||||
|
||||
|
||||
def _activate(schedule: ShiftSchedule, date: str) -> dict:
|
||||
holiday = schedule.get_holiday(date)
|
||||
if holiday is None:
|
||||
|
|
@ -78,9 +73,9 @@ def _activate(schedule: ShiftSchedule, date: str) -> dict:
|
|||
ivr = client.get_ivr(ivr_number)
|
||||
ivr_id = ivr["Id"]
|
||||
current = client.extract_ivr_routes(ivr)
|
||||
already_queue = client.route_to_extension(current.get("key0")) == str(
|
||||
queue_number
|
||||
) and client.route_to_extension(current.get("timeout")) == str(queue_number)
|
||||
already_queue = str(current.get("key0")) == str(queue_number) and str(
|
||||
current.get("timeout")
|
||||
) == str(queue_number)
|
||||
if already_queue:
|
||||
logger.info(
|
||||
"IVR %s already points at queue %s — not re-capturing routes",
|
||||
|
|
@ -94,8 +89,7 @@ def _activate(schedule: ShiftSchedule, date: str) -> dict:
|
|||
queue = client.get_queue(queue_number)
|
||||
client.set_queue_agents(queue["Id"], extensions)
|
||||
|
||||
route = _queue_route(queue_number)
|
||||
client.set_ivr_routes(ivr_id, key0_route=route, timeout_route=route)
|
||||
client.set_ivr_routes(ivr_id, key0_dn=queue_number, timeout_dn=queue_number)
|
||||
|
||||
schedule.set_holiday_activated(date, True)
|
||||
logger.info(
|
||||
|
|
@ -135,18 +129,18 @@ def _deactivate(schedule: ShiftSchedule, date: str) -> dict:
|
|||
|
||||
# Only restore a route if it currently points at the queue; otherwise leave
|
||||
# whatever destination it has now (it was changed outside this flow).
|
||||
def _restore(which: str) -> dict | None:
|
||||
live_route = live.get(which)
|
||||
if client.route_to_extension(live_route) == str(queue_number):
|
||||
# Restore the captured pre-holiday route; if it was lost, leave the
|
||||
# live route in place rather than blanking the IVR destination.
|
||||
return captured.get(which) or live_route
|
||||
return live_route
|
||||
def _restore(which: str) -> str | None:
|
||||
live_dn = live.get(which)
|
||||
if str(live_dn) == str(queue_number):
|
||||
# Restore the captured pre-holiday DN; if it was lost, keep the live
|
||||
# value rather than blanking the IVR destination.
|
||||
return captured.get(which) or live_dn
|
||||
return None # not pointing at the holiday queue — leave it untouched
|
||||
|
||||
client.set_ivr_routes(
|
||||
ivr_id,
|
||||
key0_route=_restore("key0"),
|
||||
timeout_route=_restore("timeout"),
|
||||
key0_dn=_restore("key0"),
|
||||
timeout_dn=_restore("timeout"),
|
||||
)
|
||||
|
||||
queue = client.get_queue(queue_number)
|
||||
|
|
|
|||
|
|
@ -148,50 +148,81 @@ class ThreeCXClient:
|
|||
# ── IVR (auto-attendant) routing ─────────────────────────────────────
|
||||
|
||||
def get_ivr(self, extension_number: str) -> dict:
|
||||
"""Fetch IVR (auto-attendant) config by extension number."""
|
||||
"""Fetch the IVR (3CX 'Receptionist') by number, with its Forwards.
|
||||
|
||||
In 3CX V20 the auto-attendant is the ``Receptionist`` entity. The
|
||||
no-input/timeout route is stored as scalar ``TimeoutForward*`` fields on
|
||||
the entity; the per-DTMF-key routes live in a separate ``Forwards``
|
||||
navigation collection, which is fetched and attached as ``Forwards``.
|
||||
"""
|
||||
resp = self.session.get(
|
||||
f"{self.base_url}/xapi/v1/IVRs/Pbx.GetByNumber(number='{extension_number}')",
|
||||
f"{self.base_url}/xapi/v1/Receptionists/Pbx.GetByNumber(number='{extension_number}')",
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
ivr = resp.json()
|
||||
fwd = self.session.get(
|
||||
f"{self.base_url}/xapi/v1/Receptionists({ivr['Id']})/Forwards",
|
||||
)
|
||||
fwd.raise_for_status()
|
||||
ivr["Forwards"] = fwd.json().get("value", [])
|
||||
return ivr
|
||||
|
||||
def set_ivr_routes(self, ivr_id: int, key0_route: dict, timeout_route: dict):
|
||||
"""Repoint an IVR's key-0 option and no-input/timeout route.
|
||||
def set_ivr_routes(
|
||||
self, ivr_id: int, key0_dn: str | None = None, timeout_dn: str | None = None
|
||||
):
|
||||
"""Repoint the IVR's key-0 DTMF route and/or no-input/timeout route.
|
||||
|
||||
``key0_route`` and ``timeout_route`` are 3CX Route objects (the shape
|
||||
captured from :meth:`extract_ivr_routes`). The key-0 option is matched
|
||||
within the IVR's ``Options`` list by ``Digit == "0"``; the timeout route
|
||||
is written to the IVR's ``TimeoutForward`` field.
|
||||
Targets are 3CX queues (the main 801 or holiday 802), given as
|
||||
destination numbers. The key-0 route is a child of the ``Forwards``
|
||||
collection (matched by ``Input == "0"``); 3CX requires the whole
|
||||
collection in a parent deep-PATCH, so the current Forwards are read, the
|
||||
key-0 ``ForwardDN`` is swapped, and the Receptionist is PATCHed together
|
||||
with the scalar ``TimeoutForward*`` fields. Either DN may be None to
|
||||
leave that route untouched.
|
||||
"""
|
||||
payload = {
|
||||
"Options": [{"Digit": "0", "Route": key0_route}],
|
||||
"TimeoutForward": timeout_route,
|
||||
}
|
||||
payload: dict = {}
|
||||
if key0_dn is not None:
|
||||
fwd = self.session.get(
|
||||
f"{self.base_url}/xapi/v1/Receptionists({ivr_id})/Forwards",
|
||||
)
|
||||
fwd.raise_for_status()
|
||||
forwards = [
|
||||
{k: v for k, v in f.items() if not k.startswith("@")}
|
||||
for f in fwd.json().get("value", [])
|
||||
]
|
||||
for f in forwards:
|
||||
if str(f.get("Input")) == "0":
|
||||
f["ForwardDN"] = str(key0_dn)
|
||||
f["ForwardType"] = "Queue"
|
||||
f["PeerType"] = "Queue"
|
||||
payload["Forwards"] = forwards
|
||||
if timeout_dn is not None:
|
||||
payload["TimeoutForwardDN"] = str(timeout_dn)
|
||||
payload["TimeoutForwardType"] = "Queue"
|
||||
payload["TimeoutForwardPeerType"] = "Queue"
|
||||
if not payload:
|
||||
return None
|
||||
resp = self.session.patch(
|
||||
f"{self.base_url}/xapi/v1/IVRs({ivr_id})",
|
||||
f"{self.base_url}/xapi/v1/Receptionists({ivr_id})",
|
||||
json=payload,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
logger.info("Set IVR %s key-0 and timeout routes", ivr_id)
|
||||
logger.info(
|
||||
"Set IVR (Receptionist) %s key-0=%s timeout=%s", ivr_id, key0_dn, timeout_dn
|
||||
)
|
||||
return resp.status_code
|
||||
|
||||
@staticmethod
|
||||
def extract_ivr_routes(ivr: dict) -> dict:
|
||||
"""Pull the key-0 and timeout Route objects out of an IVR payload.
|
||||
"""Return the key-0 and no-input/timeout destination numbers.
|
||||
|
||||
Returns ``{"key0": <route|None>, "timeout": <route|None>}``. The key-0
|
||||
route is the ``Route`` of the option whose ``Digit`` is ``"0"``.
|
||||
``{"key0": <dn|None>, "timeout": <dn|None>}`` — destination queue/ext
|
||||
numbers (strings), read from the ``Forwards`` entry with ``Input == "0"``
|
||||
and the scalar ``TimeoutForwardDN``.
|
||||
"""
|
||||
key0 = None
|
||||
for option in ivr.get("Options", []) or []:
|
||||
if str(option.get("Digit")) == "0":
|
||||
key0 = option.get("Route")
|
||||
for f in ivr.get("Forwards", []) or []:
|
||||
if str(f.get("Input")) == "0":
|
||||
key0 = f.get("ForwardDN")
|
||||
break
|
||||
return {"key0": key0, "timeout": ivr.get("TimeoutForward")}
|
||||
|
||||
@staticmethod
|
||||
def route_to_extension(route: dict | None) -> str | None:
|
||||
"""Return the destination extension Number of a Route, or None."""
|
||||
if not route:
|
||||
return None
|
||||
return route.get("Number")
|
||||
return {"key0": key0, "timeout": ivr.get("TimeoutForwardDN")}
|
||||
|
|
|
|||
|
|
@ -10,6 +10,15 @@ fine and still supported.
|
|||
|
||||
---
|
||||
|
||||
## v1.11.1 — June 17, 2026
|
||||
|
||||
**Fix holiday call routing to 3CX.** The holiday router was calling the wrong 3CX
|
||||
API path for the auto-attendant (IVR 800), so activating a holiday wouldn't have
|
||||
rerouted calls to the holiday queue. Corrected to the right endpoint and verified
|
||||
end-to-end against the live phone system. No change to how you use the bot.
|
||||
|
||||
---
|
||||
|
||||
## v1.11.0 — June 15, 2026
|
||||
|
||||
**Holiday coverage and last-minute pickups.** Two related additions:
|
||||
|
|
|
|||
|
|
@ -4,6 +4,10 @@ DynamoDB is moto-backed (via the shared ``schedule``/``seed`` fixtures); 3CX is
|
|||
mocked at the HTTP layer with ``responses`` so the real ThreeCXClient code runs.
|
||||
``get_secret`` is patched so ``_make_client`` resolves the 3CX domain to the test
|
||||
host without reaching AWS Secrets Manager.
|
||||
|
||||
The IVR is the 3CX ``Receptionist`` entity: the no-input/timeout route is a scalar
|
||||
``TimeoutForwardDN``; the key-0 route is a child of the ``Forwards`` collection
|
||||
(matched by ``Input == "0"``). Routes are represented as destination numbers (DNs).
|
||||
"""
|
||||
|
||||
import json
|
||||
|
|
@ -15,8 +19,10 @@ HOL = "2026-07-04"
|
|||
BASE = "https://test.3cx.us"
|
||||
QUEUE = "802"
|
||||
IVR = "800"
|
||||
ORIG_KEY0 = {"To": "Extension", "Number": "101", "External": ""}
|
||||
ORIG_TIMEOUT = {"To": "Extension", "Number": "102", "External": ""}
|
||||
ORIG_KEY0 = "801" # pre-holiday key-0 target (main dispatch queue)
|
||||
ORIG_TIMEOUT = "801" # pre-holiday no-input/timeout target
|
||||
IVR_ID = 7
|
||||
QUEUE_ID = 9
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -44,21 +50,39 @@ def _stub_oauth():
|
|||
)
|
||||
|
||||
|
||||
def _stub_ivr_get(key0=ORIG_KEY0, timeout=ORIG_TIMEOUT, ivr_id=7):
|
||||
def _stub_ivr_get(key0_dn=ORIG_KEY0, timeout_dn=ORIG_TIMEOUT, ivr_id=IVR_ID):
|
||||
"""Stub the two reads get_ivr performs: the Receptionist + its Forwards."""
|
||||
responses.add(
|
||||
responses.GET,
|
||||
f"{BASE}/xapi/v1/IVRs/Pbx.GetByNumber(number='{IVR}')",
|
||||
f"{BASE}/xapi/v1/Receptionists/Pbx.GetByNumber(number='{IVR}')",
|
||||
json={
|
||||
"Id": ivr_id,
|
||||
"Number": IVR,
|
||||
"Options": [{"Digit": "0", "Route": key0}],
|
||||
"TimeoutForward": timeout,
|
||||
"TimeoutForwardDN": timeout_dn,
|
||||
"TimeoutForwardType": "Queue",
|
||||
"TimeoutForwardPeerType": "Queue",
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
responses.add(
|
||||
responses.GET,
|
||||
f"{BASE}/xapi/v1/Receptionists({ivr_id})/Forwards",
|
||||
json={
|
||||
"value": [
|
||||
{
|
||||
"Input": "0",
|
||||
"ForwardType": "Queue",
|
||||
"PeerType": "Queue",
|
||||
"ForwardDN": key0_dn,
|
||||
"Id": 16,
|
||||
}
|
||||
]
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
|
||||
|
||||
def _stub_queue_get(queue_id=9):
|
||||
def _stub_queue_get(queue_id=QUEUE_ID):
|
||||
responses.add(
|
||||
responses.GET,
|
||||
f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='{QUEUE}')",
|
||||
|
|
@ -67,9 +91,9 @@ def _stub_queue_get(queue_id=9):
|
|||
)
|
||||
|
||||
|
||||
def _stub_patches(ivr_id=7, queue_id=9):
|
||||
def _stub_patches(ivr_id=IVR_ID, queue_id=QUEUE_ID):
|
||||
ivr_patch = responses.add(
|
||||
responses.PATCH, f"{BASE}/xapi/v1/IVRs({ivr_id})", status=200
|
||||
responses.PATCH, f"{BASE}/xapi/v1/Receptionists({ivr_id})", status=200
|
||||
)
|
||||
queue_patch = responses.add(
|
||||
responses.PATCH, f"{BASE}/xapi/v1/Queues({queue_id})", status=200
|
||||
|
|
@ -84,6 +108,10 @@ def _holiday_assignees(seed):
|
|||
}
|
||||
|
||||
|
||||
def _key0_forward(body):
|
||||
return next(f for f in body["Forwards"] if f["Input"] == "0")
|
||||
|
||||
|
||||
# ── activate ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
@ -102,21 +130,20 @@ def test_activate_captures_routes_sets_agents_repoints_ivr(
|
|||
assert result["action"] == "activate"
|
||||
assert set(result["agents"]) == {"114", "115"}
|
||||
|
||||
# Captured the live (pre-holiday) routes into CONFIG.
|
||||
# Captured the live (pre-holiday) DNs into CONFIG.
|
||||
captured = schedule.get_captured_ivr_routes()
|
||||
assert captured["key0"] == ORIG_KEY0
|
||||
assert captured["timeout"] == ORIG_TIMEOUT
|
||||
assert captured == {"key0": ORIG_KEY0, "timeout": ORIG_TIMEOUT}
|
||||
|
||||
# Queue agents set to assignees.
|
||||
qbody = json.loads(queue_patch.calls[0].request.body)
|
||||
assert {a["Number"] for a in qbody["Agents"]} == {"114", "115"}
|
||||
|
||||
# Both IVR routes repointed to the queue.
|
||||
ibody = json.loads(ivr_patch.calls[0].request.body)
|
||||
assert ibody["Options"][0]["Digit"] == "0"
|
||||
assert ibody["Options"][0]["Route"]["Number"] == QUEUE
|
||||
assert ibody["Options"][0]["Route"]["To"] == "Queue"
|
||||
assert ibody["TimeoutForward"]["Number"] == QUEUE
|
||||
# Both IVR routes repointed to the holiday queue.
|
||||
ibody = json.loads(ivr_patch.calls[-1].request.body)
|
||||
key0 = _key0_forward(ibody)
|
||||
assert key0["ForwardDN"] == QUEUE
|
||||
assert key0["ForwardType"] == "Queue"
|
||||
assert ibody["TimeoutForwardDN"] == QUEUE
|
||||
|
||||
assert schedule.get_holiday(HOL)["activated"] is True
|
||||
|
||||
|
|
@ -140,7 +167,6 @@ def test_activate_no_assignees_uses_fallback(holidayrouter_app, schedule, seed,
|
|||
def test_activate_no_record_is_noop(holidayrouter_app, schedule, env):
|
||||
result = holidayrouter_app.handler({"action": "activate", "date": HOL}, None)
|
||||
assert result["skipped"] == "no_record"
|
||||
# No 3CX calls were made.
|
||||
assert len(responses.calls) == 0
|
||||
|
||||
|
||||
|
|
@ -160,17 +186,15 @@ def test_activate_guard_does_not_recapture_when_already_queue(
|
|||
# (record somehow not flagged) must NOT overwrite the real originals.
|
||||
seed.holiday(HOL, slots=2, assignees=_holiday_assignees(seed))
|
||||
schedule.set_captured_ivr_routes({"key0": ORIG_KEY0, "timeout": ORIG_TIMEOUT})
|
||||
queue_route = {"To": "Queue", "Number": QUEUE, "External": ""}
|
||||
_stub_oauth()
|
||||
_stub_ivr_get(key0=queue_route, timeout=queue_route)
|
||||
_stub_ivr_get(key0_dn=QUEUE, timeout_dn=QUEUE)
|
||||
_stub_queue_get()
|
||||
_stub_patches()
|
||||
|
||||
holidayrouter_app.handler({"action": "activate", "date": HOL}, None)
|
||||
|
||||
captured = schedule.get_captured_ivr_routes()
|
||||
assert captured["key0"] == ORIG_KEY0
|
||||
assert captured["timeout"] == ORIG_TIMEOUT
|
||||
assert captured == {"key0": ORIG_KEY0, "timeout": ORIG_TIMEOUT}
|
||||
|
||||
|
||||
# ── deactivate ───────────────────────────────────────────────────────────
|
||||
|
|
@ -182,9 +206,8 @@ def test_deactivate_restores_routes_and_clears_agents(
|
|||
):
|
||||
seed.holiday(HOL, slots=2, assignees=_holiday_assignees(seed), activated=True)
|
||||
schedule.set_captured_ivr_routes({"key0": ORIG_KEY0, "timeout": ORIG_TIMEOUT})
|
||||
queue_route = {"To": "Queue", "Number": QUEUE, "External": ""}
|
||||
_stub_oauth()
|
||||
_stub_ivr_get(key0=queue_route, timeout=queue_route)
|
||||
_stub_ivr_get(key0_dn=QUEUE, timeout_dn=QUEUE) # live points at the queue
|
||||
_stub_queue_get()
|
||||
ivr_patch, queue_patch = _stub_patches()
|
||||
|
||||
|
|
@ -193,9 +216,9 @@ def test_deactivate_restores_routes_and_clears_agents(
|
|||
assert result["action"] == "deactivate"
|
||||
|
||||
# IVR restored to the captured originals.
|
||||
ibody = json.loads(ivr_patch.calls[0].request.body)
|
||||
assert ibody["Options"][0]["Route"] == ORIG_KEY0
|
||||
assert ibody["TimeoutForward"] == ORIG_TIMEOUT
|
||||
ibody = json.loads(ivr_patch.calls[-1].request.body)
|
||||
assert _key0_forward(ibody)["ForwardDN"] == ORIG_KEY0
|
||||
assert ibody["TimeoutForwardDN"] == ORIG_TIMEOUT
|
||||
|
||||
# Queue agents cleared.
|
||||
qbody = json.loads(queue_patch.calls[0].request.body)
|
||||
|
|
@ -213,19 +236,17 @@ def test_deactivate_only_restores_routes_pointing_at_queue(
|
|||
# only restore the timeout route which still points at the queue.
|
||||
seed.holiday(HOL, slots=2, assignees=_holiday_assignees(seed), activated=True)
|
||||
schedule.set_captured_ivr_routes({"key0": ORIG_KEY0, "timeout": ORIG_TIMEOUT})
|
||||
manual = {"To": "Extension", "Number": "199", "External": ""}
|
||||
queue_route = {"To": "Queue", "Number": QUEUE, "External": ""}
|
||||
_stub_oauth()
|
||||
_stub_ivr_get(key0=manual, timeout=queue_route)
|
||||
_stub_ivr_get(key0_dn="199", timeout_dn=QUEUE) # key-0 manually moved off queue
|
||||
_stub_queue_get()
|
||||
ivr_patch, _ = _stub_patches()
|
||||
|
||||
holidayrouter_app.handler({"action": "deactivate", "date": HOL}, None)
|
||||
|
||||
ibody = json.loads(ivr_patch.calls[0].request.body)
|
||||
# key-0 left as the manual value, timeout restored to the original.
|
||||
assert ibody["Options"][0]["Route"] == manual
|
||||
assert ibody["TimeoutForward"] == ORIG_TIMEOUT
|
||||
ibody = json.loads(ivr_patch.calls[-1].request.body)
|
||||
# key-0 left untouched (not re-PATCHed); only the timeout route is restored.
|
||||
assert "Forwards" not in ibody
|
||||
assert ibody["TimeoutForwardDN"] == ORIG_TIMEOUT
|
||||
|
||||
|
||||
@responses.activate
|
||||
|
|
@ -264,7 +285,7 @@ def test_three_cx_failure_returns_error(holidayrouter_app, schedule, seed, env):
|
|||
_stub_oauth()
|
||||
responses.add(
|
||||
responses.GET,
|
||||
f"{BASE}/xapi/v1/IVRs/Pbx.GetByNumber(number='{IVR}')",
|
||||
f"{BASE}/xapi/v1/Receptionists/Pbx.GetByNumber(number='{IVR}')",
|
||||
status=500,
|
||||
)
|
||||
result = holidayrouter_app.handler({"action": "activate", "date": HOL}, None)
|
||||
|
|
|
|||
|
|
@ -149,53 +149,70 @@ def test_get_ivr_and_set_ivr_routes():
|
|||
_stub_oauth()
|
||||
responses.add(
|
||||
responses.GET,
|
||||
f"{BASE}/xapi/v1/IVRs/Pbx.GetByNumber(number='800')",
|
||||
f"{BASE}/xapi/v1/Receptionists/Pbx.GetByNumber(number='800')",
|
||||
json={
|
||||
"Id": 3,
|
||||
"Number": "800",
|
||||
"Options": [{"Digit": "0", "Route": {"To": "Extension", "Number": "101"}}],
|
||||
"TimeoutForward": {"To": "Extension", "Number": "102"},
|
||||
"TimeoutForwardDN": "801",
|
||||
"TimeoutForwardType": "Queue",
|
||||
"TimeoutForwardPeerType": "Queue",
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
patched = responses.add(responses.PATCH, f"{BASE}/xapi/v1/IVRs(3)", 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"
|
||||
|
||||
key0 = {"To": "Queue", "Number": "802"}
|
||||
status = client.set_ivr_routes(3, key0_route=key0, timeout_route=key0)
|
||||
status = client.set_ivr_routes(3, key0_dn="802", timeout_dn="802")
|
||||
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"
|
||||
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 = {
|
||||
"Options": [
|
||||
{"Digit": "1", "Route": {"Number": "201"}},
|
||||
{"Digit": "0", "Route": {"Number": "101"}},
|
||||
"Forwards": [
|
||||
{"Input": "1", "ForwardDN": "201"},
|
||||
{"Input": "0", "ForwardDN": "101"},
|
||||
],
|
||||
"TimeoutForward": {"Number": "102"},
|
||||
"TimeoutForwardDN": "102",
|
||||
}
|
||||
routes = ThreeCXClient.extract_ivr_routes(ivr)
|
||||
assert routes["key0"] == {"Number": "101"}
|
||||
assert routes["timeout"] == {"Number": "102"}
|
||||
assert routes["key0"] == "101"
|
||||
assert routes["timeout"] == "102"
|
||||
|
||||
|
||||
def test_extract_ivr_routes_missing_key0_is_none():
|
||||
routes = ThreeCXClient.extract_ivr_routes({"Options": [], "TimeoutForward": None})
|
||||
routes = ThreeCXClient.extract_ivr_routes(
|
||||
{"Forwards": [], "TimeoutForwardDN": 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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue