mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 07:53:11 +00:00
88e782c205
Holiday day-shifts (08:00-17:00 ET) with N slots and 1.5x pay. A new afterhours-holiday-router Lambda, fired by per-holiday EventBridge Scheduler one-offs, repoints IVR 800 (key-0 + no-input/timeout) to holiday queue 802 and sets 802's membership to the day's assignees (ext 100 fallback when unfilled), reverting at 17:00. Pickups after a shift starts go through an admin Approve/Deny flow for both regular and holiday shifts. Pay (weekly post + /oncall pay) shows holiday rates distinctly. Adds HOLIDAY and PICKUP_REQUEST DynamoDB record types, scheduler IAM scoped to holiday-* schedules with conditioned PassRole, and the holiday-router function with a 60-day log group and error alarm.
273 lines
9.5 KiB
Python
273 lines
9.5 KiB
Python
"""Tests for the holiday-router Lambda handler.
|
|
|
|
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.
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
import responses
|
|
|
|
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": ""}
|
|
|
|
|
|
@pytest.fixture
|
|
def env(monkeypatch):
|
|
monkeypatch.setenv("TCX_SECRET_PREFIX", "afterhours-shift-manager/3cx-")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def secrets(holidayrouter_app, monkeypatch):
|
|
# domain -> test host; client-id/secret -> anything (oauth token is stubbed).
|
|
def fake(secret_id):
|
|
if secret_id.endswith("domain"):
|
|
return "test.3cx.us"
|
|
return "x"
|
|
|
|
monkeypatch.setattr(holidayrouter_app, "get_secret", fake)
|
|
|
|
|
|
def _stub_oauth():
|
|
responses.add(
|
|
responses.POST,
|
|
f"{BASE}/connect/token",
|
|
json={"access_token": "tok"},
|
|
status=200,
|
|
)
|
|
|
|
|
|
def _stub_ivr_get(key0=ORIG_KEY0, timeout=ORIG_TIMEOUT, ivr_id=7):
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE}/xapi/v1/IVRs/Pbx.GetByNumber(number='{IVR}')",
|
|
json={
|
|
"Id": ivr_id,
|
|
"Number": IVR,
|
|
"Options": [{"Digit": "0", "Route": key0}],
|
|
"TimeoutForward": timeout,
|
|
},
|
|
status=200,
|
|
)
|
|
|
|
|
|
def _stub_queue_get(queue_id=9):
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='{QUEUE}')",
|
|
json={"Id": queue_id, "Number": QUEUE},
|
|
status=200,
|
|
)
|
|
|
|
|
|
def _stub_patches(ivr_id=7, queue_id=9):
|
|
ivr_patch = responses.add(
|
|
responses.PATCH, f"{BASE}/xapi/v1/IVRs({ivr_id})", status=200
|
|
)
|
|
queue_patch = responses.add(
|
|
responses.PATCH, f"{BASE}/xapi/v1/Queues({queue_id})", status=200
|
|
)
|
|
return ivr_patch, queue_patch
|
|
|
|
|
|
def _holiday_assignees(seed):
|
|
return {
|
|
"114": {"name": "Alice", "claimed_at": "2026-07-01T00:00:00-04:00"},
|
|
"115": {"name": "Bob", "claimed_at": "2026-07-01T00:00:00-04:00"},
|
|
}
|
|
|
|
|
|
# ── activate ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
@responses.activate
|
|
def test_activate_captures_routes_sets_agents_repoints_ivr(
|
|
holidayrouter_app, schedule, seed, env
|
|
):
|
|
seed.holiday(HOL, slots=2, assignees=_holiday_assignees(seed))
|
|
_stub_oauth()
|
|
_stub_ivr_get()
|
|
_stub_queue_get()
|
|
ivr_patch, queue_patch = _stub_patches()
|
|
|
|
result = holidayrouter_app.handler({"action": "activate", "date": HOL}, None)
|
|
|
|
assert result["action"] == "activate"
|
|
assert set(result["agents"]) == {"114", "115"}
|
|
|
|
# Captured the live (pre-holiday) routes into CONFIG.
|
|
captured = schedule.get_captured_ivr_routes()
|
|
assert captured["key0"] == ORIG_KEY0
|
|
assert captured["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
|
|
|
|
assert schedule.get_holiday(HOL)["activated"] is True
|
|
|
|
|
|
@responses.activate
|
|
def test_activate_no_assignees_uses_fallback(holidayrouter_app, schedule, seed, env):
|
|
seed.holiday(HOL, slots=2, assignees={})
|
|
_stub_oauth()
|
|
_stub_ivr_get()
|
|
_stub_queue_get()
|
|
_, queue_patch = _stub_patches()
|
|
|
|
result = holidayrouter_app.handler({"action": "activate", "date": HOL}, None)
|
|
|
|
assert result["agents"] == ["100"]
|
|
qbody = json.loads(queue_patch.calls[0].request.body)
|
|
assert [a["Number"] for a in qbody["Agents"]] == ["100"]
|
|
|
|
|
|
@responses.activate
|
|
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
|
|
|
|
|
|
@responses.activate
|
|
def test_activate_already_active_is_noop(holidayrouter_app, schedule, seed, env):
|
|
seed.holiday(HOL, slots=2, assignees=_holiday_assignees(seed), activated=True)
|
|
result = holidayrouter_app.handler({"action": "activate", "date": HOL}, None)
|
|
assert result["skipped"] == "already_active"
|
|
assert len(responses.calls) == 0
|
|
|
|
|
|
@responses.activate
|
|
def test_activate_guard_does_not_recapture_when_already_queue(
|
|
holidayrouter_app, schedule, seed, env
|
|
):
|
|
# IVR already points at the queue (e.g. a prior run); a fresh activation
|
|
# (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_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
|
|
|
|
|
|
# ── deactivate ───────────────────────────────────────────────────────────
|
|
|
|
|
|
@responses.activate
|
|
def test_deactivate_restores_routes_and_clears_agents(
|
|
holidayrouter_app, schedule, seed, env
|
|
):
|
|
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_queue_get()
|
|
ivr_patch, queue_patch = _stub_patches()
|
|
|
|
result = holidayrouter_app.handler({"action": "deactivate", "date": HOL}, None)
|
|
|
|
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
|
|
|
|
# Queue agents cleared.
|
|
qbody = json.loads(queue_patch.calls[0].request.body)
|
|
assert qbody["Agents"] == []
|
|
|
|
assert schedule.get_holiday(HOL)["activated"] is False
|
|
assert schedule.get_captured_ivr_routes() is None
|
|
|
|
|
|
@responses.activate
|
|
def test_deactivate_only_restores_routes_pointing_at_queue(
|
|
holidayrouter_app, schedule, seed, env
|
|
):
|
|
# key-0 was manually changed away from the queue after activation; leave it,
|
|
# 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_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
|
|
|
|
|
|
@responses.activate
|
|
def test_deactivate_no_record_is_noop(holidayrouter_app, schedule, env):
|
|
result = holidayrouter_app.handler({"action": "deactivate", "date": HOL}, None)
|
|
assert result["skipped"] == "no_record"
|
|
assert len(responses.calls) == 0
|
|
|
|
|
|
@responses.activate
|
|
def test_deactivate_not_active_is_noop(holidayrouter_app, schedule, seed, env):
|
|
seed.holiday(HOL, slots=2, assignees=_holiday_assignees(seed), activated=False)
|
|
result = holidayrouter_app.handler({"action": "deactivate", "date": HOL}, None)
|
|
assert result["skipped"] == "not_active"
|
|
assert len(responses.calls) == 0
|
|
|
|
|
|
# ── dispatch / errors ────────────────────────────────────────────────────
|
|
|
|
|
|
def test_missing_date_errors(holidayrouter_app, schedule, env):
|
|
result = holidayrouter_app.handler({"action": "activate"}, None)
|
|
assert result["error"] is True
|
|
assert result["reason"] == "missing_date"
|
|
|
|
|
|
def test_unknown_action_errors(holidayrouter_app, schedule, env):
|
|
result = holidayrouter_app.handler({"action": "nope", "date": HOL}, None)
|
|
assert result["error"] is True
|
|
assert result["reason"] == "unknown_action"
|
|
|
|
|
|
@responses.activate
|
|
def test_three_cx_failure_returns_error(holidayrouter_app, schedule, seed, env):
|
|
seed.holiday(HOL, slots=2, assignees=_holiday_assignees(seed))
|
|
_stub_oauth()
|
|
responses.add(
|
|
responses.GET,
|
|
f"{BASE}/xapi/v1/IVRs/Pbx.GetByNumber(number='{IVR}')",
|
|
status=500,
|
|
)
|
|
result = holidayrouter_app.handler({"action": "activate", "date": HOL}, None)
|
|
assert result["error"] is True
|
|
# The holiday was not flagged activated since the 3CX call failed.
|
|
assert schedule.get_holiday(HOL)["activated"] is False
|