Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
70aaab5596 | ||
|
|
7182a520e2 | ||
|
|
6a2b6e090a | ||
|
|
9c12c663c1 | ||
|
|
ee91b03198 | ||
|
|
2a44e3a1a6 | ||
|
|
f0bd1e68e8 | ||
|
|
8ca4da2e1a
|
||
|
|
d93491f466 |
@@ -0,0 +1,14 @@
|
|||||||
|
name: "Check flake syntax"
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
jobs:
|
||||||
|
flake-check:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v4
|
||||||
|
- name: Install dependencies for Nix setup action
|
||||||
|
run: |
|
||||||
|
apt update -y
|
||||||
|
apt install sudo -y
|
||||||
|
- uses: cachix/install-nix-action@v30
|
||||||
|
- run: nix flake check --no-update-lock-file --accept-flake-config .
|
||||||
@@ -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
|
import logging
|
||||||
|
from datetime import UTC, datetime
|
||||||
|
|
||||||
from fastapi import APIRouter, Depends, HTTPException, status
|
from fastapi import APIRouter, Depends, HTTPException, status
|
||||||
from sqlalchemy.exc import NoResultFound
|
from sqlalchemy.exc import NoResultFound
|
||||||
@@ -27,6 +28,7 @@ def register_card(cardInput: CardCreate):
|
|||||||
name=cardInput.name,
|
name=cardInput.name,
|
||||||
card_serial=uid,
|
card_serial=uid,
|
||||||
enabled=cardInput.enabled,
|
enabled=cardInput.enabled,
|
||||||
|
last_used=datetime.now(tz=UTC),
|
||||||
)
|
)
|
||||||
return card
|
return card
|
||||||
|
|
||||||
@@ -116,3 +118,17 @@ def update_card(
|
|||||||
) from err
|
) from err
|
||||||
db_card.sqlmodel_update(card_data)
|
db_card.sqlmodel_update(card_data)
|
||||||
return add_and_refresh(db, db_card)
|
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.model.models import Card
|
||||||
from app.services.auth import auth_is_admin
|
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
|
from app.services.database import add_and_refresh, get_session
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -64,27 +63,3 @@ def list_all_cards(db: Session = Depends(get_session)):
|
|||||||
for i in cards:
|
for i in cards:
|
||||||
out.append({i.key: i.group.name})
|
out.append({i.key: i.group.name})
|
||||||
return out
|
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 fastapi import APIRouter, Depends, HTTPException, status
|
||||||
|
from sqlalchemy import func
|
||||||
from sqlmodel import Session, select
|
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.auth import auth_is_admin
|
||||||
from ..services.database import add_and_refresh, get_session
|
from ..services.database import add_and_refresh, get_session
|
||||||
|
|
||||||
@@ -32,7 +33,10 @@ def create_group(
|
|||||||
return add_and_refresh(db, db_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(
|
def delete_group(
|
||||||
*,
|
*,
|
||||||
db: Session = Depends(get_session),
|
db: Session = Depends(get_session),
|
||||||
@@ -42,6 +46,18 @@ def delete_group(
|
|||||||
db_group = db.get(GroupDB, group_id)
|
db_group = db.get(GroupDB, group_id)
|
||||||
if db_group is None:
|
if db_group is None:
|
||||||
raise HTTPException(status_code=404, detail="Group not found")
|
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.delete(db_group)
|
||||||
db.commit()
|
db.commit()
|
||||||
return {"message": "Group deleted successfully"}
|
return {"message": "Group deleted successfully"}
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ from fastapi.security import OAuth2PasswordBearer
|
|||||||
|
|
||||||
from app.controllers import (
|
from app.controllers import (
|
||||||
aaManager,
|
aaManager,
|
||||||
|
calendarManager,
|
||||||
cardManager,
|
cardManager,
|
||||||
debugManager,
|
debugManager,
|
||||||
doorManager,
|
doorManager,
|
||||||
@@ -75,4 +76,5 @@ app.include_router(groupManager.group_router)
|
|||||||
app.include_router(cardManager.card_router)
|
app.include_router(cardManager.card_router)
|
||||||
app.include_router(aaManager.aa_router)
|
app.include_router(aaManager.aa_router)
|
||||||
app.include_router(doorManager.door_router)
|
app.include_router(doorManager.door_router)
|
||||||
|
app.include_router(calendarManager.calendar_router)
|
||||||
app.include_router(debugManager.debug_router)
|
app.include_router(debugManager.debug_router)
|
||||||
|
|||||||
+2
-1
@@ -168,8 +168,9 @@ class Card(Base, table=True):
|
|||||||
card_serial: str
|
card_serial: str
|
||||||
enabled: bool = True
|
enabled: bool = True
|
||||||
name: str = Field(unique=True, max_length=32)
|
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")
|
group: GroupDB | None = Relationship(back_populates="cards")
|
||||||
|
last_used: datetime = Field(sa_type=UTCDateTime)
|
||||||
|
|
||||||
|
|
||||||
class CardCreate(Base):
|
class CardCreate(Base):
|
||||||
|
|||||||
@@ -128,11 +128,11 @@ class CaldavEventCache:
|
|||||||
|
|
||||||
if event["all_day"]: # noqa: SIM108
|
if event["all_day"]: # noqa: SIM108
|
||||||
# All-day event end dates are exclusive.
|
# All-day event end dates are exclusive.
|
||||||
is_active = start <= today
|
is_pending = start > today
|
||||||
else:
|
else:
|
||||||
is_active = start <= active_time
|
is_pending = start > active_time
|
||||||
|
|
||||||
if is_active:
|
if is_pending:
|
||||||
future.append(event)
|
future.append(event)
|
||||||
|
|
||||||
return future
|
return future
|
||||||
|
|||||||
+16
-4
@@ -1,5 +1,5 @@
|
|||||||
import logging
|
import logging
|
||||||
from datetime import datetime
|
from datetime import UTC, datetime
|
||||||
|
|
||||||
from sqlalchemy import exc
|
from sqlalchemy import exc
|
||||||
from sqlmodel import select
|
from sqlmodel import select
|
||||||
@@ -99,9 +99,17 @@ def decrementOneshot(db: Session, oneshot: OneShotAccess):
|
|||||||
add_and_refresh(db, oneshot)
|
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):
|
def checkAccess(key: str, db: Session):
|
||||||
try:
|
try:
|
||||||
card = db.exec(select(Card).where(Card.key == key)).one()
|
card = db.exec(select(Card).where(Card.key == key)).one()
|
||||||
|
updateLastUsed(db, card)
|
||||||
if not card.enabled:
|
if not card.enabled:
|
||||||
logger.info("Card Inactive!")
|
logger.info("Card Inactive!")
|
||||||
return False
|
return False
|
||||||
@@ -122,9 +130,13 @@ def checkAccess(key: str, db: Session):
|
|||||||
ends_at = auth.oneshot.ends_at
|
ends_at = auth.oneshot.ends_at
|
||||||
if ends_at.tzinfo is None:
|
if ends_at.tzinfo is None:
|
||||||
ends_at = ends_at.astimezone()
|
ends_at = ends_at.astimezone()
|
||||||
if current_time < ends_at and auth.oneshot.uses > 0:
|
if current_time < ends_at:
|
||||||
decrementOneshot(db, auth.oneshot)
|
if auth.oneshot.uses > 0:
|
||||||
return True
|
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")
|
logger.info("No more auths found")
|
||||||
return False
|
return False
|
||||||
except exc.NoResultFound as err:
|
except exc.NoResultFound as err:
|
||||||
|
|||||||
@@ -217,6 +217,7 @@ def test_card(db_session, test_group):
|
|||||||
enabled=True,
|
enabled=True,
|
||||||
name="test_card",
|
name="test_card",
|
||||||
card_serial="00:00:00:00:00:00:00",
|
card_serial="00:00:00:00:00:00:00",
|
||||||
|
last_used=datetime.now(),
|
||||||
)
|
)
|
||||||
db_session.add(card)
|
db_session.add(card)
|
||||||
db_session.commit()
|
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.config import Config
|
||||||
from alembic.script import ScriptDirectory
|
from alembic.script import ScriptDirectory
|
||||||
from sqlalchemy import create_engine, text
|
from sqlalchemy import Table, inspect, text
|
||||||
from sqlmodel import Session
|
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 (
|
from app.services.database import (
|
||||||
add_and_refresh,
|
add_and_refresh,
|
||||||
configure_engine,
|
configure_engine,
|
||||||
@@ -58,3 +63,269 @@ def test_add_and_refresh(db_session):
|
|||||||
db_user = db_session.get(UserDB, result.id)
|
db_user = db_session.get(UserDB, result.id)
|
||||||
assert db_user is not None
|
assert db_user is not None
|
||||||
assert db_user.name == "Test User"
|
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,
|
enabled=True,
|
||||||
name="test_card",
|
name="test_card",
|
||||||
card_serial="00:00:00:00:00:00:00",
|
card_serial="00:00:00:00:00:00:00",
|
||||||
|
last_used=datetime.datetime.now(),
|
||||||
)
|
)
|
||||||
db_session.add(card)
|
db_session.add(card)
|
||||||
|
|
||||||
@@ -61,6 +62,7 @@ def test_check_access_with_inactive_aa(db_session):
|
|||||||
enabled=True,
|
enabled=True,
|
||||||
name="test_card",
|
name="test_card",
|
||||||
card_serial="00:00:00:00:00:00:00",
|
card_serial="00:00:00:00:00:00:00",
|
||||||
|
last_used=datetime.datetime.now(),
|
||||||
)
|
)
|
||||||
db_session.add(card)
|
db_session.add(card)
|
||||||
|
|
||||||
@@ -90,6 +92,7 @@ def test_check_access_with_inactive_card(db_session):
|
|||||||
enabled=False,
|
enabled=False,
|
||||||
name="test_card",
|
name="test_card",
|
||||||
card_serial="00:00:00:00:00:00:00",
|
card_serial="00:00:00:00:00:00:00",
|
||||||
|
last_used=datetime.datetime.now(),
|
||||||
)
|
)
|
||||||
db_session.add(card)
|
db_session.add(card)
|
||||||
|
|
||||||
@@ -119,6 +122,7 @@ def test_regression_check_access_issue_28(db_session):
|
|||||||
enabled=True,
|
enabled=True,
|
||||||
name="test_card",
|
name="test_card",
|
||||||
card_serial="00:00:00:00:00:00:00",
|
card_serial="00:00:00:00:00:00:00",
|
||||||
|
last_used=datetime.datetime.now(),
|
||||||
)
|
)
|
||||||
db_session.add(card)
|
db_session.add(card)
|
||||||
|
|
||||||
@@ -163,6 +167,7 @@ def _create_card_with_oneshot(db_session, ends_at):
|
|||||||
enabled=True,
|
enabled=True,
|
||||||
name="test_card",
|
name="test_card",
|
||||||
card_serial="00:00:00:00:00:00:00",
|
card_serial="00:00:00:00:00:00:00",
|
||||||
|
last_used=datetime.datetime.now(),
|
||||||
)
|
)
|
||||||
db_session.add(card)
|
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):
|
def test_create_group(client, auth_headers):
|
||||||
"""Test creating a new group."""
|
"""Test creating a new group."""
|
||||||
group_data = {"name": "New Test 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)
|
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):
|
def test_delete_nonexistent_group(client, auth_headers):
|
||||||
"""Test deleting a non-existent group."""
|
"""Test deleting a non-existent group."""
|
||||||
response = client.delete("/api/v1/groups/99999", headers=auth_headers)
|
response = client.delete("/api/v1/groups/99999", headers=auth_headers)
|
||||||
|
|||||||
Reference in New Issue
Block a user