tableau-de-bord/tests/test_api.py

import pytest

from app import db
from app.api import reset_rate_limiter
from app.models import TimeSlot, WorkplacePresenceEvent

TOKEN = "test-api-token"
URL = "/api/v1/workplace-presence"


@pytest.fixture(autouse=True)
def api_state(monkeypatch):
    monkeypatch.setenv("WORKLOG_API_TOKEN", TOKEN)
    reset_rate_limiter()
    yield
    reset_rate_limiter()


def post(client, payload, **headers):
    return client.post(URL, json=payload, headers={"Authorization": f"Bearer {TOKEN}", **headers})


def test_arrival_and_departure_create_expected_records(client, app):
    arrival = post(
        client,
        {"event": "arrival", "occurred_at": "2026-08-13T08:00:00+02:00", "idempotency_key": "a"},
    )
    assert arrival.status_code == 201
    assert arrival.json["status"] == "created"
    assert arrival.json["time_slot_id"] is None

    departure = post(
        client,
        {"event": "departure", "occurred_at": "2026-08-13T17:00:00+02:00", "idempotency_key": "d"},
    )
    assert departure.status_code == 200
    assert departure.json["time_slot_id"] is not None
    with app.app_context():
        assert db.session.query(TimeSlot).count() == 1


def test_replay_does_not_duplicate(client, app):
    payload = {
        "event": "arrival",
        "occurred_at": "2026-08-13T08:00:00+02:00",
        "idempotency_key": "same",
    }
    assert post(client, payload).status_code == 201
    replay = post(client, payload)
    assert replay.status_code == 200
    assert replay.json["status"] == "replayed"
    with app.app_context():
        assert db.session.query(WorkplacePresenceEvent).count() == 1


def test_conflicts_and_idempotency_header_mismatch(client):
    departure = post(
        client,
        {"event": "departure", "occurred_at": "2026-08-13T08:00:00+02:00", "idempotency_key": "d"},
    )
    assert departure.status_code == 409
    mismatch = post(
        client,
        {"event": "arrival", "occurred_at": "2026-08-13T09:00:00+02:00", "idempotency_key": "body"},
        **{"X-Idempotency-Key": "header"},
    )
    assert mismatch.status_code == 422
    assert (
        post(
            client,
            {
                "event": "arrival",
                "occurred_at": "2026-08-13T09:00:00+02:00",
                "idempotency_key": "a",
            },
        ).status_code
        == 201
    )
    assert (
        post(
            client,
            {
                "event": "arrival",
                "occurred_at": "2026-08-13T10:00:00+02:00",
                "idempotency_key": "b",
            },
        ).status_code
        == 409
    )


@pytest.mark.parametrize(
    ("payload", "status"),
    [
        ({"event": "arrival", "occurred_at": "bad", "idempotency_key": "a"}, 422),
        ({"event": "arrival", "occurred_at": "2026-08-13T08:00:00+02:00", "extra": "x"}, 422),
        (
            {
                "event": "arrival",
                "occurred_at": "2026-08-13T08:00:00+02:00",
                "idempotency_key": "a",
            },
            201,
        ),
    ],
)
def test_validation_and_header_idempotency(client, payload, status):
    headers = {"X-Idempotency-Key": "a"} if "idempotency_key" not in payload else {}
    response = post(client, payload, **headers)
    assert response.status_code == status


def test_authentication_and_disabled_config(client, app, monkeypatch):
    monkeypatch.delenv("WORKLOG_API_TOKEN")
    assert post(client, {}).status_code == 503
    monkeypatch.setenv("WORKLOG_API_TOKEN", TOKEN)
    assert client.post(URL).status_code == 401
    app.config["HOME_ASSISTANT"] = None
    assert post(client, {}).status_code == 404


def test_http_errors_are_json_uncached_and_do_not_leak(client):
    response = client.post(
        URL, data="{}", content_type="text/plain", headers={"Authorization": f"Bearer {TOKEN}"}
    )
    assert response.status_code == 415
    assert response.headers["Cache-Control"] == "no-store"
    assert "traceback" not in response.get_data(as_text=True).lower()
    assert TOKEN not in response.get_data(as_text=True)

    malformed = client.post(
        URL,
        data="{",
        content_type="application/json",
        headers={"Authorization": f"Bearer {TOKEN}"},
    )
    assert malformed.status_code == 400
    assert client.get(URL).status_code == 405


def test_body_limit_and_rate_limit(client):
    oversized = post(
        client,
        {
            "event": "arrival",
            "occurred_at": "2026-08-13T08:00:00+02:00",
            "idempotency_key": "x" * 4020,
        },
    )
    assert oversized.status_code == 413
    reset_rate_limiter()
    for index in range(30):
        response = post(
            client,
            {
                "event": "arrival",
                "occurred_at": "2026-08-13T08:00:00+02:00",
                "idempotency_key": f"rate-{index}",
            },
        )
        assert response.status_code != 429
    assert post(client, {}).status_code == 429