From 47b84a097c1afbf2b39f944ab51e33097e048446 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Fri, 3 Jul 2026 22:02:34 +0200 Subject: [PATCH 1/6] [Models] This doesn't quite work yet but this is about the model im thinking of for the future --- app/model/models.py | 39 +++++++++++++++++++++++++++++++++++---- 1 file changed, 35 insertions(+), 4 deletions(-) diff --git a/app/model/models.py b/app/model/models.py index 7157b94..cced809 100644 --- a/app/model/models.py +++ b/app/model/models.py @@ -1,6 +1,7 @@ from sqlmodel import Field, Relationship, Session, SQLModel -from typing import List -from datetime import time +from typing import List, Literal, Union +from datetime import datetime, time +from pydantic import model_validator class Base(SQLModel): pass @@ -60,19 +61,49 @@ class GroupResponse(GroupBase): #### AccessAuthorization class AccessAuthorizationBase(Base): name: str = Field(index=True) + type: Literal["timetable", "oneshot", "somefuturespec"] is_active: bool class AccessAuthorizationDB(AccessAuthorizationBase, table=True): id: int | None = Field(default=None, primary_key=True) + type: str 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) + +class OneShotAccessBase(Base): + uses: int = 1 + ends_at: datetime + +class OneShotAccess(OneShotAccessBase, 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="oneshot") class AccessAuthorizationCreate(AccessAuthorizationBase): - timetables: List["TimetableCreate"] + 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 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 class AccessAuthorizationResponse(AccessAuthorizationBase): id: int - timetables: List["Timetable"] + timetables: List["Timetable"] = [] + oneshot: OneShotAccessBase | None = None groups: List["GroupDB"] class AccessAuthorizationUpdate(Base): From e45abb5c55a1c42e38ae8d0c51e87f4dd67008d2 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Mon, 27 Jul 2026 15:33:39 +0200 Subject: [PATCH 2/6] [AA] fix the db insert --- app/controllers/aaManager.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/app/controllers/aaManager.py b/app/controllers/aaManager.py index 4f9a1d6..a697e43 100644 --- a/app/controllers/aaManager.py +++ b/app/controllers/aaManager.py @@ -15,11 +15,18 @@ aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"]) @aa_router.post("/", response_model=AccessAuthorizationResponse) def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate, admin: bool = Depends(auth_is_admin)): logger.info(f"Creating accessauth with data: {aa}") - timetables = [Timetable.model_validate(t) for t in aa.timetables] + if aa.timetables is not []: + 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: oneshot = None db_aa = AccessAuthorizationDB( name=aa.name, + type=aa.type, is_active=aa.is_active, - timetables=timetables + timetables=timetables, + oneshot=oneshot ) return add_and_refresh(db, db_aa) From 5f19409da41d94744169896a75bb178ed07ed0ce Mon Sep 17 00:00:00 2001 From: ahtlon Date: Mon, 27 Jul 2026 16:26:58 +0200 Subject: [PATCH 3/6] [AA] Add oneshot to checkAccess function --- app/services/door.py | 40 +++++++++++++++++++++++++++------------- 1 file changed, 27 insertions(+), 13 deletions(-) diff --git a/app/services/door.py b/app/services/door.py index c4d1faf..f4d3548 100644 --- a/app/services/door.py +++ b/app/services/door.py @@ -4,9 +4,9 @@ from sqlmodel import select from fastapi import Depends, HTTPException, status from sqlalchemy.orm import selectinload import sqlalchemy.exc as exc -import datetime +from datetime import datetime, date, timedelta -from app.services.database import Session, get_session +from app.services.database import Session, get_session, add_and_refresh from app.model.models import * doorIsOpen = True @@ -28,22 +28,36 @@ def closeDoor(): def isDoorOpen(): return doorIsOpen +def decrementOneshot(db: Session, oneshot: OneShotAccess): + data = oneshot.model_dump() + if data["uses"] > 0: + data["uses"] = data["uses"] - 1 + oneshot.sqlmodel_update(oneshot, update=data) + add_and_refresh(db, oneshot) + def checkAccess(key: str, db: Session): try: - current_weekday = datetime.datetime.weekday(datetime.date.today()) - current_time = datetime.datetime.now() + current_weekday = datetime.weekday(date.today()) + current_time = datetime.now() card = db.exec(select(Card).where(Card.key == key)).one() for auth in card.group.accessauths: logger.info(f"checking auth: {auth.name}") - for timetable in auth.timetables: - logger.info(f" checking timetable {timetable.id}") - logger.info(f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}") - if current_weekday == timetable.weekday: - starttime = datetime.datetime.combine(datetime.date.today(), timetable.starttime) - endtime = starttime + datetime.timedelta(minutes=timetable.duration) - logger.info(f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}") - if starttime < current_time < endtime: - logger.info("Access Valid!") + if auth.type == "timetable": + for timetable in auth.timetables: + logger.info(f" checking timetable {timetable.id}") + logger.info(f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}") + if current_weekday == timetable.weekday: + starttime = datetime.combine(date.today(), timetable.starttime) + endtime = starttime + timedelta(minutes=timetable.duration) + logger.info(f" comparing time: Start:{starttime} 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}') + if current_time < auth.oneshot.ends_at: + if auth.oneshot.uses > 0: + decrementOneshot(db, auth.oneshot) return True logger.info("No more auths found") return False From 4b87603e06e6365c919254f04373a2f5a2923ac3 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Mon, 27 Jul 2026 16:40:45 +0200 Subject: [PATCH 4/6] [AA] fix patching AA with oneshots --- app/controllers/aaManager.py | 6 +++++- app/model/models.py | 2 ++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/app/controllers/aaManager.py b/app/controllers/aaManager.py index a697e43..e256051 100644 --- a/app/controllers/aaManager.py +++ b/app/controllers/aaManager.py @@ -75,12 +75,16 @@ def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: Acc db_aa = db.get(AccessAuthorizationDB, aa_id) if db_aa is None: raise HTTPException(status_code=404, detail="AccessAuthorization not found") - aa_data = aa.model_dump(exclude_unset=True) + 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 + aa_data.pop("oneshot") db_aa.sqlmodel_update(aa_data) return add_and_refresh(db, db_aa) diff --git a/app/model/models.py b/app/model/models.py index cced809..b198e15 100644 --- a/app/model/models.py +++ b/app/model/models.py @@ -108,8 +108,10 @@ class AccessAuthorizationResponse(AccessAuthorizationBase): class AccessAuthorizationUpdate(Base): name: str | None = None + type: Literal["timetable", "oneshot", "somefuturespec"] | None = None is_active: bool | None = None timetables: List["TimetableCreate"] | None = None + oneshot: OneShotAccessBase | None = None #### Card From 1efbad1db3e746cb19c8fb30a8d963b6d4af0385 Mon Sep 17 00:00:00 2001 From: ahtlon Date: Mon, 27 Jul 2026 17:01:01 +0200 Subject: [PATCH 5/6] [Tests] Fix tests for new model --- test/conftest.py | 14 +++++-- test/test_models.py | 5 ++- test/test_services/test_aa_manager.py | 55 ++++++++++++++------------- test/test_services/test_door.py | 4 +- 4 files changed, 44 insertions(+), 34 deletions(-) diff --git a/test/conftest.py b/test/conftest.py index 70bf024..a60fa9d 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -3,6 +3,7 @@ from fastapi.testclient import TestClient from sqlmodel import Session, create_engine, SQLModel from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool +from datetime import time from app.main import app from app.model.models import UserDB, Card, GroupDB, AccessAuthorizationDB, Timetable, AaGroupLink @@ -107,11 +108,18 @@ def test_card(db_session, test_group): @pytest.fixture -def test_aa(db_session): - """Create a test access authorization.""" +def test_aa_tt(db_session): + """Create a test access authorization with timetable.""" + tt = Timetable( + weekday=1, + starttime=time(1, 0, 0, 0), + duration=50 + ) aa = AccessAuthorizationDB( name="Test AA", - is_active=True + is_active=True, + type="timetable", + timetables=[tt] ) db_session.add(aa) db_session.commit() diff --git a/test/test_models.py b/test/test_models.py index 85e3ac9..f30710f 100644 --- a/test/test_models.py +++ b/test/test_models.py @@ -41,7 +41,7 @@ def test_group_models(): def test_access_authorization_models(): """Test access authorization model creation and validation.""" # Test AccessAuthorizationBase - aa_base = AccessAuthorizationBase(name="Test AA", is_active=True) + aa_base = AccessAuthorizationBase(name="Test AA", is_active=True, type="timetable") assert aa_base.name == "Test AA" assert aa_base.is_active is True @@ -49,7 +49,8 @@ def test_access_authorization_models(): timetable_create = TimetableCreate(weekday=1, starttime="08:00", duration=60) aa_create = AccessAuthorizationCreate( name="New AA", - is_active=False, + is_active=False, + type="timetable", timetables=[timetable_create] ) assert aa_create.name == "New AA" diff --git a/test/test_services/test_aa_manager.py b/test/test_services/test_aa_manager.py index 3bfdb3c..0fae33e 100644 --- a/test/test_services/test_aa_manager.py +++ b/test/test_services/test_aa_manager.py @@ -6,6 +6,7 @@ def test_create_access_auth(client, auth_headers): """Test creating a new access authorization.""" aa_data = { "name": "New AA", + "type": "timetable", "is_active": True, "timetables": [ {"weekday": 1, "starttime": "08:00", "duration": 60}, @@ -23,7 +24,7 @@ def test_create_access_auth(client, auth_headers): assert len(data["timetables"]) == 2 -def test_get_all_access_auths(client, auth_headers, test_aa): +def test_get_all_access_auths(client, auth_headers, test_aa_tt): """Test retrieving all access authorizations.""" response = client.get("/api/v1/aa/", headers=auth_headers) assert response.status_code == 200 @@ -32,17 +33,17 @@ def test_get_all_access_auths(client, auth_headers, test_aa): assert len(aa_list) >= 1 aa_names = [aa["name"] for aa in aa_list] - assert test_aa.name in aa_names + assert test_aa_tt.name in aa_names -def test_get_access_auth_by_id(client, auth_headers, test_aa): +def test_get_access_auth_by_id(client, auth_headers, test_aa_tt): """Test retrieving a specific access authorization by ID.""" - response = client.get(f"/api/v1/aa/{test_aa.id}", headers=auth_headers) + response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers) assert response.status_code == 200 data = response.json() - assert data["id"] == test_aa.id - assert data["name"] == test_aa.name + assert data["id"] == test_aa_tt.id + assert data["name"] == test_aa_tt.name def test_get_nonexistent_access_auth(client, auth_headers): @@ -51,10 +52,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): +def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_tt): """Test assigning an access authorization to a group.""" response = client.put( - f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", + f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) assert response.status_code == 200 @@ -65,14 +66,14 @@ def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa): # Note: The response model might not include the full relationship -def test_assign_already_assigned_access_auth(client, auth_headers, test_group, test_aa): +def test_assign_already_assigned_access_auth(client, auth_headers, test_group, test_aa_tt): """Test assigning an already assigned access authorization.""" # First assignment - client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", headers=auth_headers) + client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.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.id}", + f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) # According to the code, this returns 409 with "already assigned" message @@ -80,31 +81,31 @@ def test_assign_already_assigned_access_auth(client, auth_headers, test_group, t assert "already assigned" in response.json()["detail"].lower() -def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa): +def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_aa_tt): """Test unassigning an access authorization from a group.""" # First assign - client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa.id}", headers=auth_headers) + client.put(f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers) # Then unassign response = client.put( - f"/api/v1/aa/unassign/{test_group.id}/{test_aa.id}", + f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) assert response.status_code == 200 -def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa): +def test_unassign_nonexistent_assignment(client, auth_headers, test_group, test_aa_tt): """Test unassigning a non-existent assignment.""" response = client.put( - f"/api/v1/aa/unassign/{test_group.id}/{test_aa.id}", + f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers ) assert response.status_code == 404 -def test_assign_to_nonexistent_group(client, auth_headers, test_aa): +def test_assign_to_nonexistent_group(client, auth_headers, test_aa_tt): """Test assigning an AA to a non-existent group.""" - response = client.put(f"/api/v1/aa/assign/99999/{test_aa.id}", headers=auth_headers) + response = client.put(f"/api/v1/aa/assign/99999/{test_aa_tt.id}", headers=auth_headers) assert response.status_code == 404 @@ -114,7 +115,7 @@ 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): +def test_update_access_auth(client, auth_headers, test_aa_tt): """Test updating an access authorization.""" update_data = { "name": "Updated AA", @@ -122,7 +123,7 @@ def test_update_access_auth(client, auth_headers, test_aa): } response = client.patch( - f"/api/v1/aa/{test_aa.id}", + f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers ) @@ -133,7 +134,7 @@ def test_update_access_auth(client, auth_headers, test_aa): assert data["is_active"] is False -def test_update_access_auth_with_timetables(client, auth_headers, test_aa): +def test_update_access_auth_with_timetables(client, auth_headers, test_aa_tt): """Test updating an access authorization with new timetables.""" update_data = { "timetables": [ @@ -142,7 +143,7 @@ def test_update_access_auth_with_timetables(client, auth_headers, test_aa): } response = client.patch( - f"/api/v1/aa/{test_aa.id}", + f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers ) @@ -161,14 +162,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): +def test_delete_access_auth(client, auth_headers, test_aa_tt): """Test deleting an access authorization.""" - response = client.delete(f"/api/v1/aa/{test_aa.id}", headers=auth_headers) + response = client.delete(f"/api/v1/aa/{test_aa_tt.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.id}", headers=auth_headers) + response = client.get(f"/api/v1/aa/{test_aa_tt.id}", headers=auth_headers) assert response.status_code == 404 @@ -178,7 +179,7 @@ def test_delete_nonexistent_access_auth(client, auth_headers): assert response.status_code == 404 -def test_aa_operations_by_non_admin(client, test_aa, user_auth_headers): +def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers): """Test that non-admin users cannot perform AA operations.""" # Try to create an AA response = client.post( @@ -193,5 +194,5 @@ def test_aa_operations_by_non_admin(client, test_aa, user_auth_headers): assert response.status_code == 403 # Try to assign AA - response = client.put(f"/api/v1/aa/assign/1/{test_aa.id}", headers=user_auth_headers) + response = client.put(f"/api/v1/aa/assign/1/{test_aa_tt.id}", headers=user_auth_headers) assert response.status_code == 403 diff --git a/test/test_services/test_door.py b/test/test_services/test_door.py index 166f3e0..589c7df 100644 --- a/test/test_services/test_door.py +++ b/test/test_services/test_door.py @@ -19,7 +19,7 @@ def test_check_access_with_valid_timetable(db_session): ) db_session.add(timetable) - aa = AccessAuthorizationDB(name="Test AA", is_active=True) + aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable") db_session.add(aa) aa.timetables = [timetable] group.accessauths = [aa] @@ -46,7 +46,7 @@ def test_check_access_outside_hours(db_session): ) db_session.add(timetable) - aa = AccessAuthorizationDB(name="Test AA", is_active=True) + aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="timetable") db_session.add(aa) aa.timetables = [timetable] group.accessauths = [aa] From ea2a1f916d2a1b50d93753a751736d923e5957da Mon Sep 17 00:00:00 2001 From: ahtlon Date: Mon, 27 Jul 2026 17:18:04 +0200 Subject: [PATCH 6/6] [Tests] add new tests for oneshot type aa --- test/test_services/test_aa_manager.py | 39 ++++++++++++++++++++++++--- test/test_services/test_door.py | 29 +++++++++++++++++++- 2 files changed, 64 insertions(+), 4 deletions(-) diff --git a/test/test_services/test_aa_manager.py b/test/test_services/test_aa_manager.py index 0fae33e..1714257 100644 --- a/test/test_services/test_aa_manager.py +++ b/test/test_services/test_aa_manager.py @@ -2,10 +2,10 @@ import pytest from fastapi import status -def test_create_access_auth(client, auth_headers): +def test_create_access_auth_tt(client, auth_headers): """Test creating a new access authorization.""" aa_data = { - "name": "New AA", + "name": "New tt_AA", "type": "timetable", "is_active": True, "timetables": [ @@ -18,11 +18,44 @@ def test_create_access_auth(client, auth_headers): assert response.status_code == 200 data = response.json() - assert data["name"] == "New AA" + 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 = { + "name": "New os_AA", + "type": "oneshot", + "is_active": True, + "oneshot": { + "uses": 1, + "ends_at": "2029-07-27" + } + } + + response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers) + assert response.status_code == 200 + + data = response.json() + assert data["name"] == "New os_AA" + assert data["type"] == "oneshot" + assert data["is_active"] is True + assert "id" in data + assert data["oneshot"]["uses"] == 1 + +def test_create_wrong_aa_type(client, auth_headers): + """Test creating a new access authorization with oneshot type.""" + aa_data = { + "name": "New AA", + "type": "wrong", + "is_active": True, + } + + response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers) + assert response.status_code == 422 def test_get_all_access_auths(client, auth_headers, test_aa_tt): """Test retrieving all access authorizations.""" diff --git a/test/test_services/test_door.py b/test/test_services/test_door.py index 589c7df..7e332e8 100644 --- a/test/test_services/test_door.py +++ b/test/test_services/test_door.py @@ -1,7 +1,7 @@ import pytest import datetime from app.services.door import checkAccess -from app.model.models import Card, GroupDB, AccessAuthorizationDB, Timetable +from app.model.models import Card, GroupDB, AccessAuthorizationDB, Timetable, OneShotAccess def test_check_access_with_valid_timetable(db_session): # Setup: create card with valid access @@ -55,6 +55,33 @@ def test_check_access_outside_hours(db_session): result = checkAccess("test-key-123", db_session) assert result == False +def test_check_access_with_valid_oneshot(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) + + oneshot = OneShotAccess( + uses=1, + ends_at=datetime.datetime.now() + datetime.timedelta(days=1) + ) + db_session.add(oneshot) + + aa = AccessAuthorizationDB(name="Test AA", is_active=True, type="oneshot") + db_session.add(aa) + aa.oneshot = oneshot + group.accessauths = [aa] + + db_session.commit() + + # Test: access should be granted within time window + result = checkAccess("test-key-123", db_session) + assert result == True + assert aa.oneshot.uses == 0 + def test_check_access_invalid_card(db_session): # Should raise exception for non-existent card with pytest.raises(Exception):