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

"""Tests for the roster-api Lambda handler."""

import base64
import json
from pathlib import Path
from unittest.mock import MagicMock

import pytest

TOKEN = "roster-test-token"
SECRET_NAME = "afterhours-shift-manager/roster-api-token"


@pytest.fixture
def env(monkeypatch):
    monkeypatch.setenv("ROSTER_API_TOKEN_SECRET", SECRET_NAME)


@pytest.fixture
def secrets(rosterapi_app, monkeypatch, env):
    monkeypatch.setattr(rosterapi_app, "get_secret", lambda _id: TOKEN)
    rosterapi_app._cached_token = None


def _event(
    method="PUT",
    path="/roster",
    body=None,
    token=TOKEN,
    headers=None,
    is_base64=False,
    extension=None,
    include_auth=True,
):
    hdrs = {}
    if headers:
        hdrs.update(headers)
    if include_auth and token is not None:
        hdrs.setdefault("authorization", f"Bearer {token}")
    payload = None
    if body is not None:
        raw = (
            json.dumps(body).encode("utf-8")
            if not isinstance(body, (bytes, str))
            else body
        )
        if isinstance(raw, str):
            raw = raw.encode("utf-8")
        if is_base64:
            payload = base64.b64encode(raw).decode("ascii")
        else:
            payload = raw.decode("utf-8")
    event = {
        "version": "2.0",
        "routeKey": f"{method} {path}",
        "rawPath": path,
        "headers": hdrs,
        "requestContext": {"http": {"method": method, "path": path}},
        "body": payload,
        "isBase64Encoded": is_base64,
    }
    if extension is not None:
        event["pathParameters"] = {"extension": extension}
    return event


def _put_body(**overrides):
    data = {
        "name": "Pat Smith",
        "extension": "110",
        "slack_user_id": "U123ABCDE",
    }
    data.update(overrides)
    return data


def test_put_writes_all_three_fields(rosterapi_app, schedule, secrets):
    result = rosterapi_app.handler(_event(body=_put_body()), None)
    assert result["statusCode"] == 200
    assert json.loads(result["body"]) == {"ok": True}
    emp = schedule.get_employee_by_extension("110")
    assert emp["name"] == "Pat Smith"
    assert emp["extension"] == "110"
    assert emp["slack_user_id"] == "U123ABCDE"


def test_put_updates_existing_and_preserves_shift_rate(
    rosterapi_app, schedule, seed, secrets
):
    seed.roster("110", "Old Name", slack_user_id="", shift_rate="80")
    result = rosterapi_app.handler(_event(body=_put_body()), None)
    assert result["statusCode"] == 200
    emp = schedule.get_employee_by_extension("110")
    assert emp["name"] == "Pat Smith"
    assert emp["slack_user_id"] == "U123ABCDE"
    assert emp["shift_rate"] == "80"


def test_put_trims_surrounding_whitespace(rosterapi_app, schedule, secrets):
    result = rosterapi_app.handler(
        _event(
            body={
                "name": "  Pat Smith  ",
                "extension": " 110 ",
                "slack_user_id": " U123ABCDE ",
            }
        ),
        None,
    )
    assert result["statusCode"] == 200
    emp = schedule.get_employee_by_extension("110")
    assert emp["name"] == "Pat Smith"
    assert emp["slack_user_id"] == "U123ABCDE"


def test_put_accepts_base64_body(rosterapi_app, schedule, secrets):
    result = rosterapi_app.handler(_event(body=_put_body(), is_base64=True), None)
    assert result["statusCode"] == 200
    assert schedule.get_employee_by_extension("110")["slack_user_id"] == "U123ABCDE"


def test_put_accepts_case_insensitive_authorization_header(
    rosterapi_app, schedule, secrets
):
    result = rosterapi_app.handler(
        _event(
            body=_put_body(),
            include_auth=False,
            headers={"Authorization": f"Bearer {TOKEN}"},
        ),
        None,
    )
    assert result["statusCode"] == 200


def test_401_missing_token(rosterapi_app, schedule, secrets):
    result = rosterapi_app.handler(_event(body=_put_body(), include_auth=False), None)
    assert result["statusCode"] == 401
    assert schedule.get_employee_by_extension("110") is None


def test_401_wrong_token(rosterapi_app, schedule, secrets):
    result = rosterapi_app.handler(_event(body=_put_body(), token="nope"), None)
    assert result["statusCode"] == 401
    assert schedule.get_employee_by_extension("110") is None


def test_authorize_strips_secret_trailing_newline(
    rosterapi_app, schedule, env, monkeypatch
):
    monkeypatch.setattr(rosterapi_app, "get_secret", lambda _id: TOKEN + "\n")
    rosterapi_app._cached_token = None
    result = rosterapi_app.handler(_event(body=_put_body()), None)
    assert result["statusCode"] == 200
    assert schedule.get_employee_by_extension("110")["slack_user_id"] == "U123ABCDE"


def test_503_when_secret_is_whitespace_only(rosterapi_app, schedule, env, monkeypatch):
    monkeypatch.setattr(rosterapi_app, "get_secret", lambda _id: "\n")
    rosterapi_app._cached_token = None
    result = rosterapi_app.handler(_event(body=_put_body()), None)
    assert result["statusCode"] == 503
    assert schedule.get_employee_by_extension("110") is None


def test_503_when_secret_read_fails(rosterapi_app, schedule, env, monkeypatch):
    def boom(_id):
        raise RuntimeError("secrets down")

    monkeypatch.setattr(rosterapi_app, "get_secret", boom)
    rosterapi_app._cached_token = None
    result = rosterapi_app.handler(_event(body=_put_body()), None)
    assert result["statusCode"] == 503
    assert schedule.get_employee_by_extension("110") is None


def test_400_missing_field(rosterapi_app, schedule, secrets):
    body = _put_body()
    del body["slack_user_id"]
    result = rosterapi_app.handler(_event(body=body), None)
    assert result["statusCode"] == 400
    assert schedule.get_employee_by_extension("110") is None


def test_400_extra_field(rosterapi_app, schedule, secrets):
    body = _put_body(extra="nope")
    result = rosterapi_app.handler(_event(body=body), None)
    assert result["statusCode"] == 400
    assert schedule.get_employee_by_extension("110") is None


def test_400_blank_and_control_values(rosterapi_app, schedule, secrets):
    for body in (
        _put_body(name="   "),
        _put_body(name="Pat\nSmith"),
        _put_body(extension="11a"),
        _put_body(slack_user_id="U123 AB"),
        _put_body(extension=""),
    ):
        result = rosterapi_app.handler(_event(body=body), None)
        assert result["statusCode"] == 400, body
    assert schedule.get_roster() == []


def test_400_oversized_body(rosterapi_app, schedule, secrets):
    huge = _put_body(name="P" * 5000)
    result = rosterapi_app.handler(_event(body=huge), None)
    assert result["statusCode"] == 400
    assert schedule.get_employee_by_extension("110") is None


def test_400_invalid_json(rosterapi_app, schedule, secrets):
    result = rosterapi_app.handler(_event(body="{not json"), None)
    assert result["statusCode"] == 400


def test_delete_missing_row_is_204(rosterapi_app, schedule, secrets):
    result = rosterapi_app.handler(
        _event(method="DELETE", path="/roster/999", extension="999", body=None),
        None,
    )
    assert result["statusCode"] == 204
    assert result["body"] == ""
    assert schedule.get_employee_by_extension("999") is None


def test_delete_existing_row_is_204(rosterapi_app, schedule, seed, secrets):
    seed.roster("110", "Pat Smith", slack_user_id="U123ABCDE")
    result = rosterapi_app.handler(
        _event(method="DELETE", path="/roster/110", extension="110", body=None),
        None,
    )
    assert result["statusCode"] == 204
    assert schedule.get_employee_by_extension("110") is None


def test_405_other_methods(rosterapi_app, secrets):
    result = rosterapi_app.handler(
        _event(method="GET", path="/roster", body=None), None
    )
    assert result["statusCode"] == 405


def test_never_logs_authorization_or_body(rosterapi_app, schedule, secrets, caplog):
    import logging

    caplog.set_level(logging.DEBUG)
    body = _put_body(name="SecretName")
    rosterapi_app.handler(
        _event(body=body, token=TOKEN, headers={"authorization": f"Bearer {TOKEN}"}),
        None,
    )
    ours = "\n".join(
        rec.getMessage() for rec in caplog.records if "roster-api" in rec.pathname
    )
    assert TOKEN not in ours
    assert "SecretName" not in ours
    assert "Bearer" not in ours


def test_never_calls_roster_sync(rosterapi_app):
    source = (
        Path(__file__)
        .resolve()
        .parents[2]
        .joinpath("src/roster-api/app.py")
        .read_text()
    )
    assert "roster-sync" not in source
    assert "roster_sync" not in source
    assert "InvokeFunction" not in source
    assert rosterapi_app.handler.__module__ == "rosterapi_app"


def test_500_on_unexpected_failure(rosterapi_app, secrets, monkeypatch):
    monkeypatch.setattr(
        rosterapi_app,
        "ShiftSchedule",
        MagicMock(side_effect=RuntimeError("ddb down")),
    )
    result = rosterapi_app.handler(_event(body=_put_body()), None)
    assert result["statusCode"] == 500