afterhours-shift-manager/tests/holiday_router/test_handler.py

"""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.

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

import pytest
import responses

HOL = "2026-07-04"
BASE = "https://test.3cx.us"
QUEUE = "802"
IVR = "800"
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
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_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/Receptionists/Pbx.GetByNumber(number='{IVR}')",
        json={
            "Id": ivr_id,
            "Number": IVR,
            "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=QUEUE_ID):
    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=IVR_ID, queue_id=QUEUE_ID):
    ivr_patch = responses.add(
        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
    )
    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"},
    }


def _key0_forward(body):
    return next(f for f in body["Forwards"] if f["Input"] == "0")


# ── 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) DNs into CONFIG.
    captured = schedule.get_captured_ivr_routes()
    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 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


@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"
    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})
    _stub_oauth()
    _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, "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})
    _stub_oauth()
    _stub_ivr_get(key0_dn=QUEUE, timeout_dn=QUEUE)  # live points at the queue
    _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[-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)
    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})
    _stub_oauth()
    _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[-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
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/Receptionists/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