malobeo/gatekeeper
3
0
Fork 0
Code Issues 7 Pull Requests 2 Actions Activity

Merge pull request 'Change accessauth, timetable model' (#16) from change_tt_model into master

Browse Source 
Reviewed-on: #16
Fixes #9
This commit was merged in pull request #16.
This commit is contained in:
ahtlon
2026-07-27 17:19:58 +02:00
parent 1883662b5f ea2a1f916d
commit 5f1030c831
7 changed files with 186 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
app/controllers/aaManager.py
View File

@@ -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)
@@ -68,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)

41
app/model/models.py
View File

@@ -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,25 +61,57 @@ 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):
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

40
app/services/door.py
View File

@@ -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

14
test/conftest.py
View File

@@ -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()

5
test/test_models.py
View File

@@ -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"

94
test/test_services/test_aa_manager.py
View File

@@ -2,10 +2,11 @@ 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": [
{"weekday": 1, "starttime": "08:00", "duration": 60},
@@ -17,13 +18,46 @@ 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"
}
}
def test_get_all_access_auths(client, auth_headers, test_aa):
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."""
response = client.get("/api/v1/aa/", headers=auth_headers)
assert response.status_code == 200
@@ -32,17 +66,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 +85,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 +99,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 +114,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 +148,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 +156,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 +167,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 +176,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 +195,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 +212,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 +227,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

33
test/test_services/test_door.py
View File

@@ -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
@@ -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]
@@ -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):