From ff170af93514db29b635bb1a68a488f381fbe6dd Mon Sep 17 00:00:00 2001 From: ahtlon Date: Thu, 5 Nov 2026 12:13:46 +0100 Subject: [PATCH] Add timezone marker to requests Tests helpfully written by local-Qwen3.8 --- app/model/models.py | 31 ++++- app/services/door.py | 13 +- test/conftest.py | 22 ++++ test/test_services/test_aa_manager.py | 168 ++++++++++++++++++++++++++ test/test_services/test_door.py | 84 +++++++++++++ 5 files changed, 313 insertions(+), 5 deletions(-) diff --git a/app/model/models.py b/app/model/models.py index 6935e13..e951e7d 100644 --- a/app/model/models.py +++ b/app/model/models.py @@ -1,7 +1,8 @@ -from datetime import datetime, time +from datetime import UTC, datetime, time from typing import Literal from pydantic import model_validator +from sqlalchemy import DateTime, TypeDecorator from sqlmodel import Field, Relationship, SQLModel @@ -9,6 +10,32 @@ class Base(SQLModel): pass +#### Extra Column Types +class UTCDateTime(TypeDecorator): + """Modified datetime for timezones + Input time gets read as given tz or local tz, then converted to UTC + Output everything as UTC + """ + + impl = DateTime(timezone=True) + LOCAL_TIMEZONE = datetime.now(UTC).astimezone().tzinfo + cache_ok = True + + def process_bind_param(self, value: datetime | None, dialect) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + value = value.astimezone(self.LOCAL_TIMEZONE) + return value.astimezone(UTC) + + def process_result_value(self, value: datetime | None, dialect) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + #### User class UserBase(Base): name: str = Field(index=True, unique=True) @@ -102,7 +129,7 @@ class AccessAuthorizationDB(AccessAuthorizationBase, table=True): class OneShotAccessBase(Base): uses: int = 1 - ends_at: datetime + ends_at: datetime = Field(sa_type=UTCDateTime) class OneShotAccess(OneShotAccessBase, table=True): diff --git a/app/services/door.py b/app/services/door.py index 316ad2e..1d2a00b 100644 --- a/app/services/door.py +++ b/app/services/door.py @@ -116,7 +116,7 @@ def checkAccess(key: str, db: Session): return True current_weekday = datetime.weekday(date.today()) - current_time = datetime.now() + current_time = datetime.now().astimezone() for auth in card.group.accessauths: logger.info(f"checking auth: {auth.name}") if not auth.is_active: @@ -129,7 +129,11 @@ def checkAccess(key: str, db: Session): f" weekday: CUR:{current_weekday} TT:{timetable.weekday}" ) if current_weekday == timetable.weekday: - starttime = datetime.combine(date.today(), timetable.starttime) + starttime = datetime.combine( + date.today(), + timetable.starttime, + tzinfo=current_time.tzinfo, + ) endtime = starttime + timedelta(minutes=timetable.duration) logger.info( f" comparing time: Start:{starttime} " @@ -140,7 +144,10 @@ def checkAccess(key: str, db: Session): return True if auth.type == "oneshot": logger.info(f" oneshot auth found: {auth.oneshot}") - if current_time < auth.oneshot.ends_at and auth.oneshot.uses > 0: + ends_at = auth.oneshot.ends_at + if ends_at.tzinfo is None: + ends_at = ends_at.astimezone() + if current_time < ends_at and auth.oneshot.uses > 0: decrementOneshot(db, auth.oneshot) return True logger.info("No more auths found") diff --git a/test/conftest.py b/test/conftest.py index 8f12ee3..448eae0 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,5 +1,6 @@ import os from datetime import time +from time import tzset import pytest from fastapi.testclient import TestClient @@ -28,6 +29,27 @@ engine = create_engine( ) +@pytest.fixture +def system_tz(request): + """Run a test with the host-local timezone forced to a POSIX TZ string. + + POSIX strings (e.g. "CET-1CEST,M3.5.0,M10.5.0/3") are used instead of + zoneinfo names because glibc in this container does not resolve names such + as "Europe/Berlin". + """ + previous_tz = os.environ.get("TZ") + os.environ["TZ"] = request.param + tzset() + try: + yield request.param + finally: + if previous_tz is None: + os.environ.pop("TZ", None) + else: + os.environ["TZ"] = previous_tz + tzset() + + @pytest.fixture(scope="function") def db_session(): """Create a fresh database session for each test.""" diff --git a/test/test_services/test_aa_manager.py b/test/test_services/test_aa_manager.py index fdfce11..6c30130 100644 --- a/test/test_services/test_aa_manager.py +++ b/test/test_services/test_aa_manager.py @@ -1,3 +1,29 @@ +from datetime import UTC, datetime + +import pytest +from sqlalchemy import text +from sqlmodel import select + +from app.model.models import AccessAuthorizationDB, 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_tt(client, auth_headers): """Test creating a new access authorization.""" aa_data = { @@ -224,3 +250,145 @@ def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers): f"/api/v1/aa/assign/1/{test_aa_tt.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 + ) + diff --git a/test/test_services/test_door.py b/test/test_services/test_door.py index 0cabcf5..46dcce3 100644 --- a/test/test_services/test_door.py +++ b/test/test_services/test_door.py @@ -220,3 +220,87 @@ def test_regression_check_access_issue_28(db_session): db_session.commit() assert checkAccess("test-key-123", db_session) + + +# 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 _create_card_with_oneshot(db_session, ends_at): + """Create a card whose only authorization is a oneshot ending at ends_at.""" + group = GroupDB(name="Test Group") + db_session.add(group) + db_session.commit() + + card = Card( + key="test-key-123", + group_id=group.id, + enabled=True, + name="test_card", + card_serial="00:00:00:00:00:00:00", + ) + db_session.add(card) + + oneshot = OneShotAccess(uses=1, ends_at=ends_at) + db_session.add(oneshot) + + aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="oneshot") + db_session.add(aa) + aa.oneshot = oneshot + group.accessauths = [aa] + + db_session.commit() + + +@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True) +def test_check_access_oneshot_end_time_is_compared_in_utc(db_session, system_tz): + """A oneshot ending in 30 minutes is valid even though the local clock is + ahead of UTC: the stored end time is UTC and must not be read as local time.""" + ends_at = datetime.datetime.now(datetime.UTC) + datetime.timedelta(minutes=30) + _create_card_with_oneshot(db_session, ends_at) + + assert checkAccess("test-key-123", db_session) + + +@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True) +def test_check_access_oneshot_expired(db_session, system_tz): + """A oneshot that already ended must be rejected whatever the local clock is.""" + ends_at = datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=30) + _create_card_with_oneshot(db_session, ends_at) + + assert not checkAccess("test-key-123", db_session) + + +@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True) +def test_check_access_timetable_window_is_local_wall_clock(db_session, system_tz): + """Timetables are local wall-clock times and still work with an aware door clock.""" + now_local = datetime.datetime.now().astimezone() + + group = GroupDB(name="Test Group") + db_session.add(group) + db_session.commit() + + card = Card( + key="test-key-123", + group_id=group.id, + enabled=True, + name="test_card", + card_serial="00:00:00:00:00:00:00", + ) + db_session.add(card) + + timetable = Timetable( + weekday=now_local.weekday(), + starttime=now_local.time(), + duration=120, # 2 hours + ) + db_session.add(timetable) + + aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable") + db_session.add(aa) + aa.timetables = [timetable] + group.accessauths = [aa] + + db_session.commit() + + assert checkAccess("test-key-123", db_session)