315 lines
9.7 KiB
Python
315 lines
9.7 KiB
Python
import re
|
|
|
|
import pytest
|
|
|
|
from app import db
|
|
from app.api import reset_rate_limiter
|
|
from app.models import TimeSlot, WorkEntry, 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_arrival_entry_can_be_fully_edited_from_web_form(client, app):
|
|
arrival = post(
|
|
client,
|
|
{
|
|
"event": "arrival",
|
|
"occurred_at": "2026-08-13T08:00:00+02:00",
|
|
"idempotency_key": "web-edit-arrival",
|
|
},
|
|
)
|
|
assert arrival.status_code == 201
|
|
entry_id = arrival.json["entry_id"]
|
|
|
|
form = client.get(f"/entries/{entry_id}/edit")
|
|
assert form.status_code == 200
|
|
assert "Véhicule à moteur seul" in form.text
|
|
assert re.search(
|
|
r'<option value="moteur_seul".*?data-has-motor="true".*?selected',
|
|
form.text,
|
|
re.DOTALL,
|
|
)
|
|
assert "Citadine électrique" in form.text
|
|
assert re.search(r'name="motor_vehicle_id" value="citadine".*?checked', form.text, re.DOTALL)
|
|
assert 'name="start_time"' in form.text
|
|
assert 'name="end_time"' in form.text
|
|
|
|
formation = client.post(
|
|
f"/entries/{entry_id}/edit",
|
|
data={
|
|
"date": "2026-08-13",
|
|
"day_type": "FORMATION",
|
|
"journey_profile_id": "moteur_velo",
|
|
"motor_vehicle_id": "familiale",
|
|
"start_time": ["08:00"],
|
|
"end_time": ["12:00"],
|
|
"comment": "Formation",
|
|
},
|
|
)
|
|
assert formation.status_code == 302
|
|
|
|
with app.app_context():
|
|
entry = db.session.get(WorkEntry, entry_id)
|
|
assert entry.day_type == "FORMATION"
|
|
assert entry.journey_profile_id == "moteur_velo"
|
|
assert entry.motor_vehicle_id == "familiale"
|
|
assert [(slot.start_time.hour, slot.end_time.hour) for slot in entry.time_slots] == [
|
|
(8, 12)
|
|
]
|
|
|
|
garde = client.post(
|
|
f"/entries/{entry_id}/edit",
|
|
data={
|
|
"date": "2026-08-13",
|
|
"day_type": "GARDE",
|
|
"journey_profile_id": "moteur_seul",
|
|
"motor_vehicle_id": "moto",
|
|
"start_time": ["08:00", "13:00", ""],
|
|
"end_time": ["12:00", "17:00", ""],
|
|
"comment": "Garde",
|
|
},
|
|
)
|
|
assert garde.status_code == 302
|
|
|
|
with app.app_context():
|
|
entry = db.session.get(WorkEntry, entry_id)
|
|
assert entry.day_type == "GARDE"
|
|
assert entry.journey_profile_id == "moteur_seul"
|
|
assert entry.motor_vehicle_id == "moto"
|
|
assert [(slot.start_time.hour, slot.end_time.hour) for slot in entry.time_slots] == [
|
|
(8, 12),
|
|
(13, 17),
|
|
]
|
|
event = db.session.scalar(
|
|
db.select(WorkplacePresenceEvent).where(
|
|
WorkplacePresenceEvent.idempotency_key == "web-edit-arrival"
|
|
)
|
|
)
|
|
assert event.entry_id == entry_id
|
|
assert event.time_slot_id is None
|
|
|
|
empty_slots = client.post(
|
|
f"/entries/{entry_id}/edit",
|
|
data={
|
|
"date": "2026-08-13",
|
|
"day_type": "GARDE",
|
|
"journey_profile_id": "moteur_seul",
|
|
"motor_vehicle_id": "moto",
|
|
"start_time": [""],
|
|
"end_time": [""],
|
|
"comment": "Garde sans plage",
|
|
},
|
|
)
|
|
assert empty_slots.status_code == 302
|
|
|
|
with app.app_context():
|
|
entry = db.session.get(WorkEntry, entry_id)
|
|
assert entry.time_slots == []
|
|
assert entry.day_type == "GARDE"
|
|
assert db.session.query(WorkplacePresenceEvent).count() == 1
|
|
|
|
|
|
def test_web_edit_recreates_slots_without_losing_arrival_departure_events(client, app):
|
|
arrival = post(
|
|
client,
|
|
{
|
|
"event": "arrival",
|
|
"occurred_at": "2026-08-14T08:00:00+02:00",
|
|
"idempotency_key": "recreate-arrival",
|
|
},
|
|
)
|
|
departure = post(
|
|
client,
|
|
{
|
|
"event": "departure",
|
|
"occurred_at": "2026-08-14T17:00:00+02:00",
|
|
"idempotency_key": "recreate-departure",
|
|
},
|
|
)
|
|
assert arrival.status_code == 201
|
|
assert departure.status_code == 200
|
|
entry_id = arrival.json["entry_id"]
|
|
|
|
response = client.post(
|
|
f"/entries/{entry_id}/edit",
|
|
data={
|
|
"date": "2026-08-14",
|
|
"day_type": "FORMATION",
|
|
"journey_profile_id": "moteur_velo",
|
|
"motor_vehicle_id": "familiale",
|
|
"start_time": ["09:00", "13:00", ""],
|
|
"end_time": ["12:00", "17:00", ""],
|
|
"comment": "Plages recréées",
|
|
},
|
|
)
|
|
assert response.status_code == 302
|
|
|
|
with app.app_context():
|
|
entry = db.session.get(WorkEntry, entry_id)
|
|
assert len(entry.time_slots) == 2
|
|
events = db.session.scalars(
|
|
db.select(WorkplacePresenceEvent)
|
|
.where(WorkplacePresenceEvent.entry_id == entry_id)
|
|
.order_by(WorkplacePresenceEvent.event_type)
|
|
).all()
|
|
assert {event.idempotency_key for event in events} == {
|
|
"recreate-arrival",
|
|
"recreate-departure",
|
|
}
|
|
assert all(event.time_slot_id is None for event in events)
|
|
|
|
|
|
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
|