Wunschliste 1 #36
@@ -0,0 +1,86 @@
|
||||
"""card last_used column and non-nullable card.group_id
|
||||
|
||||
Revision ID: b7c1d2e3f4a5
|
||||
Revises: 93e8dda267e5
|
||||
Create Date: 2026-11-10 12:15:04.113902
|
||||
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "b7c1d2e3f4a5"
|
||||
down_revision: str | Sequence[str] | None = "93e8dda267e5"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Apply the two card model changes to databases that predate them.
|
||||
|
||||
create_db_and_tables() only creates and stamps tables while the database is
|
||||
empty, so a deployment whose card table was built from the old model keeps a
|
||||
nullable group_id and no last_used column until `alembic upgrade head`.
|
||||
That now breaks the app: every read of Card serializes `last_used`, which
|
||||
fails with 'no such column: card.last_used', and a NULL group_id can still
|
||||
be written, which is what makes door.checkAccess() crash on card.group.name
|
||||
(app/services/door.py:117).
|
||||
|
||||
SQLite cannot add a NOT NULL column to a table that already has rows, cannot
|
||||
add a column with a non-constant default, and cannot make a nullable column
|
||||
NOT NULL in place, so the column is added nullable, filled, and only then
|
||||
tightened inside op.batch_alter_table. No server_default is used anywhere:
|
||||
batch_alter_table would leave `DEFAULT CURRENT_TIMESTAMP` in the recreated
|
||||
table, a default the model does not have and which would silently paper over
|
||||
a card written without last_used.
|
||||
|
||||
The batch copy reflects the existing card table, so the unnamed
|
||||
`FOREIGN KEY(group_id) REFERENCES groupdb (id)` SQLite emitted back then is
|
||||
carried over unchanged. drop_constraint/create_foreign_key are deliberately
|
||||
absent: on such a table alembic fails with
|
||||
"No such constraint: 'card_group_id_fkey'", and on a table whose FK *is*
|
||||
named they would leave a second duplicate FK behind.
|
||||
"""
|
||||
op.execute(sa.text("ALTER TABLE card ADD COLUMN last_used DATETIME"))
|
||||
|
||||
# A NOT NULL column needs values; existing cards have no recorded scan yet.
|
||||
op.execute(sa.text("UPDATE card SET last_used = CURRENT_TIMESTAMP"))
|
||||
|
||||
# A NOT NULL column also needs values; delete the orphaned cards the nullable
|
||||
# FK allowed. groupManager.delete_group() now refuses to orphan a group, so
|
||||
# no card can be lost this way again.
|
||||
op.execute(sa.text("DELETE FROM card WHERE group_id IS NULL"))
|
||||
|
||||
with op.batch_alter_table("card") as batch_op:
|
||||
batch_op.alter_column(
|
||||
"last_used",
|
||||
existing_type=sa.String(),
|
||||
type_=sa.DateTime(),
|
||||
nullable=False,
|
||||
)
|
||||
batch_op.alter_column(
|
||||
"group_id",
|
||||
existing_type=sa.Integer(),
|
||||
type_=sa.Integer(),
|
||||
nullable=False,
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Make group_id nullable again and drop last_used.
|
||||
|
||||
The last_used timestamps and the deleted orphaned cards are gone for good;
|
||||
a downgrade only restores the schema.
|
||||
"""
|
||||
with op.batch_alter_table("card") as batch_op:
|
||||
batch_op.alter_column(
|
||||
"last_used", existing_type=sa.DateTime(), type_=sa.String(), nullable=True
|
||||
)
|
||||
batch_op.alter_column(
|
||||
"group_id", existing_type=sa.Integer(), type_=sa.Integer(), nullable=True
|
||||
)
|
||||
op.execute(sa.text("ALTER TABLE card DROP COLUMN last_used"))
|
||||
@@ -0,0 +1,43 @@
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from app.services.auth import auth_is_admin
|
||||
from app.services.caldav import get_caldav_handler
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
calendar_router = APIRouter(
|
||||
prefix="/api/v1/calendar",
|
||||
tags=["Calendar diagnostics"],
|
||||
dependencies=[Depends(auth_is_admin)],
|
||||
)
|
||||
|
||||
|
||||
@calendar_router.get("/isActive")
|
||||
def cal_is_active():
|
||||
return get_caldav_handler() is not None
|
||||
|
||||
|
||||
@calendar_router.get("/calGetActiveEvents")
|
||||
def cal_get_active_events():
|
||||
handler = get_caldav_handler()
|
||||
if handler is None:
|
||||
raise HTTPException(503, "Caldav handler not initialized")
|
||||
return handler.get_active_events()
|
||||
|
||||
|
||||
@calendar_router.get("/calGetFutureEvents")
|
||||
def cal_get_future_events():
|
||||
handler = get_caldav_handler()
|
||||
if handler is None:
|
||||
raise HTTPException(503, "Caldav handler not initialized")
|
||||
return handler.get_future_events()
|
||||
|
||||
|
||||
@calendar_router.get("/calGetActiveGroups")
|
||||
def cal_get_active_groups():
|
||||
handler = get_caldav_handler()
|
||||
if handler is None:
|
||||
raise HTTPException(503, "Caldav handler not initialized")
|
||||
return handler.get_active_groups()
|
||||
@@ -1,4 +1,5 @@
|
||||
import logging
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.exc import NoResultFound
|
||||
@@ -27,6 +28,7 @@ def register_card(cardInput: CardCreate):
|
||||
name=cardInput.name,
|
||||
card_serial=uid,
|
||||
enabled=cardInput.enabled,
|
||||
last_used=datetime.now(tz=UTC),
|
||||
)
|
||||
return card
|
||||
|
||||
@@ -116,3 +118,17 @@ def update_card(
|
||||
) from err
|
||||
db_card.sqlmodel_update(card_data)
|
||||
return add_and_refresh(db, db_card)
|
||||
|
||||
|
||||
@card_router.get("/inactive")
|
||||
def inactive_card_search(
|
||||
enabled: bool = False,
|
||||
uses: bool = False,
|
||||
ended: bool = False,
|
||||
db: Session = Depends(get_session),
|
||||
admin: bool = Depends(auth_is_admin),
|
||||
):
|
||||
if not enabled and not uses and not ended:
|
||||
return {"message": "No results. Please activate at least one filter rule"}
|
||||
if enabled:
|
||||
return {"message": "Results with disabled cards heres"}
|
||||
|
||||
@@ -6,7 +6,6 @@ from sqlmodel import Session, select
|
||||
|
||||
from app.model.models import Card
|
||||
from app.services.auth import auth_is_admin
|
||||
from app.services.caldav import get_caldav_handler
|
||||
from app.services.database import add_and_refresh, get_session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -64,27 +63,3 @@ def list_all_cards(db: Session = Depends(get_session)):
|
||||
for i in cards:
|
||||
out.append({i.key: i.group.name})
|
||||
return out
|
||||
|
||||
|
||||
@debug_router.get("/calGetActiveEvents")
|
||||
def cal_get_active_events(db: Session = Depends(get_session)):
|
||||
handler = get_caldav_handler()
|
||||
if handler is None:
|
||||
raise HTTPException(503, "Caldav handler not initialized")
|
||||
return handler.get_active_events()
|
||||
|
||||
|
||||
@debug_router.get("/calGetFutureEvents")
|
||||
def cal_get_future_events(db: Session = Depends(get_session)):
|
||||
handler = get_caldav_handler()
|
||||
if handler is None:
|
||||
raise HTTPException(503, "Caldav handler not initialized")
|
||||
return handler.get_future_events()
|
||||
|
||||
|
||||
@debug_router.get("/calGetActiveGroups")
|
||||
def cal_get_active_groups(db: Session = Depends(get_session)):
|
||||
handler = get_caldav_handler()
|
||||
if handler is None:
|
||||
raise HTTPException(503, "Caldav handler not initialized")
|
||||
return handler.get_active_groups()
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy import func
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from ..model.models import GroupCreate, GroupDB, GroupResponse
|
||||
from ..model.models import Card, GroupCreate, GroupDB, GroupResponse
|
||||
from ..services.auth import auth_is_admin
|
||||
from ..services.database import add_and_refresh, get_session
|
||||
|
||||
@@ -32,7 +33,10 @@ def create_group(
|
||||
return add_and_refresh(db, db_group)
|
||||
|
||||
|
||||
@group_router.delete("/{group_id}")
|
||||
@group_router.delete(
|
||||
"/{group_id}",
|
||||
responses={409: {"description": "The group still has cards assigned"}},
|
||||
)
|
||||
def delete_group(
|
||||
*,
|
||||
db: Session = Depends(get_session),
|
||||
@@ -42,6 +46,18 @@ def delete_group(
|
||||
db_group = db.get(GroupDB, group_id)
|
||||
if db_group is None:
|
||||
raise HTTPException(status_code=404, detail="Group not found")
|
||||
card_count = db.exec(
|
||||
select(func.count()).select_from(Card).where(Card.group_id == group_id)
|
||||
).one()
|
||||
if card_count:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail=(
|
||||
f"Group still has {card_count} card(s) assigned. "
|
||||
"Move or delete them before deleting the group."
|
||||
),
|
||||
)
|
||||
|
||||
db.delete(db_group)
|
||||
db.commit()
|
||||
return {"message": "Group deleted successfully"}
|
||||
|
||||
@@ -7,6 +7,7 @@ from fastapi.security import OAuth2PasswordBearer
|
||||
|
||||
from app.controllers import (
|
||||
aaManager,
|
||||
calendarManager,
|
||||
cardManager,
|
||||
debugManager,
|
||||
doorManager,
|
||||
@@ -75,4 +76,5 @@ app.include_router(groupManager.group_router)
|
||||
app.include_router(cardManager.card_router)
|
||||
app.include_router(aaManager.aa_router)
|
||||
app.include_router(doorManager.door_router)
|
||||
app.include_router(calendarManager.calendar_router)
|
||||
app.include_router(debugManager.debug_router)
|
||||
|
||||
+2
-1
@@ -168,8 +168,9 @@ class Card(Base, table=True):
|
||||
card_serial: str
|
||||
enabled: bool = True
|
||||
name: str = Field(unique=True, max_length=32)
|
||||
group_id: int | None = Field(default=None, foreign_key="groupdb.id")
|
||||
group_id: int = Field(foreign_key="groupdb.id")
|
||||
group: GroupDB | None = Relationship(back_populates="cards")
|
||||
last_used: datetime = Field(sa_type=UTCDateTime)
|
||||
|
||||
|
||||
class CardCreate(Base):
|
||||
|
||||
@@ -128,11 +128,11 @@ class CaldavEventCache:
|
||||
|
||||
if event["all_day"]: # noqa: SIM108
|
||||
# All-day event end dates are exclusive.
|
||||
is_active = start <= today
|
||||
is_pending = start > today
|
||||
else:
|
||||
is_active = start <= active_time
|
||||
is_pending = start > active_time
|
||||
|
||||
if is_active:
|
||||
if is_pending:
|
||||
future.append(event)
|
||||
|
||||
return future
|
||||
|
||||
+16
-4
@@ -1,5 +1,5 @@
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import exc
|
||||
from sqlmodel import select
|
||||
@@ -99,9 +99,17 @@ def decrementOneshot(db: Session, oneshot: OneShotAccess):
|
||||
add_and_refresh(db, oneshot)
|
||||
|
||||
|
||||
def updateLastUsed(db: Session, card: Card):
|
||||
data = card.model_dump()
|
||||
data["last_used"] = datetime.now(UTC)
|
||||
card.sqlmodel_update(card, update=data)
|
||||
add_and_refresh(db, card)
|
||||
|
||||
|
||||
def checkAccess(key: str, db: Session):
|
||||
try:
|
||||
card = db.exec(select(Card).where(Card.key == key)).one()
|
||||
updateLastUsed(db, card)
|
||||
if not card.enabled:
|
||||
logger.info("Card Inactive!")
|
||||
return False
|
||||
@@ -122,9 +130,13 @@ def checkAccess(key: str, db: Session):
|
||||
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
|
||||
if current_time < ends_at:
|
||||
if auth.oneshot.uses > 0:
|
||||
decrementOneshot(db, auth.oneshot)
|
||||
return True
|
||||
elif auth.oneshot.uses == -1:
|
||||
logger.debug("Unlimited use card found")
|
||||
return True
|
||||
logger.info("No more auths found")
|
||||
return False
|
||||
except exc.NoResultFound as err:
|
||||
|
||||
@@ -217,6 +217,7 @@ def test_card(db_session, test_group):
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
last_used=datetime.now(),
|
||||
)
|
||||
db_session.add(card)
|
||||
db_session.commit()
|
||||
|
||||
@@ -0,0 +1,571 @@
|
||||
from datetime import UTC, date, datetime, time, timedelta
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import exc as sa_exc
|
||||
|
||||
from app.main import app
|
||||
from app.services.caldav import CaldavEventCache
|
||||
|
||||
# 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"
|
||||
|
||||
TIMED_START = datetime(2026, 11, 2, 10, 0, tzinfo=UTC)
|
||||
TIMED_END = datetime(2026, 11, 2, 11, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
def _cache(events=None):
|
||||
"""Build a CaldavEventCache without calling __init__ (no CalDAV server needed)."""
|
||||
cache = CaldavEventCache.__new__(CaldavEventCache)
|
||||
cache.url = "http://caldav.invalid/dav.php"
|
||||
cache.username = "u"
|
||||
cache.password = "p"
|
||||
cache.calendarNames = ["c"]
|
||||
cache.is_running = False
|
||||
cache.thread = None
|
||||
cache._stopped = __import__("threading").Event()
|
||||
cache.events = (
|
||||
events
|
||||
if events is not None
|
||||
else [
|
||||
{
|
||||
"uid": "1",
|
||||
"summary": "Timed event",
|
||||
"groups": ["team-a", "shared"],
|
||||
"start": TIMED_START,
|
||||
"end": TIMED_END,
|
||||
"all_day": False,
|
||||
},
|
||||
{
|
||||
"uid": "2",
|
||||
"summary": "All-day event",
|
||||
"groups": ["team-b", "shared"],
|
||||
"start": date(2026, 11, 2),
|
||||
"end": date(2026, 11, 3),
|
||||
"all_day": True,
|
||||
},
|
||||
]
|
||||
)
|
||||
return cache
|
||||
|
||||
|
||||
# ------------------------------------------------------------- calendar router
|
||||
|
||||
|
||||
def test_calendar_routes_are_registered():
|
||||
paths = [route.path for route in app.routes]
|
||||
assert any("/calendar" in path for path in paths), paths
|
||||
|
||||
|
||||
def test_calendar_is_active_requires_admin(client):
|
||||
response = client.get("/api/v1/calendar/isActive")
|
||||
assert response.status_code == 401, response.text
|
||||
|
||||
|
||||
def test_calendar_is_active_reports_disabled_handler(client, auth_headers):
|
||||
"""With DISABLE_CALDAV=True the handler is None, so isActive must be false."""
|
||||
response = client.get("/api/v1/calendar/isActive", headers=auth_headers)
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json() is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"/api/v1/calendar/calGetActiveEvents",
|
||||
"/api/v1/calendar/calGetFutureEvents",
|
||||
"/api/v1/calendar/calGetActiveGroups",
|
||||
],
|
||||
)
|
||||
def test_calendar_diagnostics_without_handler_are_503(client, auth_headers, path):
|
||||
response = client.get(path, headers=auth_headers)
|
||||
assert response.status_code == 503, response.text
|
||||
|
||||
|
||||
# ------------------------------------------------------------- event evaluation
|
||||
|
||||
|
||||
def test_active_events_only_within_the_window():
|
||||
cache = _cache()
|
||||
before = datetime(2026, 11, 2, 9, 59, tzinfo=UTC)
|
||||
during = datetime(2026, 11, 2, 10, 30, tzinfo=UTC)
|
||||
|
||||
assert [e["uid"] for e in cache.get_active_events(before)] == ["2"]
|
||||
assert {e["uid"] for e in cache.get_active_events(during)} == {"1", "2"}
|
||||
|
||||
|
||||
def test_active_groups_are_flattened_and_unique():
|
||||
cache = _cache()
|
||||
during = datetime(2026, 11, 2, 10, 30, tzinfo=UTC)
|
||||
assert cache.get_active_groups(during) == {"team-a", "team-b", "shared"}
|
||||
|
||||
|
||||
def test_active_groups_empty_when_no_event_runs():
|
||||
cache = _cache()
|
||||
later = datetime(2026, 11, 4, 10, 30, tzinfo=UTC)
|
||||
assert cache.get_active_groups(later) == set()
|
||||
|
||||
|
||||
def test_future_events_excludes_events_that_already_started():
|
||||
today = date.today()
|
||||
cache = _cache(
|
||||
events=[
|
||||
{
|
||||
"uid": "timed",
|
||||
"summary": "Timed event today",
|
||||
"groups": ["team-a"],
|
||||
"start": datetime.combine(today, time(10, 0), tzinfo=UTC),
|
||||
"end": datetime.combine(today, time(11, 0), tzinfo=UTC),
|
||||
"all_day": False,
|
||||
},
|
||||
{
|
||||
"uid": "all-day",
|
||||
"summary": "All-day event today",
|
||||
"groups": ["team-b"],
|
||||
"start": today,
|
||||
"end": today + timedelta(days=1),
|
||||
"all_day": True,
|
||||
},
|
||||
]
|
||||
)
|
||||
later_today = datetime.combine(today, time(10, 30), tzinfo=UTC)
|
||||
tomorrow = datetime.combine(today + timedelta(days=1), time(12, 0), tzinfo=UTC)
|
||||
|
||||
# Nothing is pending: the timed event is in progress and the all-day event
|
||||
# already started on its start date, although both are still running.
|
||||
assert {e["uid"] for e in cache.get_future_events(later_today)} == set()
|
||||
# The day after the all-day event ended (its end date is exclusive).
|
||||
assert {e["uid"] for e in cache.get_future_events(tomorrow)} == set()
|
||||
|
||||
|
||||
def test_future_events_start_boundary_is_exclusive():
|
||||
today = date.today()
|
||||
cache = _cache(
|
||||
events=[
|
||||
{
|
||||
"uid": "timed",
|
||||
"summary": "Timed event today",
|
||||
"groups": ["team-a"],
|
||||
"start": datetime.combine(today, time(10, 0), tzinfo=UTC),
|
||||
"end": datetime.combine(today, time(11, 0), tzinfo=UTC),
|
||||
"all_day": False,
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
at_start = datetime.combine(today, time(10, 0), tzinfo=UTC)
|
||||
one_minute_earlier = datetime.combine(today, time(9, 59), tzinfo=UTC)
|
||||
|
||||
# The event is running at its own start, it is not future anymore.
|
||||
assert {e["uid"] for e in cache.get_future_events(at_start)} == set()
|
||||
assert {e["uid"] for e in cache.get_future_events(one_minute_earlier)} == {"timed"}
|
||||
|
||||
|
||||
def test_all_day_event_compares_against_aware_time(db_session):
|
||||
"""get_active_events() must handle date-valued all-day events alongside
|
||||
an aware reference time (this is the path door.checkAccess() uses)."""
|
||||
aware_now = datetime(2026, 11, 2, 10, 30, tzinfo=UTC)
|
||||
cache = _cache()
|
||||
assert cache.is_group_active("team-b", aware_now)
|
||||
|
||||
|
||||
def test_future_events_reports_events_later_today():
|
||||
"""A timed event later on the same day must show up (this needs the full
|
||||
datetime comparison, not the date of the reference time).
|
||||
|
||||
Anchored on the wall clock instead of a fixed date so the assertion holds
|
||||
whenever the suite runs.
|
||||
"""
|
||||
now = datetime.now(UTC)
|
||||
cache = _cache(
|
||||
events=[
|
||||
{
|
||||
"uid": "earlier-today",
|
||||
"summary": "Already over",
|
||||
"groups": ["team-e"],
|
||||
"start": now - timedelta(hours=3),
|
||||
"end": now - timedelta(hours=2),
|
||||
"all_day": False,
|
||||
},
|
||||
{
|
||||
"uid": "later-today",
|
||||
"summary": "Starts in 5 hours",
|
||||
"groups": ["team-l"],
|
||||
"start": now + timedelta(hours=5),
|
||||
"end": now + timedelta(hours=6),
|
||||
"all_day": False,
|
||||
},
|
||||
{
|
||||
"uid": "tomorrow",
|
||||
"summary": "Starts tomorrow",
|
||||
"groups": ["team-t"],
|
||||
"start": now + timedelta(days=1),
|
||||
"end": now + timedelta(days=1, hours=1),
|
||||
"all_day": False,
|
||||
},
|
||||
]
|
||||
)
|
||||
assert {e["uid"] for e in cache.get_future_events(now)} == {
|
||||
"later-today",
|
||||
"tomorrow",
|
||||
}
|
||||
|
||||
|
||||
def test_future_events_only_lists_all_day_events_starting_later():
|
||||
today = date.today()
|
||||
cache = _cache(
|
||||
events=[
|
||||
{
|
||||
"uid": "all-day-running",
|
||||
"summary": "Running all day",
|
||||
"groups": ["team-now"],
|
||||
"start": today,
|
||||
"end": today + timedelta(days=1),
|
||||
"all_day": True,
|
||||
},
|
||||
{
|
||||
"uid": "all-day-later",
|
||||
"summary": "Starts in a week",
|
||||
"groups": ["team-soon"],
|
||||
"start": today + timedelta(days=7),
|
||||
"end": today + timedelta(days=8),
|
||||
"all_day": True,
|
||||
},
|
||||
]
|
||||
)
|
||||
assert {e["uid"] for e in cache.get_future_events()} == {"all-day-later"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"system_tz,expect_active",
|
||||
[("UTC", False), (BERLIN_TZ, True)],
|
||||
indirect=["system_tz"],
|
||||
)
|
||||
def test_naive_current_time_is_treated_as_local_time(
|
||||
db_session, system_tz, expect_active
|
||||
):
|
||||
"""get_active_events() reads a naive reference time in the system timezone:
|
||||
11:30 naive is outside the 10:00-11:00 UTC event when the host is UTC,
|
||||
but inside it when the host is CET (11:30 CET == 10:30 UTC)."""
|
||||
cache = _cache()
|
||||
naive_local = datetime(2026, 11, 2, 11, 30)
|
||||
uids = {e["uid"] for e in cache.get_active_events(naive_local)}
|
||||
assert ("1" in uids) is expect_active
|
||||
|
||||
|
||||
def test_naive_stored_times_are_rejected_by_comparison():
|
||||
"""Regression pin: refresh() converts naive event times with astimezone(),
|
||||
so the cache must never hold naive datetimes — comparing them against the
|
||||
aware 'now' raises TypeError inside door.checkAccess()."""
|
||||
cache = _cache(
|
||||
events=[
|
||||
{
|
||||
"uid": "naive-stored",
|
||||
"summary": "Naive event",
|
||||
"groups": ["naive-team"],
|
||||
"start": datetime(2026, 11, 2, 12, 0),
|
||||
"end": datetime(2026, 11, 2, 13, 0),
|
||||
"all_day": False,
|
||||
}
|
||||
]
|
||||
)
|
||||
with pytest.raises(TypeError, match="offset-naive and offset-aware"):
|
||||
cache.get_active_events(datetime(2026, 11, 2, 11, 30, tzinfo=UTC))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"system_tz,expected_utc",
|
||||
[
|
||||
("CET-1CEST,M3.5.0,M10.5.0/3", datetime(2026, 11, 2, 11, 0, tzinfo=UTC)),
|
||||
("UTC", datetime(2026, 11, 2, 12, 0, tzinfo=UTC)),
|
||||
],
|
||||
indirect=["system_tz"],
|
||||
)
|
||||
def test_naive_event_times_follow_the_system_timezone(
|
||||
system_tz, expected_utc, monkeypatch
|
||||
):
|
||||
"""refresh() turns a naive event time into UTC using the system timezone."""
|
||||
|
||||
class FakeCalendar:
|
||||
def search(self, **kwargs):
|
||||
class Comp:
|
||||
start = datetime(2026, 11, 2, 12, 0) # naive/floating time
|
||||
end = datetime(2026, 11, 2, 13, 0)
|
||||
|
||||
def get(self, key, default=""):
|
||||
return {"uid": "n", "summary": "s", "description": "t"}.get(
|
||||
key, default
|
||||
)
|
||||
|
||||
class Ev:
|
||||
component = Comp()
|
||||
|
||||
return [Ev()]
|
||||
|
||||
monkeypatch.setattr(
|
||||
"app.services.caldav.get_calendars", lambda **kw: [FakeCalendar()]
|
||||
)
|
||||
|
||||
cache = CaldavEventCache(
|
||||
url="http://caldav.invalid/dav.php",
|
||||
username="u",
|
||||
password="p",
|
||||
calendarNames=["c"],
|
||||
)
|
||||
cache.refresh()
|
||||
|
||||
assert len(cache.events) == 1
|
||||
assert cache.events[0]["start"] == expected_utc
|
||||
|
||||
|
||||
# ----------------------------------------------------------------- cache refresh
|
||||
|
||||
|
||||
def test_refresh_keeps_previous_events_on_empty_result(monkeypatch):
|
||||
"""An empty search result is treated as a connection error, not as 'no events'."""
|
||||
|
||||
class FakeCalendar:
|
||||
def search(self, **kwargs):
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(
|
||||
"app.services.caldav.get_calendars", lambda **kw: [FakeCalendar()]
|
||||
)
|
||||
|
||||
cache = _cache()
|
||||
previous = list(cache.events)
|
||||
cache.refresh()
|
||||
assert cache.events == previous
|
||||
|
||||
|
||||
def test_refresh_replaces_cache_with_search_result(monkeypatch):
|
||||
class FakeComponent:
|
||||
def __init__(self):
|
||||
self.start = datetime(2026, 11, 5, 9, 0, tzinfo=UTC)
|
||||
self.end = datetime(2026, 11, 5, 10, 0, tzinfo=UTC)
|
||||
|
||||
def get(self, key, default=""):
|
||||
return {"uid": "fresh", "summary": "Fresh", "description": "team-c"}.get(
|
||||
key, default
|
||||
)
|
||||
|
||||
class FakeEvent:
|
||||
component = FakeComponent()
|
||||
|
||||
class FakeCalendar:
|
||||
def search(self, **kwargs):
|
||||
return [FakeEvent()]
|
||||
|
||||
monkeypatch.setattr(
|
||||
"app.services.caldav.get_calendars", lambda **kw: [FakeCalendar()]
|
||||
)
|
||||
|
||||
cache = CaldavEventCache(
|
||||
url="http://caldav.invalid/dav.php",
|
||||
username="u",
|
||||
password="p",
|
||||
calendarNames=["c"],
|
||||
)
|
||||
cache.events = []
|
||||
cache.refresh()
|
||||
|
||||
assert [e["uid"] for e in cache.events] == ["fresh"]
|
||||
assert cache.events[0]["groups"] == ["team-c"]
|
||||
assert cache.events[0]["all_day"] is False
|
||||
|
||||
|
||||
def test_refresh_failure_keeps_cache_and_logs(monkeypatch, caplog):
|
||||
def boom(**kwargs):
|
||||
raise ConnectionError("caldav unreachable")
|
||||
|
||||
monkeypatch.setattr("app.services.caldav.get_calendars", boom)
|
||||
|
||||
cache = _cache()
|
||||
previous = list(cache.events)
|
||||
with pytest.raises(ConnectionError):
|
||||
cache.refresh()
|
||||
assert cache.events == previous
|
||||
|
||||
|
||||
def test_refresh_loop_survives_errors_and_stops_promptly(monkeypatch):
|
||||
"""stop() must not wait for the refresh interval to elapse."""
|
||||
|
||||
def boom(*args, **kwargs):
|
||||
raise ConnectionError("caldav unreachable")
|
||||
|
||||
monkeypatch.setattr("app.services.caldav.get_calendars", boom)
|
||||
monkeypatch.setattr("app.services.caldav.settings.caldav_refresh_interval", 60)
|
||||
|
||||
cache = CaldavEventCache(
|
||||
url="http://caldav.invalid/dav.php",
|
||||
username="u",
|
||||
password="p",
|
||||
calendarNames=["c"],
|
||||
)
|
||||
cache.start()
|
||||
assert cache.thread.is_alive()
|
||||
|
||||
cache.stop()
|
||||
|
||||
assert cache.is_running is False
|
||||
assert cache.thread is None or not cache.thread.is_alive()
|
||||
|
||||
|
||||
def test_stop_without_start_is_safe():
|
||||
cache = CaldavEventCache(
|
||||
url="http://caldav.invalid/dav.php",
|
||||
username="u",
|
||||
password="p",
|
||||
calendarNames=["c"],
|
||||
)
|
||||
cache.stop()
|
||||
assert cache.is_running is False
|
||||
|
||||
|
||||
# ------------------------------------------------------------------- door hook
|
||||
|
||||
|
||||
def test_door_uses_caldav_handler_when_present(db_session, monkeypatch):
|
||||
"""A group that is active in the calendar gets access without any AA."""
|
||||
import app.services.door as door_service
|
||||
from app.model.models import Card, GroupDB
|
||||
from app.services.door import checkAccess
|
||||
|
||||
group = GroupDB(name="Calendar Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
card = Card(
|
||||
key="cal-key",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="cal_card",
|
||||
card_serial="00:00:00:00:00:00:03",
|
||||
last_used=datetime.now(UTC) - timedelta(days=1),
|
||||
)
|
||||
db_session.add(card)
|
||||
db_session.commit()
|
||||
|
||||
# The card's group is not part of the fixture events; make it active instead.
|
||||
active_cache = _cache(
|
||||
events=[
|
||||
{
|
||||
"uid": "cal",
|
||||
"summary": "Calendar access",
|
||||
"groups": ["Calendar Group"],
|
||||
"start": datetime.now(UTC) - timedelta(minutes=5),
|
||||
"end": datetime.now(UTC) + timedelta(minutes=30),
|
||||
"all_day": False,
|
||||
}
|
||||
]
|
||||
)
|
||||
monkeypatch.setattr(door_service, "get_caldav_handler", lambda: active_cache)
|
||||
|
||||
assert checkAccess("cal-key", db_session)
|
||||
|
||||
|
||||
def test_door_denies_when_calendar_handler_has_no_matching_group(
|
||||
db_session, monkeypatch
|
||||
):
|
||||
import app.services.door as door_service
|
||||
from app.model.models import Card, GroupDB
|
||||
from app.services.door import checkAccess
|
||||
|
||||
group = GroupDB(name="Not In Calendar")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
card = Card(
|
||||
key="cal-key-2",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="cal_card_2",
|
||||
card_serial="00:00:00:00:00:00:04",
|
||||
last_used=datetime.now(UTC) - timedelta(days=1),
|
||||
)
|
||||
db_session.add(card)
|
||||
db_session.commit()
|
||||
|
||||
monkeypatch.setattr(door_service, "get_caldav_handler", lambda: _cache())
|
||||
|
||||
assert not checkAccess("cal-key-2", db_session)
|
||||
|
||||
|
||||
def test_door_check_access_without_caldav_handler(db_session, monkeypatch):
|
||||
"""DISABLE_CALDAV=True leaves the handler None; access checks must still work."""
|
||||
import app.services.door as door_service
|
||||
from app.model.models import AccessAuthorizationDB, Card, GroupDB, OneShotAccess
|
||||
from app.services.door import checkAccess
|
||||
|
||||
monkeypatch.setattr(door_service, "get_caldav_handler", lambda: None)
|
||||
|
||||
group = GroupDB(name="NoCal Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
card = Card(
|
||||
key="no-cal-key",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="no_cal_card",
|
||||
card_serial="00:00:00:00:00:00:05",
|
||||
last_used=datetime.now(UTC),
|
||||
)
|
||||
db_session.add(card)
|
||||
oneshot = OneShotAccess(uses=1, ends_at=datetime.now(UTC) + timedelta(days=1))
|
||||
db_session.add(oneshot)
|
||||
aa = AccessAuthorizationDB(name="NoCal AA", is_active=True, type="oneshot")
|
||||
db_session.add(aa)
|
||||
aa.oneshot = oneshot
|
||||
group.accessauths = [aa]
|
||||
db_session.commit()
|
||||
|
||||
assert checkAccess("no-cal-key", db_session)
|
||||
|
||||
|
||||
def test_door_card_without_a_group_is_refused_by_the_database(db_session, monkeypatch):
|
||||
"""Guard for the calendar lookup on card.group.name in checkAccess(): a card
|
||||
without a group can no longer exist.
|
||||
|
||||
Used to be reproducible with group_id=None, which made checkAccess() crash
|
||||
with AttributeError: 'NoneType' object has no attribute 'name'. The NOT NULL
|
||||
card.group_id column (model plus migration b7c1d2e3f4a5) and the 409 on
|
||||
DELETE /api/v1/groups/{id} both close that path, so the lookup is now
|
||||
exercised with a card that really has a group instead.
|
||||
"""
|
||||
import app.services.door as door_service
|
||||
from app.model.models import Card, GroupDB
|
||||
from app.services.door import checkAccess
|
||||
|
||||
group = GroupDB(name="Guard Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
|
||||
card = Card(
|
||||
key="orphan-key",
|
||||
group_id=None,
|
||||
enabled=True,
|
||||
name="orphan_card",
|
||||
card_serial="00:00:00:00:00:00:06",
|
||||
last_used=datetime.now(UTC),
|
||||
)
|
||||
db_session.add(card)
|
||||
with pytest.raises(sa_exc.IntegrityError) as excinfo:
|
||||
db_session.commit()
|
||||
assert "NOT NULL constraint failed: card.group_id" in str(excinfo.value)
|
||||
db_session.rollback()
|
||||
|
||||
monkeypatch.setattr(door_service, "get_caldav_handler", lambda: None)
|
||||
attached = Card(
|
||||
key="attached-key",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="attached_card",
|
||||
card_serial="00:00:00:00:00:00:07",
|
||||
last_used=datetime(2020, 1, 1, tzinfo=UTC),
|
||||
)
|
||||
db_session.add(attached)
|
||||
db_session.commit()
|
||||
|
||||
# The calendar lookup dereferences card.group.name; with a real group it
|
||||
# resolves instead of raising AttributeError, and a group without any
|
||||
# authorization simply denies access.
|
||||
assert not checkAccess("attached-key", db_session)
|
||||
db_session.refresh(attached)
|
||||
assert attached.last_used > datetime(2020, 1, 1, tzinfo=UTC)
|
||||
@@ -0,0 +1,203 @@
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from app.model.models import AccessAuthorizationDB, Card, GroupDB, OneShotAccess
|
||||
from app.services.door import checkAccess
|
||||
|
||||
# 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_group_with_card(db_session, *, enabled=True):
|
||||
"""Create a group with a single card and return the card."""
|
||||
group = GroupDB(name="LastUsed Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
|
||||
card = Card(
|
||||
key="last-used-key",
|
||||
group_id=group.id,
|
||||
enabled=enabled,
|
||||
name="last_used_card",
|
||||
card_serial="00:00:00:00:00:00:01",
|
||||
last_used=datetime(2020, 1, 1, tzinfo=UTC),
|
||||
)
|
||||
db_session.add(card)
|
||||
db_session.commit()
|
||||
db_session.refresh(card)
|
||||
return card
|
||||
|
||||
|
||||
def _attach_oneshot(db_session, card, *, uses=1, ends_in=timedelta(days=1)):
|
||||
"""Attach an active oneshot authorization to the card's group."""
|
||||
oneshot = OneShotAccess(uses=uses, ends_at=datetime.now(UTC) + ends_in)
|
||||
db_session.add(oneshot)
|
||||
aa = AccessAuthorizationDB(name="LastUsed AA", is_active=True, type="oneshot")
|
||||
db_session.add(aa)
|
||||
aa.oneshot = oneshot
|
||||
card.group.accessauths = [aa]
|
||||
db_session.commit()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- register_card
|
||||
|
||||
|
||||
def test_register_card_sets_last_used(monkeypatch):
|
||||
"""register_card() must stamp last_used instead of leaving the field unset."""
|
||||
from app.controllers import cardManager
|
||||
from app.controllers.cardManager import register_card
|
||||
from app.model.models import CardCreate
|
||||
|
||||
monkeypatch.setattr(
|
||||
cardManager, "WriteNewCard", lambda: ("k", "00:00:00:00:00:00:00")
|
||||
)
|
||||
|
||||
payload = CardCreate(name="reg_card", group_id=1, enabled=True)
|
||||
before = datetime.now(UTC) - timedelta(seconds=5)
|
||||
|
||||
card = register_card(payload)
|
||||
|
||||
assert card.last_used is not None, "register_card left last_used unset"
|
||||
assert card.last_used.tzinfo is not None, "last_used must be timezone aware"
|
||||
assert card.last_used >= before, "last_used is not the registration time"
|
||||
|
||||
|
||||
def test_register_card_last_used_survives_the_db(
|
||||
monkeypatch, client, auth_headers, test_group
|
||||
):
|
||||
"""A card registered through POST /api/v1/cards/ keeps a readable last_used."""
|
||||
from app.controllers import cardManager
|
||||
|
||||
monkeypatch.setattr(
|
||||
cardManager,
|
||||
"WriteNewCard",
|
||||
lambda: ("reg-key-1", "00:11:22:33:44:55:66"),
|
||||
)
|
||||
|
||||
before = datetime.now(UTC) - timedelta(seconds=5)
|
||||
response = client.post(
|
||||
"/api/v1/cards/",
|
||||
json={"name": "api_reg_card", "group_id": test_group.id, "enabled": True},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
body = response.json()
|
||||
assert body["key"] == "reg-key-1"
|
||||
last_used = datetime.fromisoformat(body["last_used"])
|
||||
assert last_used.tzinfo is not None, f"last_used not tz aware: {body['last_used']}"
|
||||
assert last_used >= before
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ checkAccess
|
||||
|
||||
|
||||
def test_check_access_updates_last_used(db_session):
|
||||
card = _create_group_with_card(db_session)
|
||||
_attach_oneshot(db_session, card)
|
||||
before_check = datetime.now(UTC)
|
||||
|
||||
assert checkAccess("last-used-key", db_session)
|
||||
|
||||
db_session.refresh(card)
|
||||
assert card.last_used >= before_check, (
|
||||
f"last_used stayed {card.last_used!r} after a successful scan"
|
||||
)
|
||||
|
||||
|
||||
def test_check_access_keeps_last_used_readable(db_session):
|
||||
"""Whatever checkAccess does to last_used, the value must stay timezone aware."""
|
||||
card = _create_group_with_card(db_session)
|
||||
_attach_oneshot(db_session, card)
|
||||
|
||||
assert checkAccess("last-used-key", db_session)
|
||||
|
||||
db_session.refresh(card)
|
||||
assert card.last_used.tzinfo is not None
|
||||
|
||||
|
||||
def test_check_access_records_scan_of_disabled_card(db_session):
|
||||
"""A disabled card is still scanned, so last_used must move to now.
|
||||
|
||||
Documents the intended semantics of updating last_used *before* the enabled
|
||||
check in checkAccess(). Fails together with the updateLastUsed() bug.
|
||||
"""
|
||||
card = _create_group_with_card(db_session, enabled=False)
|
||||
before_check = datetime.now(UTC)
|
||||
|
||||
assert not checkAccess("last-used-key", db_session)
|
||||
|
||||
db_session.refresh(card)
|
||||
assert card.last_used >= before_check
|
||||
|
||||
|
||||
# --------------------------------------------------- unlimited oneshot (uses=-1)
|
||||
|
||||
|
||||
def test_check_access_unlimited_oneshot(db_session):
|
||||
"""uses == -1 means unlimited: several scans in a row must be granted and the
|
||||
counter must not be touched."""
|
||||
card = _create_group_with_card(db_session)
|
||||
_attach_oneshot(db_session, card, uses=-1)
|
||||
|
||||
assert checkAccess("last-used-key", db_session)
|
||||
assert checkAccess("last-used-key", db_session)
|
||||
assert checkAccess("last-used-key", db_session)
|
||||
|
||||
oneshot = card.group.accessauths[0].oneshot
|
||||
db_session.refresh(oneshot)
|
||||
assert oneshot.uses == -1
|
||||
|
||||
|
||||
def test_check_access_oneshot_exhausted(db_session):
|
||||
card = _create_group_with_card(db_session)
|
||||
_attach_oneshot(db_session, card, uses=1)
|
||||
|
||||
assert checkAccess("last-used-key", db_session)
|
||||
assert not checkAccess("last-used-key", db_session)
|
||||
|
||||
|
||||
# ------------------------------------------------------------ GET /cards/inactive
|
||||
|
||||
|
||||
def test_cards_inactive_route_is_reachable(client, auth_headers):
|
||||
response = client.get("/api/v1/cards/inactive", headers=auth_headers)
|
||||
assert response.status_code == 200, response.text
|
||||
|
||||
|
||||
def test_cards_inactive_is_not_a_group_id(client, auth_headers):
|
||||
"""Regression guard: today the endpoint is swallowed by /{group_id}."""
|
||||
response = client.get("/api/v1/cards/inactive", headers=auth_headers)
|
||||
assert response.status_code == 422
|
||||
detail = response.json()["detail"]
|
||||
assert detail[0]["loc"] == ["path", "group_id"]
|
||||
|
||||
|
||||
# ------------------------------------------------------------- UTCDateTime column
|
||||
|
||||
|
||||
@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
|
||||
def test_card_last_used_naive_input_is_read_as_local_time(db_session, system_tz):
|
||||
"""A naive datetime is interpreted in the system timezone and read back as UTC."""
|
||||
group = GroupDB(name="TZ Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
|
||||
naive_local_noon = datetime(2026, 1, 15, 12, 0) # 12:00 CET == 11:00 UTC
|
||||
card = Card(
|
||||
key="tz-card",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="tz_card",
|
||||
card_serial="00:00:00:00:00:00:02",
|
||||
last_used=naive_local_noon,
|
||||
)
|
||||
db_session.add(card)
|
||||
db_session.commit()
|
||||
db_session.expire_all()
|
||||
|
||||
stored = db_session.get(Card, card.id)
|
||||
assert stored.last_used.tzinfo is not None
|
||||
assert stored.last_used.utcoffset() == timedelta(0)
|
||||
assert stored.last_used == datetime(2026, 1, 15, 11, 0, tzinfo=UTC)
|
||||
@@ -1,9 +1,14 @@
|
||||
import re
|
||||
|
||||
import pytest
|
||||
from alembic.config import Config
|
||||
from alembic.script import ScriptDirectory
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlmodel import Session
|
||||
from sqlalchemy import Table, inspect, text
|
||||
from sqlalchemy.exc import OperationalError
|
||||
from sqlmodel import Session, SQLModel, create_engine, select
|
||||
|
||||
from app.model.models import UserDB
|
||||
from alembic import command
|
||||
from app.model.models import Card, UserDB
|
||||
from app.services.database import (
|
||||
add_and_refresh,
|
||||
configure_engine,
|
||||
@@ -58,3 +63,269 @@ def test_add_and_refresh(db_session):
|
||||
db_user = db_session.get(UserDB, result.id)
|
||||
assert db_user is not None
|
||||
assert db_user.name == "Test User"
|
||||
|
||||
|
||||
CARD_MIGRATION_REVISION = "b7c1d2e3f4a5"
|
||||
CARD_COLUMNS = ["id", "key", "card_serial", "enabled", "name", "group_id", "last_used"]
|
||||
PREVIOUS_REVISION = "93e8dda267e5"
|
||||
|
||||
|
||||
def _upgrade_head(db_url: str) -> None:
|
||||
alembic_cfg = Config(settings.alembic_config)
|
||||
alembic_cfg.set_main_option("sqlalchemy.url", db_url)
|
||||
command.upgrade(alembic_cfg, "head")
|
||||
|
||||
|
||||
def _downgrade(db_url: str, to: str) -> None:
|
||||
alembic_cfg = Config(settings.alembic_config)
|
||||
alembic_cfg.set_main_option("sqlalchemy.url", db_url)
|
||||
command.downgrade(alembic_cfg, to)
|
||||
|
||||
|
||||
def _stamp(db_url: str, revision: str) -> None:
|
||||
"""Mark a database as being exactly at *revision*.
|
||||
|
||||
command.stamp() walks the whole chain and would run the intermediate
|
||||
revisions' upgrade(), which fails on a database that never had the timetable
|
||||
table. The version row is all alembic needs to know where to continue.
|
||||
"""
|
||||
engine = create_engine(db_url)
|
||||
with engine.connect() as connection:
|
||||
connection.exec_driver_sql(
|
||||
"CREATE TABLE IF NOT EXISTS alembic_version "
|
||||
"(version_num VARCHAR(32) NOT NULL)"
|
||||
)
|
||||
connection.exec_driver_sql("DELETE FROM alembic_version")
|
||||
connection.exec_driver_sql(
|
||||
"INSERT INTO alembic_version (version_num) VALUES (?)", (revision,)
|
||||
)
|
||||
connection.commit()
|
||||
engine.dispose()
|
||||
|
||||
|
||||
LEGACY_CARD_DDL = """CREATE TABLE card (
|
||||
\tid INTEGER NOT NULL,
|
||||
\t"key" VARCHAR NOT NULL,
|
||||
\tcard_serial VARCHAR NOT NULL,
|
||||
\tenabled BOOLEAN NOT NULL,
|
||||
\tname VARCHAR(32) NOT NULL,
|
||||
\tgroup_id INTEGER,
|
||||
\tPRIMARY KEY (id),
|
||||
\tUNIQUE (name),
|
||||
\tFOREIGN KEY(group_id) REFERENCES groupdb (id)
|
||||
)"""
|
||||
|
||||
|
||||
def _legacy_card_db(tmp_path):
|
||||
"""A database as it looked before the last_used / non-nullable group_id change.
|
||||
|
||||
The card table is written with the DDL SQLite produced from the old model
|
||||
(nullable group_id, unnamed FK, no last_used) and the database is stamped at
|
||||
the previous revision, so `alembic upgrade head` has to repair it.
|
||||
"""
|
||||
db_file = tmp_path / "legacy.db"
|
||||
engine = create_engine(f"sqlite:///{db_file}")
|
||||
|
||||
# Base tables from the current model, minus the card table built afterwards.
|
||||
metadata = SQLModel.metadata
|
||||
with engine.begin() as connection:
|
||||
for table in metadata.sorted_tables:
|
||||
if table.name != "card":
|
||||
table.create(connection)
|
||||
connection.execute(text(LEGACY_CARD_DDL))
|
||||
connection.execute(text("INSERT INTO groupdb (name) VALUES ('legacy group')"))
|
||||
connection.execute(
|
||||
text(
|
||||
"INSERT INTO card (key, card_serial, enabled, name, group_id)"
|
||||
" VALUES ('legacy-key', '00:00:00:00:00:00:00', 1,"
|
||||
" 'legacy_attached_card', 1)"
|
||||
)
|
||||
)
|
||||
# Allowed only because the old column was nullable.
|
||||
connection.execute(
|
||||
text(
|
||||
"INSERT INTO card (key, card_serial, enabled, name, group_id)"
|
||||
" VALUES ('orphan-key', '00:00:00:00:00:00:00', 1,"
|
||||
" 'legacy_orphan_card', NULL)"
|
||||
)
|
||||
)
|
||||
|
||||
_stamp(f"sqlite:///{db_file}", PREVIOUS_REVISION)
|
||||
return db_file, engine
|
||||
|
||||
|
||||
def _dispose(*engines) -> None:
|
||||
"""sqlite engines keep their connection open, which pytest reports as
|
||||
ResourceWarning: unclosed database."""
|
||||
for engine in engines:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _card_columns(engine):
|
||||
return [
|
||||
(column["name"], str(column["type"]), column["nullable"])
|
||||
for column in inspect(engine).get_columns("card")
|
||||
]
|
||||
|
||||
|
||||
def _split_clauses(ddl: str) -> list[str]:
|
||||
"""Split a CREATE TABLE body on top level commas (VARCHAR(32) keeps its own)."""
|
||||
body = ddl[ddl.index("(") + 1 : ddl.rindex(")")]
|
||||
clauses, depth, current = [], 0, ""
|
||||
for char in body:
|
||||
if char == "(":
|
||||
depth += 1
|
||||
elif char == ")":
|
||||
depth -= 1
|
||||
if char == "," and depth == 0:
|
||||
clauses.append(current.strip())
|
||||
current = ""
|
||||
else:
|
||||
current += char
|
||||
clauses.append(current.strip())
|
||||
return [clause for clause in clauses if clause]
|
||||
|
||||
|
||||
def _card_ddl(engine) -> str:
|
||||
"""The card CREATE TABLE as stored in sqlite_master, normalized so it can be
|
||||
compared across databases.
|
||||
|
||||
op.batch_alter_table quotes the table name and reorders the table level
|
||||
constraints, which is cosmetic; the columns and their nullability are what
|
||||
the comparison is about.
|
||||
"""
|
||||
with engine.connect() as connection:
|
||||
row = connection.execute(
|
||||
text("SELECT sql FROM sqlite_master WHERE name = 'card'")
|
||||
).one()
|
||||
clauses = _split_clauses(row[0].replace('"', ""))
|
||||
columns = [clause for clause in clauses if clause.split(" ")[0] in CARD_COLUMNS]
|
||||
constraints = sorted(clause for clause in clauses if clause not in columns)
|
||||
return re.sub(
|
||||
r"\s+", " ", "CREATE TABLE card ( " + ", ".join(columns + constraints) + " )"
|
||||
).strip()
|
||||
|
||||
|
||||
def test_card_not_null_migration_matches_the_model_schema(tmp_path):
|
||||
"""alembic upgrade head on a pre-change database ends up with exactly the
|
||||
schema SQLModel.metadata.create_all() builds from the current model.
|
||||
|
||||
create_db_and_tables() only bootstraps an *empty* database, so without this
|
||||
migration every existing deployment keeps a nullable group_id and no
|
||||
last_used column, and reading a Card fails with
|
||||
'no such column: card.last_used'.
|
||||
"""
|
||||
db_file, legacy_engine = _legacy_card_db(tmp_path)
|
||||
assert "last_used" not in [name for name, _, _ in _card_columns(legacy_engine)]
|
||||
assert [c for c in _card_columns(legacy_engine) if c[0] == "group_id"][0][2] is True
|
||||
|
||||
_upgrade_head(f"sqlite:///{db_file}")
|
||||
|
||||
fresh_engine = create_engine(f"sqlite:///{tmp_path / 'fresh.db'}")
|
||||
SQLModel.metadata.create_all(fresh_engine)
|
||||
assert _card_columns(legacy_engine) == _card_columns(fresh_engine)
|
||||
assert _card_ddl(legacy_engine) == _card_ddl(fresh_engine)
|
||||
_dispose(legacy_engine, fresh_engine)
|
||||
assert _card_columns(legacy_engine) == [
|
||||
("id", "INTEGER", False),
|
||||
("key", "VARCHAR", False),
|
||||
("card_serial", "VARCHAR", False),
|
||||
("enabled", "BOOLEAN", False),
|
||||
("name", "VARCHAR(32)", False),
|
||||
("group_id", "INTEGER", False),
|
||||
("last_used", "DATETIME", False),
|
||||
]
|
||||
|
||||
|
||||
def test_card_not_null_migration_backfills_and_keeps_assigned_cards(tmp_path):
|
||||
"""Existing rows get a last_used value and the attached card keeps its group;
|
||||
only the orphaned row the nullable FK allowed is removed."""
|
||||
db_file, engine = _legacy_card_db(tmp_path)
|
||||
_upgrade_head(f"sqlite:///{db_file}")
|
||||
|
||||
with Session(engine) as session:
|
||||
attached = session.exec(
|
||||
select(Card).where(Card.name == "legacy_attached_card")
|
||||
).one()
|
||||
assert attached.group_id is not None
|
||||
assert attached.last_used is not None
|
||||
assert attached.last_used.tzinfo is not None
|
||||
assert (
|
||||
session.exec(
|
||||
select(Card).where(Card.name == "legacy_orphan_card")
|
||||
).one_or_none()
|
||||
is None
|
||||
)
|
||||
|
||||
|
||||
def test_card_not_null_migration_round_trip(tmp_path):
|
||||
"""The downgrade restores the old nullable schema and the upgrade reapplies
|
||||
the change without a batch reflection error."""
|
||||
db_file, engine = _legacy_card_db(tmp_path)
|
||||
_upgrade_head(f"sqlite:///{db_file}")
|
||||
migrated_ddl = _card_ddl(engine)
|
||||
|
||||
_downgrade(f"sqlite:///{db_file}", PREVIOUS_REVISION)
|
||||
assert [c for c in _card_columns(engine) if c[0] == "group_id"][0][2] is True
|
||||
assert "last_used" not in [name for name, _, _ in _card_columns(engine)]
|
||||
|
||||
_upgrade_head(f"sqlite:///{db_file}")
|
||||
assert _card_ddl(engine) == migrated_ddl
|
||||
_dispose(engine)
|
||||
|
||||
|
||||
def test_card_not_null_migration_leaves_a_bootstrapped_database_alone(tmp_path):
|
||||
"""A database create_db_and_tables() bootstrapped from the current model is
|
||||
stamped at head, so `alembic upgrade head` is a no-op on it."""
|
||||
db_file = tmp_path / "current.db"
|
||||
url = f"sqlite:///{db_file}"
|
||||
engine = create_engine(url)
|
||||
with engine.begin() as connection:
|
||||
for table in SQLModel.metadata.sorted_tables:
|
||||
Table(table.name, SQLModel.metadata).create(connection)
|
||||
_stamp(url, CARD_MIGRATION_REVISION)
|
||||
before = _card_ddl(engine)
|
||||
|
||||
_upgrade_head(url)
|
||||
|
||||
assert _card_ddl(engine) == before
|
||||
_dispose(engine)
|
||||
|
||||
|
||||
def test_card_not_null_migration_is_not_repeatable_on_an_already_tightened_table(
|
||||
tmp_path,
|
||||
):
|
||||
"""The revision is not idempotent SQL: it assumes a pre-change card table.
|
||||
|
||||
A deployment that already has the column (manually patched, or rebuilt from
|
||||
the model) but is still stamped at the previous revision fails here. That is
|
||||
why create_db_and_tables() stamps a freshly created database at head instead
|
||||
of running the scripts.
|
||||
"""
|
||||
db_file = tmp_path / "already_tightened.db"
|
||||
url = f"sqlite:///{db_file}"
|
||||
engine = create_engine(url)
|
||||
with engine.begin() as connection:
|
||||
for table in SQLModel.metadata.sorted_tables:
|
||||
Table(table.name, SQLModel.metadata).create(connection)
|
||||
_stamp(url, PREVIOUS_REVISION)
|
||||
|
||||
with pytest.raises(OperationalError) as excinfo:
|
||||
_upgrade_head(url)
|
||||
assert "duplicate column name: last_used" in str(excinfo.value)
|
||||
_dispose(engine)
|
||||
|
||||
|
||||
def test_card_not_null_migration_head_is_the_scripts_head(tmp_path):
|
||||
"""The new revision is the head the rest of the suite stamps databases with."""
|
||||
db_file, engine = _legacy_card_db(tmp_path)
|
||||
_upgrade_head(f"sqlite:///{db_file}")
|
||||
with engine.connect() as connection:
|
||||
stamped = connection.execute(
|
||||
text("SELECT version_num FROM alembic_version")
|
||||
).all()
|
||||
head = ScriptDirectory.from_config(
|
||||
Config(settings.alembic_config)
|
||||
).get_current_head()
|
||||
assert [row[0] for row in stamped] == [head] == [CARD_MIGRATION_REVISION]
|
||||
_dispose(engine)
|
||||
|
||||
@@ -23,6 +23,7 @@ def test_check_access_with_valid_oneshot(db_session):
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
last_used=datetime.datetime.now(),
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
@@ -61,6 +62,7 @@ def test_check_access_with_inactive_aa(db_session):
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
last_used=datetime.datetime.now(),
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
@@ -90,6 +92,7 @@ def test_check_access_with_inactive_card(db_session):
|
||||
enabled=False,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
last_used=datetime.datetime.now(),
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
@@ -119,6 +122,7 @@ def test_regression_check_access_issue_28(db_session):
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
last_used=datetime.datetime.now(),
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
@@ -163,6 +167,7 @@ def _create_card_with_oneshot(db_session, ends_at):
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
last_used=datetime.datetime.now(),
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
|
||||
@@ -1,3 +1,9 @@
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import pytest
|
||||
from sqlmodel import select
|
||||
|
||||
|
||||
def test_create_group(client, auth_headers):
|
||||
"""Test creating a new group."""
|
||||
group_data = {"name": "New Test Group"}
|
||||
@@ -43,6 +49,163 @@ def test_delete_group(client, auth_headers, test_group):
|
||||
assert not any(group["id"] == test_group.id for group in groups)
|
||||
|
||||
|
||||
def _group_with_card(client, auth_headers, db_session, name):
|
||||
"""Create a group through the API and attach a card to it directly in the
|
||||
db (POST /api/v1/cards/ needs a physical card to write to)."""
|
||||
from app.model.models import Card
|
||||
|
||||
group = client.post("/api/v1/groups/", json={"name": name}, headers=auth_headers)
|
||||
assert group.status_code == 200
|
||||
group_id = group.json()["id"]
|
||||
db_session.add(
|
||||
Card(
|
||||
key=f"key-{name}",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
name=f"card-{name}",
|
||||
enabled=True,
|
||||
group_id=group_id,
|
||||
last_used=datetime.now(UTC),
|
||||
)
|
||||
)
|
||||
db_session.commit()
|
||||
return group_id
|
||||
|
||||
|
||||
def test_delete_group_with_assigned_card_is_refused(client, auth_headers, db_session):
|
||||
"""A group that still has a card assigned must not be deleted."""
|
||||
from app.model.models import Card, GroupDB
|
||||
|
||||
group_id = _group_with_card(client, auth_headers, db_session, "group_with_card")
|
||||
|
||||
response = client.delete(f"/api/v1/groups/{group_id}", headers=auth_headers)
|
||||
assert response.status_code == 409
|
||||
assert "card" in response.json()["detail"].lower()
|
||||
|
||||
db_session.expire_all()
|
||||
assert db_session.get(GroupDB, group_id) is not None
|
||||
assert (
|
||||
len(db_session.exec(select(Card).where(Card.group_id == group_id)).all()) == 1
|
||||
)
|
||||
|
||||
|
||||
def test_delete_group_refuses_even_when_the_card_is_disabled(
|
||||
client, auth_headers, db_session
|
||||
):
|
||||
"""A disabled card is still assigned, so it still blocks the deletion."""
|
||||
from app.model.models import Card
|
||||
|
||||
group_id = _group_with_card(client, auth_headers, db_session, "group_disabled_card")
|
||||
card = db_session.exec(select(Card).where(Card.group_id == group_id)).one()
|
||||
card.enabled = False
|
||||
db_session.add(card)
|
||||
db_session.commit()
|
||||
|
||||
response = client.delete(f"/api/v1/groups/{group_id}", headers=auth_headers)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
def test_group_delete_allowed_after_its_last_card_is_gone(
|
||||
client, auth_headers, db_session
|
||||
):
|
||||
"""Deleting the group's cards first unblocks the group deletion."""
|
||||
from app.model.models import Card
|
||||
|
||||
group_id = _group_with_card(client, auth_headers, db_session, "group_to_empty")
|
||||
for card in db_session.exec(select(Card).where(Card.group_id == group_id)).all():
|
||||
db_session.delete(card)
|
||||
db_session.commit()
|
||||
|
||||
response = client.delete(f"/api/v1/groups/{group_id}", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
response = client.get("/api/v1/groups/", headers=auth_headers)
|
||||
assert not any(group["id"] == group_id for group in response.json())
|
||||
|
||||
|
||||
def test_card_create_without_group_id_is_rejected(client, auth_headers):
|
||||
"""POST /cards/ requires a group_id, so a new card cannot start out orphaned."""
|
||||
for body in (
|
||||
{"name": "no_group_card", "key": "k", "card_serial": "00:00:00:00:00:00:00"},
|
||||
{
|
||||
"name": "null_group_card",
|
||||
"key": "k",
|
||||
"card_serial": "00:00:00:00:00:00:00",
|
||||
"group_id": None,
|
||||
},
|
||||
):
|
||||
response = client.post("/api/v1/cards/", json=body, headers=auth_headers)
|
||||
assert response.status_code == 422
|
||||
assert "group_id" in str(response.json()["detail"])
|
||||
|
||||
|
||||
def test_patch_cannot_orphan_a_card_by_nulling_group_id(
|
||||
client, auth_headers, db_session
|
||||
):
|
||||
"""update_card() dumps with exclude_unset+exclude_none, so an explicit null is
|
||||
dropped and never reaches sqlmodel_update. A card can be moved between groups
|
||||
but never unassigned through the API."""
|
||||
from app.model.models import Card
|
||||
|
||||
group_id = _group_with_card(client, auth_headers, db_session, "patch_null_group")
|
||||
card = db_session.exec(select(Card).where(Card.group_id == group_id)).one()
|
||||
|
||||
response = client.patch(
|
||||
f"/api/v1/cards/{card.id}", json={"group_id": None}, headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["group_id"] == group_id
|
||||
|
||||
db_session.expire_all()
|
||||
assert db_session.get(Card, card.id).group_id == group_id
|
||||
|
||||
|
||||
def test_patch_cannot_orphan_a_card_by_omitting_group_id(
|
||||
client, auth_headers, db_session
|
||||
):
|
||||
"""A PATCH that only touches other fields leaves group_id at its old value."""
|
||||
from app.model.models import Card
|
||||
|
||||
group_id = _group_with_card(client, auth_headers, db_session, "patch_no_group")
|
||||
card = db_session.exec(select(Card).where(Card.group_id == group_id)).one()
|
||||
|
||||
response = client.patch(
|
||||
f"/api/v1/cards/{card.id}", json={"enabled": False}, headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["group_id"] == group_id
|
||||
|
||||
db_session.expire_all()
|
||||
assert db_session.get(Card, card.id).group_id == group_id
|
||||
|
||||
|
||||
def test_card_without_group_id_cannot_be_written_to_the_db(db_session):
|
||||
"""Card.group_id is not nullable anymore. Neither Pydantic nor SQLModel validate
|
||||
annotations (Card(group_id=None) constructs fine), so the NOT NULL column built
|
||||
from the model is what actually keeps cards from being orphaned."""
|
||||
from sqlalchemy import exc as sa_exc
|
||||
|
||||
from app.model.models import Card
|
||||
|
||||
card = Card(
|
||||
key="orphan-key",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
name="orphan_card",
|
||||
enabled=True,
|
||||
group_id=None,
|
||||
last_used=datetime.now(UTC),
|
||||
)
|
||||
db_session.add(card)
|
||||
with pytest.raises(sa_exc.IntegrityError) as excinfo:
|
||||
db_session.commit()
|
||||
assert "NOT NULL constraint failed: card.group_id" in str(excinfo.value)
|
||||
|
||||
db_session.rollback()
|
||||
assert (
|
||||
db_session.exec(select(Card).where(Card.name == "orphan_card")).one_or_none()
|
||||
is None
|
||||
)
|
||||
|
||||
|
||||
def test_delete_nonexistent_group(client, auth_headers):
|
||||
"""Test deleting a non-existent group."""
|
||||
response = client.delete("/api/v1/groups/99999", headers=auth_headers)
|
||||
|
||||
Reference in New Issue
Block a user