Remove timetable in favour of the new calendar stuff
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
"""remove timetable access authorizations
|
||||
|
||||
Revision ID: 93e8dda267e5
|
||||
Revises: 4a40fa23e086
|
||||
Create Date: 2026-11-05 15:12:41.908153
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers used by Alembic.
|
||||
revision: str = '93e8dda267e5'
|
||||
down_revision: Union[str, Sequence[str], None] = '4a40fa23e086'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Drop the timetable table and the authorizations that used it.
|
||||
|
||||
create_db_and_tables() only bootstraps an empty database, so existing
|
||||
deployments keep the timetable table and their type='timetable' rows until
|
||||
this migration is applied with `alembic upgrade head`. Such rows can no
|
||||
longer be read: AccessAuthorizationDB has no timetables relationship
|
||||
anymore, door.checkAccess() ignores the type, and AccessAuthorizationResponse
|
||||
rejects the value, which turns every GET on them into a ResponseValidationError.
|
||||
"""
|
||||
# Delete the authorizations before the table so the FK ordering stays clean.
|
||||
op.execute(
|
||||
sa.text(
|
||||
"DELETE FROM accessauthorizationdb WHERE type = 'timetable'"
|
||||
)
|
||||
)
|
||||
op.drop_table('timetable')
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Recreate the timetable table.
|
||||
|
||||
The deleted type='timetable' authorizations and their timetable rows are
|
||||
gone for good; a downgrade only restores the schema.
|
||||
"""
|
||||
op.create_table(
|
||||
'timetable',
|
||||
sa.Column('weekday', sa.INTEGER(), nullable=False),
|
||||
sa.Column('starttime', sa.TIME(), nullable=False),
|
||||
sa.Column('duration', sa.INTEGER(), nullable=False),
|
||||
sa.Column('id', sa.INTEGER(), nullable=False),
|
||||
sa.Column('accessauth_id', sa.INTEGER(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
sa.ForeignKeyConstraint(
|
||||
['accessauth_id'], ['accessauthorizationdb.id'],
|
||||
),
|
||||
)
|
||||
@@ -1,7 +1,6 @@
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import selectinload
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from app.model.models import (
|
||||
@@ -12,7 +11,6 @@ from app.model.models import (
|
||||
GroupDB,
|
||||
GroupResponse,
|
||||
OneShotAccess,
|
||||
Timetable,
|
||||
)
|
||||
from app.services.auth import auth_is_admin
|
||||
from app.services.database import add_and_refresh, get_session
|
||||
@@ -30,10 +28,6 @@ def add_accessauth(
|
||||
admin: bool = Depends(auth_is_admin),
|
||||
):
|
||||
logger.info(f"Creating accessauth with data: {aa}")
|
||||
if aa.timetables != []:
|
||||
timetables = [Timetable.model_validate(t) for t in aa.timetables]
|
||||
else:
|
||||
timetables = []
|
||||
if aa.oneshot is not None:
|
||||
oneshot = OneShotAccess.model_validate(aa.oneshot)
|
||||
else:
|
||||
@@ -42,7 +36,6 @@ def add_accessauth(
|
||||
name=aa.name,
|
||||
type=aa.type,
|
||||
is_active=aa.is_active,
|
||||
timetables=timetables,
|
||||
oneshot=oneshot,
|
||||
)
|
||||
return add_and_refresh(db, db_aa)
|
||||
@@ -52,11 +45,7 @@ def add_accessauth(
|
||||
def get_all_accessauths(
|
||||
db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)
|
||||
):
|
||||
return db.exec(
|
||||
select(AccessAuthorizationDB).options(
|
||||
selectinload(AccessAuthorizationDB.timetables)
|
||||
)
|
||||
).all()
|
||||
return db.exec(select(AccessAuthorizationDB)).all()
|
||||
|
||||
|
||||
@aa_router.get("/{aa_id}", response_model=AccessAuthorizationResponse)
|
||||
@@ -126,11 +115,6 @@ def change_accessauth(
|
||||
if db_aa is None:
|
||||
raise HTTPException(status_code=404, detail="AccessAuthorization not found")
|
||||
aa_data = aa.model_dump(exclude_unset=True, exclude_none=True)
|
||||
if "timetables" in aa_data and aa_data["timetables"] is not None:
|
||||
db_aa.timetables.clear()
|
||||
timetables = [Timetable.model_validate(t) for t in aa_data["timetables"]]
|
||||
db_aa.timetables = timetables
|
||||
aa_data.pop("timetables")
|
||||
if "oneshot" in aa_data and aa_data["oneshot"] is not None:
|
||||
oneshot = OneShotAccess.model_validate(aa_data["oneshot"])
|
||||
db_aa.oneshot = oneshot
|
||||
|
||||
+4
-35
@@ -1,4 +1,4 @@
|
||||
from datetime import UTC, datetime, time
|
||||
from datetime import UTC, datetime
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import model_validator
|
||||
@@ -109,7 +109,7 @@ class GroupResponse(GroupBase):
|
||||
#### AccessAuthorization
|
||||
class AccessAuthorizationBase(Base):
|
||||
name: str = Field(index=True)
|
||||
type: Literal["timetable", "oneshot", "somefuturespec"]
|
||||
type: Literal["oneshot", "somefuturespec"]
|
||||
is_active: bool
|
||||
|
||||
|
||||
@@ -119,9 +119,6 @@ class AccessAuthorizationDB(AccessAuthorizationBase, table=True):
|
||||
groups: list["GroupDB"] = Relationship(
|
||||
back_populates="accessauths", link_model=AaGroupLink
|
||||
)
|
||||
timetables: list["Timetable"] = Relationship(
|
||||
back_populates="accessauth", cascade_delete=True
|
||||
)
|
||||
oneshot: "OneShotAccess" = Relationship(
|
||||
back_populates="accessauth", cascade_delete=True
|
||||
)
|
||||
@@ -139,23 +136,13 @@ class OneShotAccess(OneShotAccessBase, table=True):
|
||||
|
||||
|
||||
class AccessAuthorizationCreate(AccessAuthorizationBase):
|
||||
timetables: list["TimetableCreate"] = []
|
||||
oneshot: OneShotAccessBase | None = None
|
||||
|
||||
@model_validator(mode="after")
|
||||
def check_type(self):
|
||||
if self.type == "timetable":
|
||||
if not self.timetables:
|
||||
raise ValueError(
|
||||
"timetable auths require at least one timetable object"
|
||||
)
|
||||
if self.oneshot is not None:
|
||||
raise ValueError("timetable auths are not allowed oneshot objects")
|
||||
elif self.type == "oneshot":
|
||||
if self.type == "oneshot":
|
||||
if not self.oneshot:
|
||||
raise ValueError("oneshot auths require a oneshot object")
|
||||
if self.timetables:
|
||||
raise ValueError("oneshot auths are not allowed timetable objects")
|
||||
elif self.type == "somefuturespec":
|
||||
raise ValueError("somefuturespec is not jet implemented. Please do not use")
|
||||
return self
|
||||
@@ -163,16 +150,14 @@ class AccessAuthorizationCreate(AccessAuthorizationBase):
|
||||
|
||||
class AccessAuthorizationResponse(AccessAuthorizationBase):
|
||||
id: int
|
||||
timetables: list["Timetable"] = []
|
||||
oneshot: OneShotAccessBase | None = None
|
||||
groups: list["GroupDB"]
|
||||
|
||||
|
||||
class AccessAuthorizationUpdate(Base):
|
||||
name: str | None = None
|
||||
type: Literal["timetable", "oneshot", "somefuturespec"] | None = None
|
||||
type: Literal["oneshot", "somefuturespec"] | None = None
|
||||
is_active: bool | None = None
|
||||
timetables: list["TimetableCreate"] | None = None
|
||||
oneshot: OneShotAccessBase | None = None
|
||||
|
||||
|
||||
@@ -197,19 +182,3 @@ class CardUpdate(Base):
|
||||
name: str | None = None
|
||||
enabled: bool | None = None
|
||||
group_id: int | None = None
|
||||
|
||||
|
||||
class TimetableBase(Base):
|
||||
weekday: int = Field(le=6, ge=0)
|
||||
starttime: time
|
||||
duration: int = Field(gt=0, lt=1440)
|
||||
|
||||
|
||||
class Timetable(TimetableBase, table=True):
|
||||
id: int | None = Field(default=None, primary_key=True)
|
||||
accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id")
|
||||
accessauth: AccessAuthorizationDB = Relationship(back_populates="timetables")
|
||||
|
||||
|
||||
class TimetableCreate(TimetableBase):
|
||||
pass
|
||||
|
||||
+1
-21
@@ -1,5 +1,5 @@
|
||||
import logging
|
||||
from datetime import date, datetime, timedelta
|
||||
from datetime import date, datetime
|
||||
|
||||
from sqlalchemy import exc
|
||||
from sqlmodel import select
|
||||
@@ -122,26 +122,6 @@ def checkAccess(key: str, db: Session):
|
||||
if not auth.is_active:
|
||||
logger.info("AA inactive!")
|
||||
continue
|
||||
if auth.type == "timetable":
|
||||
for timetable in auth.timetables:
|
||||
logger.info(f" checking timetable {timetable.id}")
|
||||
logger.info(
|
||||
f" weekday: CUR:{current_weekday} TT:{timetable.weekday}"
|
||||
)
|
||||
if current_weekday == timetable.weekday:
|
||||
starttime = datetime.combine(
|
||||
date.today(),
|
||||
timetable.starttime,
|
||||
tzinfo=current_time.tzinfo,
|
||||
)
|
||||
endtime = starttime + timedelta(minutes=timetable.duration)
|
||||
logger.info(
|
||||
f" comparing time: Start:{starttime} "
|
||||
f"Current:{current_time} End:{endtime}"
|
||||
)
|
||||
if starttime < current_time < endtime:
|
||||
logger.info("Access Valid!")
|
||||
return True
|
||||
if auth.type == "oneshot":
|
||||
logger.info(f" oneshot auth found: {auth.oneshot}")
|
||||
ends_at = auth.oneshot.ends_at
|
||||
|
||||
@@ -1,10 +1,3 @@
|
||||
def test_app_startup(client):
|
||||
"""Test that the application starts up correctly."""
|
||||
response = client.get("/")
|
||||
# Application should respond (even if it's a 404)
|
||||
assert response.status_code in [404, 200]
|
||||
|
||||
|
||||
def test_router_includes():
|
||||
"""Test that all routers are included in the app."""
|
||||
from app.main import app
|
||||
|
||||
+22
-89
@@ -1,102 +1,35 @@
|
||||
import datetime
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.model.models import (
|
||||
AaGroupLink,
|
||||
AccessAuthorizationBase,
|
||||
AccessAuthorizationCreate,
|
||||
Card,
|
||||
GroupBase,
|
||||
GroupCreate,
|
||||
TimetableCreate,
|
||||
Token,
|
||||
TokenData,
|
||||
UserBase,
|
||||
UserCreate,
|
||||
UserUpdate,
|
||||
)
|
||||
from app.model.models import AccessAuthorizationCreate, UserUpdate
|
||||
|
||||
|
||||
def test_user_models():
|
||||
"""Test user model creation and validation."""
|
||||
# Test UserBase
|
||||
user_base = UserBase(name="Test User", email="test@example.com", is_admin=False)
|
||||
assert user_base.name == "Test User"
|
||||
assert user_base.email == "test@example.com"
|
||||
assert user_base.is_admin is False
|
||||
|
||||
# Test UserCreate
|
||||
user_create = UserCreate(
|
||||
name="New User", email="new@example.com", password="secret123"
|
||||
)
|
||||
assert user_create.password == "secret123"
|
||||
|
||||
# Test UserUpdate
|
||||
def test_user_update_fields_are_optional():
|
||||
"""Partial update models must leave unspecified fields as None."""
|
||||
user_update = UserUpdate(name="Updated Name")
|
||||
assert user_update.name == "Updated Name"
|
||||
assert user_update.email is None
|
||||
|
||||
|
||||
def test_group_models():
|
||||
"""Test group model creation and validation."""
|
||||
# Test GroupBase
|
||||
group_base = GroupBase(name="Test Group")
|
||||
assert group_base.name == "Test Group"
|
||||
def test_access_authorization_type_is_validated():
|
||||
"""AA validation test"""
|
||||
base = {"name": "AA", "is_active": True}
|
||||
|
||||
# Test GroupCreate
|
||||
group_create = GroupCreate(name="New Group")
|
||||
assert group_create.name == "New Group"
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
AccessAuthorizationCreate(type="timetable", **base)
|
||||
assert exc_info.value.errors()[0]["type"] == "literal_error"
|
||||
|
||||
# A oneshot type without its oneshot payload is incomplete.
|
||||
with pytest.raises(ValueError, match="require a oneshot object"):
|
||||
AccessAuthorizationCreate(type="oneshot", **base)
|
||||
|
||||
def test_access_authorization_models():
|
||||
"""Test access authorization model creation and validation."""
|
||||
# Test AccessAuthorizationBase
|
||||
aa_base = AccessAuthorizationBase(name="Test AA", is_active=True, type="timetable")
|
||||
assert aa_base.name == "Test AA"
|
||||
assert aa_base.is_active is True
|
||||
with pytest.raises(ValueError, match="not jet implemented"):
|
||||
AccessAuthorizationCreate(type="somefuturespec", **base)
|
||||
|
||||
# Test AccessAuthorizationCreate with timetables
|
||||
timetable_create = TimetableCreate(weekday=1, starttime="08:00", duration=60)
|
||||
aa_create = AccessAuthorizationCreate(
|
||||
name="New AA", is_active=False, type="timetable", timetables=[timetable_create]
|
||||
valid = AccessAuthorizationCreate(
|
||||
type="oneshot",
|
||||
name="AA",
|
||||
is_active=True,
|
||||
oneshot={"uses": 1, "ends_at": "2029-07-27"},
|
||||
)
|
||||
assert aa_create.name == "New AA"
|
||||
assert aa_create.is_active is False
|
||||
assert len(aa_create.timetables) == 1
|
||||
|
||||
|
||||
def test_card_model():
|
||||
"""Test card model creation and validation."""
|
||||
card = Card(key="test-key", group_id=1)
|
||||
assert card.key == "test-key"
|
||||
assert card.group_id == 1
|
||||
|
||||
|
||||
def test_timetable_models():
|
||||
"""Test timetable model creation and validation."""
|
||||
# Test TimetableBase with valid values
|
||||
timetable = TimetableCreate(weekday=1, starttime="09:00", duration=120)
|
||||
assert timetable.weekday == 1
|
||||
assert timetable.starttime == datetime.time(9, 0)
|
||||
assert timetable.duration == 120
|
||||
|
||||
# Test boundary values
|
||||
max_duration = TimetableCreate(weekday=6, starttime="23:59", duration=1439)
|
||||
assert max_duration.duration == 1439
|
||||
assert max_duration.weekday == 6
|
||||
|
||||
|
||||
def test_token_models():
|
||||
"""Test token model creation and validation."""
|
||||
token = Token(access_token="test-token", token_type="bearer")
|
||||
assert token.access_token == "test-token"
|
||||
assert token.token_type == "bearer"
|
||||
|
||||
token_data = TokenData(username="testuser")
|
||||
assert token_data.username == "testuser"
|
||||
|
||||
|
||||
def test_aa_group_link_model():
|
||||
"""Test many-to-many relationship link model."""
|
||||
link = AaGroupLink(group_id=1, accessauth_id=2)
|
||||
assert link.group_id == 1
|
||||
assert link.accessauth_id == 2
|
||||
assert valid.type == "oneshot"
|
||||
|
||||
@@ -4,7 +4,7 @@ import pytest
|
||||
from sqlalchemy import text
|
||||
from sqlmodel import select
|
||||
|
||||
from app.model.models import AccessAuthorizationDB, OneShotAccess
|
||||
from app.model.models import OneShotAccess
|
||||
|
||||
# Berlin's DST rules as a POSIX TZ string: +01:00 in winter, +02:00 in summer.
|
||||
BERLIN_TZ = "CET-1CEST,M3.5.0,M10.5.0/3"
|
||||
@@ -24,29 +24,6 @@ def _post_oneshot_aa(client, auth_headers, name: str, ends_at: str):
|
||||
)
|
||||
|
||||
|
||||
def test_create_access_auth_tt(client, auth_headers):
|
||||
"""Test creating a new access authorization."""
|
||||
aa_data = {
|
||||
"name": "New tt_AA",
|
||||
"type": "timetable",
|
||||
"is_active": True,
|
||||
"timetables": [
|
||||
{"weekday": 1, "starttime": "08:00", "duration": 60},
|
||||
{"weekday": 2, "starttime": "09:00", "duration": 90},
|
||||
],
|
||||
}
|
||||
|
||||
response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert data["name"] == "New tt_AA"
|
||||
assert data["is_active"] is True
|
||||
assert data["type"] == "timetable"
|
||||
assert "id" in data
|
||||
assert len(data["timetables"]) == 2
|
||||
|
||||
|
||||
def test_create_access_auth_os(client, auth_headers):
|
||||
"""Test creating a new access authorization with oneshot type."""
|
||||
aa_data = {
|
||||
@@ -79,26 +56,49 @@ def test_create_wrong_aa_type(client, auth_headers):
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_get_all_access_auths(client, auth_headers, test_aa_tt):
|
||||
def test_create_timetable_aa_is_rejected(client, auth_headers):
|
||||
"""The deprecated timetable type must be refused"""
|
||||
response = client.post(
|
||||
"/api/v1/aa/",
|
||||
json={"name": "Legacy timetable AA", "type": "timetable", "is_active": True},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_update_access_auth_to_timetable_is_rejected(client, auth_headers, test_aa_os):
|
||||
"""PATCH must refuse type="timetable" """
|
||||
response = client.patch(
|
||||
f"/api/v1/aa/{test_aa_os.id}",
|
||||
json={"type": "timetable"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
still_fine = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
||||
assert still_fine.status_code == 200
|
||||
assert still_fine.json()["type"] == "oneshot"
|
||||
|
||||
|
||||
def test_get_all_access_auths(client, auth_headers, test_aa_os):
|
||||
"""Test retrieving all access authorizations."""
|
||||
response = client.get("/api/v1/aa/", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
aa_list = response.json()
|
||||
assert len(aa_list) >= 1
|
||||
|
||||
aa_names = [aa["name"] for aa in aa_list]
|
||||
assert test_aa_tt.name in aa_names
|
||||
assert test_aa_os.name in aa_names
|
||||
|
||||
|
||||
def test_get_access_auth_by_id(client, auth_headers, test_aa_tt):
|
||||
def test_get_access_auth_by_id(client, auth_headers, test_aa_os):
|
||||
"""Test retrieving a specific access authorization by ID."""
|
||||
response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
|
||||
response = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert data["id"] == test_aa_tt.id
|
||||
assert data["name"] == test_aa_tt.name
|
||||
assert data["id"] == test_aa_os.id
|
||||
assert data["name"] == test_aa_os.name
|
||||
|
||||
|
||||
def test_get_nonexistent_access_auth(client, auth_headers):
|
||||
@@ -107,10 +107,10 @@ def test_get_nonexistent_access_auth(client, auth_headers):
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_tt):
|
||||
def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_os):
|
||||
"""Test assigning an access authorization to a group."""
|
||||
response = client.put(
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
@@ -121,49 +121,49 @@ def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_t
|
||||
|
||||
|
||||
def test_assign_already_assigned_access_auth(
|
||||
client, auth_headers, test_group, test_aa_tt
|
||||
client, auth_headers, test_group, test_aa_os
|
||||
):
|
||||
"""Test assigning an already assigned access authorization."""
|
||||
# First assignment
|
||||
client.put(
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
||||
)
|
||||
|
||||
# Second assignment should indicate it's already assigned
|
||||
response = client.put(
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
||||
)
|
||||
# According to the code, this returns 409 with "already assigned" message
|
||||
assert response.status_code == 409
|
||||
assert "already assigned" in response.json()["detail"].lower()
|
||||
|
||||
|
||||
def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_tt):
|
||||
def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_os):
|
||||
"""Test unassigning an access authorization from a group."""
|
||||
# First assign
|
||||
client.put(
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
|
||||
f"/api/v1/aa/assign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
||||
)
|
||||
|
||||
# Then unassign
|
||||
response = client.put(
|
||||
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
|
||||
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_tt):
|
||||
def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_os):
|
||||
"""Test unassigning a non-existent assignment."""
|
||||
response = client.put(
|
||||
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
|
||||
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_os.id}", headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_assign_to_nonexistent_group(client, auth_headers, test_aa_tt):
|
||||
def test_assign_to_nonexistent_group(client, auth_headers, test_aa_os):
|
||||
"""Test assigning an AA to a non-existent group."""
|
||||
response = client.put(
|
||||
f"/api/v1/aa/assign/99999/{test_aa_tt.id}", headers=auth_headers
|
||||
f"/api/v1/aa/assign/99999/{test_aa_os.id}", headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
@@ -176,12 +176,12 @@ def test_assign_nonexistent_aa(client, auth_headers, test_group):
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_update_access_auth(client, auth_headers, test_aa_tt):
|
||||
def test_update_access_auth(client, auth_headers, test_aa_os):
|
||||
"""Test updating an access authorization."""
|
||||
update_data = {"name": "Updated AA", "is_active": False}
|
||||
|
||||
response = client.patch(
|
||||
f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers
|
||||
f"/api/v1/aa/{test_aa_os.id}", json=update_data, headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
@@ -190,23 +190,6 @@ def test_update_access_auth(client, auth_headers, test_aa_tt):
|
||||
assert data["is_active"] is False
|
||||
|
||||
|
||||
def test_update_access_auth_with_timetables(client, auth_headers, test_aa_tt):
|
||||
"""Test updating an access authorization with new timetables."""
|
||||
update_data = {
|
||||
"timetables": [{"weekday": 5, "starttime": "10:00", "duration": 120}]
|
||||
}
|
||||
|
||||
response = client.patch(
|
||||
f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers
|
||||
)
|
||||
assert response.status_code == 200
|
||||
jresponse = response.json()
|
||||
assert len(jresponse["timetables"]) == 1
|
||||
assert jresponse["timetables"][0]["weekday"] == 5
|
||||
assert jresponse["timetables"][0]["starttime"] == "10:00:00"
|
||||
assert jresponse["timetables"][0]["duration"] == 120
|
||||
|
||||
|
||||
def test_update_nonexistent_access_auth(client, auth_headers):
|
||||
"""Test updating a non-existent access authorization."""
|
||||
update_data = {"name": "Updated"}
|
||||
@@ -214,14 +197,14 @@ def test_update_nonexistent_access_auth(client, auth_headers):
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_delete_access_auth(client, auth_headers, test_aa_tt):
|
||||
def test_delete_access_auth(client, auth_headers, test_aa_os):
|
||||
"""Test deleting an access authorization."""
|
||||
response = client.delete(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
|
||||
response = client.delete(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
assert "deleted successfully" in response.json()["message"].lower()
|
||||
|
||||
# Verify AA is deleted
|
||||
response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers)
|
||||
response = client.get(f"/api/v1/aa/{test_aa_os.id}", headers=auth_headers)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@@ -231,12 +214,12 @@ def test_delete_nonexistent_access_auth(client, auth_headers):
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers):
|
||||
def test_aa_os_operations_by_non_admin(client, test_aa_os, user_auth_headers):
|
||||
"""Test that non-admin users cannot perform AA operations."""
|
||||
# Try to create an AA
|
||||
response = client.post(
|
||||
"/api/v1/aa/",
|
||||
json={"name": "test", "is_active": True, "timetables": []},
|
||||
json={"name": "test", "is_active": True},
|
||||
headers=user_auth_headers,
|
||||
)
|
||||
assert response.status_code == 403
|
||||
@@ -247,7 +230,7 @@ def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers):
|
||||
|
||||
# Try to assign AA
|
||||
response = client.put(
|
||||
f"/api/v1/aa/assign/1/{test_aa_tt.id}", headers=user_auth_headers
|
||||
f"/api/v1/aa/assign/1/{test_aa_os.id}", headers=user_auth_headers
|
||||
)
|
||||
assert response.status_code == 403
|
||||
|
||||
@@ -391,4 +374,3 @@ def test_oneshot_local_time_uses_the_dst_rule_of_its_own_date(
|
||||
assert datetime.fromisoformat(winter.json()["oneshot"]["ends_at"]) == datetime(
|
||||
2029, 12, 27, 11, 0, tzinfo=UTC
|
||||
)
|
||||
|
||||
|
||||
+22
-132
@@ -7,75 +7,10 @@ from app.model.models import (
|
||||
Card,
|
||||
GroupDB,
|
||||
OneShotAccess,
|
||||
Timetable,
|
||||
)
|
||||
from app.services.door import checkAccess
|
||||
|
||||
|
||||
def test_check_access_with_valid_timetable(db_session):
|
||||
# Setup: create card with valid access
|
||||
group = GroupDB(name="Test Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
|
||||
card = Card(
|
||||
key="test-key-123",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
timetable = Timetable(
|
||||
weekday=datetime.datetime.weekday(datetime.date.today()),
|
||||
starttime=datetime.datetime.now().time(),
|
||||
duration=120, # 2 hours
|
||||
)
|
||||
db_session.add(timetable)
|
||||
|
||||
aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
|
||||
db_session.add(aa)
|
||||
aa.timetables = [timetable]
|
||||
group.accessauths = [aa]
|
||||
|
||||
db_session.commit()
|
||||
|
||||
# Test: access should be granted within time window
|
||||
assert checkAccess("test-key-123", db_session)
|
||||
|
||||
|
||||
def test_check_access_outside_hours(db_session):
|
||||
# Test when current time is outside valid hours
|
||||
group = GroupDB(name="Test Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
|
||||
card = Card(
|
||||
key="test-key-123",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
timetable = Timetable(
|
||||
weekday=datetime.datetime.weekday(datetime.date.today()),
|
||||
starttime=datetime.time(1, 0),
|
||||
duration=1, # 2 hours
|
||||
)
|
||||
db_session.add(timetable)
|
||||
|
||||
aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
|
||||
db_session.add(aa)
|
||||
aa.timetables = [timetable]
|
||||
group.accessauths = [aa]
|
||||
|
||||
db_session.commit()
|
||||
assert not checkAccess("test-key-123", db_session)
|
||||
|
||||
|
||||
def test_check_access_with_valid_oneshot(db_session):
|
||||
# Setup: create card with valid access
|
||||
group = GroupDB(name="Test Group")
|
||||
@@ -129,16 +64,13 @@ def test_check_access_with_inactive_aa(db_session):
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
timetable = Timetable(
|
||||
weekday=datetime.datetime.weekday(datetime.date.today()),
|
||||
starttime=datetime.datetime.now().time(),
|
||||
duration=120, # 2 hours
|
||||
oneshot = OneShotAccess(
|
||||
uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
|
||||
)
|
||||
db_session.add(timetable)
|
||||
|
||||
aa = AccessAuthorizationDB(name="Test AA", is_active=False, type="timetable")
|
||||
db_session.add(oneshot)
|
||||
aa = AccessAuthorizationDB(name="Test oneshot AA", is_active=False, type="oneshot")
|
||||
db_session.add(aa)
|
||||
aa.timetables = [timetable]
|
||||
aa.oneshot = oneshot
|
||||
group.accessauths = [aa]
|
||||
|
||||
db_session.commit()
|
||||
@@ -161,16 +93,13 @@ def test_check_access_with_inactive_card(db_session):
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
timetable = Timetable(
|
||||
weekday=datetime.datetime.weekday(datetime.date.today()),
|
||||
starttime=datetime.datetime.now().time(),
|
||||
duration=120, # 2 hours
|
||||
oneshot = OneShotAccess(
|
||||
uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
|
||||
)
|
||||
db_session.add(timetable)
|
||||
|
||||
aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
|
||||
db_session.add(oneshot)
|
||||
aa = AccessAuthorizationDB(name="Test oneshot AA", is_active=True, type="oneshot")
|
||||
db_session.add(aa)
|
||||
aa.timetables = [timetable]
|
||||
aa.oneshot = oneshot
|
||||
group.accessauths = [aa]
|
||||
|
||||
db_session.commit()
|
||||
@@ -193,27 +122,23 @@ def test_regression_check_access_issue_28(db_session):
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
# Creating timetable with inactive AA
|
||||
timetable1 = Timetable(
|
||||
weekday=datetime.datetime.weekday(datetime.date.today()),
|
||||
starttime=datetime.datetime.now().time(),
|
||||
duration=120, # 2 hours
|
||||
# Creating oneshot with inactive AA
|
||||
oneshot1 = OneShotAccess(
|
||||
uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
|
||||
)
|
||||
db_session.add(timetable1)
|
||||
aa1 = AccessAuthorizationDB(name="First AA", is_active=False, type="timetable")
|
||||
db_session.add(oneshot1)
|
||||
aa1 = AccessAuthorizationDB(name="First AA", is_active=False, type="oneshot")
|
||||
db_session.add(aa1)
|
||||
aa1.timetables = [timetable1]
|
||||
aa1.oneshot = oneshot1
|
||||
|
||||
# Creating timetable with active AA
|
||||
timetable2 = Timetable(
|
||||
weekday=datetime.datetime.weekday(datetime.date.today()),
|
||||
starttime=datetime.datetime.now().time(),
|
||||
duration=120, # 2 hours
|
||||
# Creating oneshot with active AA
|
||||
oneshot2 = OneShotAccess(
|
||||
uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
|
||||
)
|
||||
db_session.add(timetable2)
|
||||
aa2 = AccessAuthorizationDB(name="Second AA", is_active=True, type="timetable")
|
||||
db_session.add(oneshot2)
|
||||
aa2 = AccessAuthorizationDB(name="Second AA", is_active=True, type="oneshot")
|
||||
db_session.add(aa2)
|
||||
aa2.timetables = [timetable2]
|
||||
aa2.oneshot = oneshot2
|
||||
|
||||
group.accessauths = [aa1, aa2]
|
||||
|
||||
@@ -269,38 +194,3 @@ def test_check_access_oneshot_expired(db_session, system_tz):
|
||||
_create_card_with_oneshot(db_session, ends_at)
|
||||
|
||||
assert not checkAccess("test-key-123", db_session)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("system_tz", [BERLIN_TZ], indirect=True)
|
||||
def test_check_access_timetable_window_is_local_wall_clock(db_session, system_tz):
|
||||
"""Timetables are local wall-clock times and still work with an aware door clock."""
|
||||
now_local = datetime.datetime.now().astimezone()
|
||||
|
||||
group = GroupDB(name="Test Group")
|
||||
db_session.add(group)
|
||||
db_session.commit()
|
||||
|
||||
card = Card(
|
||||
key="test-key-123",
|
||||
group_id=group.id,
|
||||
enabled=True,
|
||||
name="test_card",
|
||||
card_serial="00:00:00:00:00:00:00",
|
||||
)
|
||||
db_session.add(card)
|
||||
|
||||
timetable = Timetable(
|
||||
weekday=now_local.weekday(),
|
||||
starttime=now_local.time(),
|
||||
duration=120, # 2 hours
|
||||
)
|
||||
db_session.add(timetable)
|
||||
|
||||
aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable")
|
||||
db_session.add(aa)
|
||||
aa.timetables = [timetable]
|
||||
group.accessauths = [aa]
|
||||
|
||||
db_session.commit()
|
||||
|
||||
assert checkAccess("test-key-123", db_session)
|
||||
|
||||
Reference in New Issue
Block a user