377 lines
13 KiB
Python
377 lines
13 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
from sqlmodel import select
|
|
|
|
from app.model.models import OneShotAccess
|
|
|
|
# Berlin's DST rules as a POSIX TZ string: +01:00 in winter, +02:00 in summer.
|
|
BERLIN_TZ = "CET-1CEST,M3.5.0,M10.5.0/3"
|
|
|
|
|
|
def _post_oneshot_aa(client, auth_headers, name: str, ends_at: str):
|
|
"""Create a oneshot access authorization and return the response."""
|
|
return client.post(
|
|
"/api/v1/aa/",
|
|
json={
|
|
"name": name,
|
|
"type": "oneshot",
|
|
"is_active": True,
|
|
"oneshot": {"uses": 1, "ends_at": ends_at},
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
|
|
def test_create_access_auth_os(client, auth_headers):
|
|
"""Test creating a new access authorization with oneshot type."""
|
|
aa_data = {
|
|
"name": "New os_AA",
|
|
"type": "oneshot",
|
|
"is_active": True,
|
|
"oneshot": {"uses": 1, "ends_at": "2029-07-27"},
|
|
}
|
|
|
|
response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["name"] == "New os_AA"
|
|
assert data["type"] == "oneshot"
|
|
assert data["is_active"] is True
|
|
assert "id" in data
|
|
assert data["oneshot"]["uses"] == 1
|
|
|
|
|
|
def test_create_wrong_aa_type(client, auth_headers):
|
|
"""Test creating a new access authorization with oneshot type."""
|
|
aa_data = {
|
|
"name": "New AA",
|
|
"type": "wrong",
|
|
"is_active": True,
|
|
}
|
|
|
|
response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_create_timetable_aa_is_rejected(client, auth_headers):
|
|
"""The deprecated timetable type must be refused"""
|
|
response = client.post(
|
|
"/api/v1/aa/",
|
|
json={"name": "Legacy timetable AA", "type": "timetable", "is_active": True},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_update_access_auth_to_timetable_is_rejected(client, auth_headers, test_aa_os):
|
|
"""PATCH must refuse type="timetable" """
|
|
response = client.patch(
|
|
f"/api/v1/aa/{test_aa_os.id}",
|
|
json={"type": "timetable"},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
still_fine = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
|
assert still_fine.status_code == 200
|
|
assert still_fine.json()["type"] == "oneshot"
|
|
|
|
|
|
def test_get_all_access_auths(client, auth_headers, test_aa_os):
|
|
"""Test retrieving all access authorizations."""
|
|
response = client.get("/api/v1/aa/", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
aa_list = response.json()
|
|
assert len(aa_list) >= 1
|
|
aa_names = [aa["name"] for aa in aa_list]
|
|
assert test_aa_os.name in aa_names
|
|
|
|
|
|
def test_get_access_auth_by_id(client, auth_headers, test_aa_os):
|
|
"""Test retrieving a specific access authorization by ID."""
|
|
response = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["id"] == test_aa_os.id
|
|
assert data["name"] == test_aa_os.name
|
|
|
|
|
|
def test_get_nonexistent_access_auth(client, auth_headers):
|
|
"""Test retrieving a non-existent access authorization."""
|
|
response = client.get("/api/v1/aa/99999", headers=auth_headers)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_os):
|
|
"""Test assigning an access authorization to a group."""
|
|
response = client.put(
|
|
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["id"] == test_group.id
|
|
# The AA should now be in the group's accessauths
|
|
# Note: The response model might not include the full relationship
|
|
|
|
|
|
def test_assign_already_assigned_access_auth(
|
|
client, auth_headers, test_group, test_aa_os
|
|
):
|
|
"""Test assigning an already assigned access authorization."""
|
|
# First assignment
|
|
client.put(
|
|
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
|
)
|
|
|
|
# Second assignment should indicate it's already assigned
|
|
response = client.put(
|
|
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
|
)
|
|
# According to the code, this returns 409 with "already assigned" message
|
|
assert response.status_code == 409
|
|
assert "already assigned" in response.json()["detail"].lower()
|
|
|
|
|
|
def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_os):
|
|
"""Test unassigning an access authorization from a group."""
|
|
# First assign
|
|
client.put(
|
|
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
|
)
|
|
|
|
# Then unassign
|
|
response = client.put(
|
|
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_os):
|
|
"""Test unassigning a non-existent assignment."""
|
|
response = client.put(
|
|
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_assign_to_nonexistent_group(client, auth_headers, test_aa_os):
|
|
"""Test assigning an AA to a non-existent group."""
|
|
response = client.put(
|
|
f"/api/v1/aa/assign/99999/{test_aa_os.id}", headers=auth_headers
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_assign_nonexistent_aa(client, auth_headers, test_group):
|
|
"""Test assigning a non-existent AA to a group."""
|
|
response = client.put(
|
|
f"/api/v1/aa/assign/{test_group.id}/99999", headers=auth_headers
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_update_access_auth(client, auth_headers, test_aa_os):
|
|
"""Test updating an access authorization."""
|
|
update_data = {"name": "Updated AA", "is_active": False}
|
|
|
|
response = client.patch(
|
|
f"/api/v1/aa/{test_aa_os.id}", json=update_data, headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["name"] == "Updated AA"
|
|
assert data["is_active"] is False
|
|
|
|
|
|
def test_update_nonexistent_access_auth(client, auth_headers):
|
|
"""Test updating a non-existent access authorization."""
|
|
update_data = {"name": "Updated"}
|
|
response = client.patch("/api/v1/aa/99999", json=update_data, headers=auth_headers)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_delete_access_auth(client, auth_headers, test_aa_os):
|
|
"""Test deleting an access authorization."""
|
|
response = client.delete(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
assert "deleted successfully" in response.json()["message"].lower()
|
|
|
|
# Verify AA is deleted
|
|
response = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_delete_nonexistent_access_auth(client, auth_headers):
|
|
"""Test deleting a non-existent access authorization."""
|
|
response = client.delete("/api/v1/aa/99999", headers=auth_headers)
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_aa_os_operations_by_non_admin(client, test_aa_os, user_auth_headers):
|
|
"""Test that non-admin users cannot perform AA operations."""
|
|
# Try to create an AA
|
|
response = client.post(
|
|
"/api/v1/aa/",
|
|
json={"name": "test", "is_active": True},
|
|
headers=user_auth_headers,
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
# Try to get all AAs
|
|
response = client.get("/api/v1/aa/", headers=user_auth_headers)
|
|
assert response.status_code == 403
|
|
|
|
# Try to assign AA
|
|
response = client.put(
|
|
f"/api/v1/aa/assign/1/{test_aa_os.id}", headers=user_auth_headers
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.parametrize("system_tz", [BERLIN_TZ, "UTC-2"], indirect=True)
|
|
def test_create_access_auth_os_returns_timezone(client, auth_headers, system_tz):
|
|
"""The oneshot end time must come back with a timezone, not as a naive value."""
|
|
response = _post_oneshot_aa(
|
|
client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
ends_at = response.json()["oneshot"]["ends_at"]
|
|
parsed = datetime.fromisoformat(ends_at)
|
|
assert parsed.utcoffset() is not None, f"response has no timezone: {ends_at}"
|
|
assert parsed == datetime(2029, 7, 27, 10, 0, tzinfo=UTC), (
|
|
f"the +02:00 offset was not normalised to UTC: {ends_at}"
|
|
)
|
|
|
|
|
|
def test_get_access_auth_os_returns_timezone(client, auth_headers):
|
|
"""Reading the value back from the database keeps it timezone aware."""
|
|
created = _post_oneshot_aa(
|
|
client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
|
|
)
|
|
assert created.status_code == 200
|
|
aa_id = created.json()["id"]
|
|
|
|
for endpoint in (f"/api/v1/aa/{aa_id}", "/api/v1/aa/"):
|
|
response = client.get(endpoint, headers=auth_headers)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
items = body if isinstance(body, list) else [body]
|
|
auths = [aa for aa in items if aa.get("oneshot")]
|
|
assert auths, f"no oneshot auth in {endpoint}: {body}"
|
|
for auth in auths:
|
|
ends_at = auth["oneshot"]["ends_at"]
|
|
parsed = datetime.fromisoformat(ends_at)
|
|
assert parsed.utcoffset() is not None, (
|
|
f"{endpoint} has no timezone: {ends_at}"
|
|
)
|
|
assert parsed == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
def test_oneshot_timezone_is_stored_in_db(client, auth_headers, db_session):
|
|
"""The submitted offset is normalized before the value reaches the database."""
|
|
response = _post_oneshot_aa(
|
|
client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
db_session.expire_all()
|
|
stored = db_session.exec(select(OneShotAccess)).one()
|
|
assert stored.ends_at == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
|
|
|
|
# The column itself must hold the UTC wall clock, not the submitted local time.
|
|
raw = db_session.exec(text("SELECT ends_at FROM oneshotaccess")).one()[0]
|
|
assert datetime.fromisoformat(raw) == datetime(2029, 7, 27, 10, 0), (
|
|
f"unexpected value in the database: {raw}"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
|
|
def test_oneshot_without_offset_is_treated_as_local_time(
|
|
client, auth_headers, db_session, system_tz
|
|
):
|
|
"""A datetime without an offset is wall-clock time on the host, not UTC."""
|
|
response = _post_oneshot_aa(
|
|
client, auth_headers, "naive os_AA", "2029-07-27T12:00:00"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# 2029-07-27 is summer in Berlin, so 12:00 local is 10:00 UTC.
|
|
parsed = datetime.fromisoformat(response.json()["oneshot"]["ends_at"])
|
|
assert parsed == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
|
|
|
|
db_session.expire_all()
|
|
stored = db_session.exec(select(OneShotAccess)).one()
|
|
assert stored.ends_at == datetime(2029, 7, 27, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
|
|
def test_oneshot_date_only_is_local_midnight(client, auth_headers, system_tz):
|
|
"""A bare date is local midnight, which is not the same instant as 00:00 UTC."""
|
|
response = _post_oneshot_aa(client, auth_headers, "date os_AA", "2029-07-27")
|
|
assert response.status_code == 200
|
|
|
|
ends_at = response.json()["oneshot"]["ends_at"]
|
|
parsed = datetime.fromisoformat(ends_at)
|
|
assert parsed == datetime(2029, 7, 26, 22, 0, tzinfo=UTC), (
|
|
f"local midnight was not converted to UTC: {ends_at}"
|
|
)
|
|
|
|
|
|
def test_update_access_auth_os_returns_timezone(client, auth_headers, db_session):
|
|
"""PATCHing the oneshot keeps the timezone in the response and in the DB."""
|
|
created = _post_oneshot_aa(
|
|
client, auth_headers, "tz os_AA", "2029-07-27T12:00:00+02:00"
|
|
)
|
|
assert created.status_code == 200
|
|
aa_id = created.json()["id"]
|
|
|
|
response = client.patch(
|
|
f"/api/v1/aa/{aa_id}",
|
|
json={"oneshot": {"uses": 2, "ends_at": "2030-01-01T05:00:00+05:00"}},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
ends_at = response.json()["oneshot"]["ends_at"]
|
|
parsed = datetime.fromisoformat(ends_at)
|
|
assert parsed.utcoffset() is not None, f"response has no timezone: {ends_at}"
|
|
assert parsed == datetime(2030, 1, 1, 0, 0, tzinfo=UTC)
|
|
|
|
db_session.expire_all()
|
|
stored = db_session.exec(select(OneShotAccess)).one()
|
|
assert stored.ends_at == datetime(2030, 1, 1, 0, 0, tzinfo=UTC)
|
|
|
|
|
|
@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
|
|
def test_oneshot_local_time_uses_the_dst_rule_of_its_own_date(
|
|
client, auth_headers, system_tz
|
|
):
|
|
"""Naive local times must pick the offset in force on that date.
|
|
|
|
Without this the host's *current* offset gets applied to every value, so a
|
|
summer date entered in winter is stored an hour off.
|
|
"""
|
|
summer = _post_oneshot_aa(
|
|
client, auth_headers, "summer os_AA", "2029-07-27T12:00:00"
|
|
)
|
|
winter = _post_oneshot_aa(
|
|
client, auth_headers, "winter os_AA", "2029-12-27T12:00:00"
|
|
)
|
|
assert summer.status_code == 200 and winter.status_code == 200
|
|
|
|
# CEST is UTC+2, CET is UTC+1.
|
|
assert datetime.fromisoformat(summer.json()["oneshot"]["ends_at"]) == datetime(
|
|
2029, 7, 27, 10, 0, tzinfo=UTC
|
|
)
|
|
assert datetime.fromisoformat(winter.json()["oneshot"]["ends_at"]) == datetime(
|
|
2029, 12, 27, 11, 0, tzinfo=UTC
|
|
)
|