Wunschliste 1 #36

Merged
ahtlon merged 6 commits from wunschliste_1 into master 2026-11-10 16:36:26 +01:00
15 changed files with 1403 additions and 38 deletions
@@ -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"))
+43
View File
@@ -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()
+16
View File
@@ -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"}
-25
View File
@@ -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()
+18 -2
View File
@@ -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"}
+2
View File
@@ -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
View File
@@ -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):
+3 -3
View File
@@ -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
View File
@@ -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:
+1
View File
@@ -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()
+571
View File
@@ -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)
+203
View File
@@ -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)
+274 -3
View File
@@ -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)
+5
View File
@@ -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)
+163
View File
@@ -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)