Card model changes #14

Merged
ahtlon merged 9 commits from change_card_model into master 2026-07-01 17:28:04 +02:00
11 changed files with 93 additions and 55 deletions
+34 -15
View File
@@ -5,7 +5,7 @@ from sqlmodel import Session, select
from typing import List from typing import List
from sqlalchemy.exc import NoResultFound from sqlalchemy.exc import NoResultFound
from ..model.models import Card from ..model.models import Card, CardCreate, CardUpdate, GroupDB
from ..services.database import engine, get_session, add_and_refresh from ..services.database import engine, get_session, add_and_refresh
from ..services.auth import auth_is_admin from ..services.auth import auth_is_admin
import uuid as gen_uuid import uuid as gen_uuid
@@ -13,28 +13,40 @@ from app.services.scanner import WriteNewCard, DeleteCard
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"]) card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
def register_card(group_id: int): def register_card(cardInput: CardCreate):
key = WriteNewCard() key, uid = WriteNewCard()
if key == None: if key == None:
logger.info("No card registered. Check logs!") logger.info("No card registered. Check logs!")
raise HTTPException(status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!") raise HTTPException(status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!")
card = Card(group_id=group_id, uuid=key) card = Card(
group_id=cardInput.group_id,
key=key,
name=cardInput.name,
card_serial=uid,
enabled=cardInput.enabled
)
return card return card
@card_router.post("/{group_id}", response_model=Card) @card_router.post("/", response_model=Card)
def add_card(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)): def add_card(cardInput: CardCreate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
card = register_card(group_id) try: assert db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() == None
except AssertionError:
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None
except AssertionError:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!")
card = register_card(cardInput)
return add_and_refresh(db, card) return add_and_refresh(db, card)
@card_router.get("/delete") @card_router.delete("/")
def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
key = DeleteCard() key = DeleteCard()
logger.info(key) logger.info(key)
try: try:
card = db.exec(select(Card).where(Card.uuid == key)).one() card = db.exec(select(Card).where(Card.key == key)).one()
except NoResultFound: except NoResultFound:
logger.info(f"The key:'{key}' was not found in db!") logger.info(f"The key:'{key}' was not found in db!")
raise HTTPException(status_code=500, detail="Key on card not found in DB. Please tell an admin about this. KEY={key}") raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Key on card not found in DB. Please tell an admin about this. KEY={key}")
db.delete(card) db.delete(card)
db.commit() db.commit()
return {"message": "Card deleted successfully"} return {"message": "Card deleted successfully"}
@@ -44,8 +56,15 @@ def get_cards(*, db: Session = Depends(get_session), group_id: int, admin: bool
cards = db.exec(select(Card).where(Card.group_id == group_id)).all() cards = db.exec(select(Card).where(Card.group_id == group_id)).all()
return cards return cards
@card_router.patch("/{card_id}", response_model=Card)
#TODO: def update_card(card_id: int, cardInput: CardUpdate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
# -Split Authorisations + Cards db_card = db.get(Card, card_id)
# -Deactivation if db_card is None:
# -Deleting raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Card not found!")
card_data = cardInput.model_dump(exclude_unset=True, exclude_none=True)
if "group_id" in card_data:
try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None
except AssertionError:
raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!")
db_card.sqlmodel_update(card_data)
return add_and_refresh(db, db_card)
+22 -9
View File
@@ -3,6 +3,7 @@ logger = logging.getLogger(__name__)
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from app.services.auth import auth_is_admin from app.services.auth import auth_is_admin
from sqlmodel import Session, select from sqlmodel import Session, select
import sqlalchemy.exc as exc
from app.model.models import * from app.model.models import *
from app.services.database import get_session, add_and_refresh from app.services.database import get_session, add_and_refresh
@@ -12,19 +13,31 @@ debug_router = APIRouter(
dependencies=[Depends(auth_is_admin)], dependencies=[Depends(auth_is_admin)],
) )
@debug_router.get("/addcard/{groupid}/{card_key}") @debug_router.put("/addcard/")
def add_card_manually(groupid: int, card_key: str, db: Session=Depends(get_session)): def add_card_manually(groupid: int, card_key: str, name: str, enabled: bool, db: Session=Depends(get_session)):
"""Add cards manually (you also have to delete them manually)""" """Add cards manually (you also have to delete them manually)"""
logger.critical(f"Manual db change: adding a card with key: {card_key} to group: {groupid}") logger.critical(f"Manual db change: adding a card with key: {card_key} to group: {groupid}")
card = Card(group_id=groupid, uuid=card_key) card = Card(
group_id=groupid,
key=card_key,
name=name,
enabled=enabled,
card_serial="00:00:00:00:00:00:00"
)
add_and_refresh(db, card) add_and_refresh(db, card)
return card return card
@debug_router.get("/rmcard/{card_key}") @debug_router.get("/rmcard/{card_id}")
def remove_card_manually(card_key: str, db: Session = Depends(get_session)): def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
logger.critical(f"Manual db change: removing a card with key: {card_key}") try:
card = db.exec(select(Card).where(Card.uuid == card_key)).one() card = db.exec(select(Card).where(Card.id == card_id)).one()
add_and_refresh(db, card) except exc.NoResultFound:
raise HTTPException(status_code=500, detail="No card with that id found.")
logger.critical(f"Manual db change: removing a card with attrs: {card}")
db.delete(card)
db.commit()
return {"message": "Card deleted successfully"}
@debug_router.put("/getcards") @debug_router.put("/getcards")
def list_all_cards(db: Session = Depends(get_session)): def list_all_cards(db: Session = Depends(get_session)):
@@ -33,5 +46,5 @@ def list_all_cards(db: Session = Depends(get_session)):
print(cards) print(cards)
out = [] out = []
for i in cards: for i in cards:
out.append({i.uuid: i.group.name}) out.append({i.key: i.group.name})
return out return out
+4 -2
View File
@@ -1,6 +1,6 @@
import logging import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
from fastapi import APIRouter, HTTPException, Depends from fastapi import APIRouter, HTTPException, Depends, status
from sqlmodel import Session, select from sqlmodel import Session, select
from typing import List from typing import List
@@ -12,8 +12,10 @@ user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
@user_router.post("/", response_model=UserResponse) @user_router.post("/", response_model=UserResponse)
def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin: bool = Depends(auth_is_admin)): def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin: bool = Depends(auth_is_admin)):
logger.info(f"creating user with data: {user}")
hashed_password = {"passwordhash": get_password_hash(user.password)} hashed_password = {"passwordhash": get_password_hash(user.password)}
try: assert db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none() == None
except AssertionError:
raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!")
db_user = UserDB.model_validate(user, update=hashed_password) db_user = UserDB.model_validate(user, update=hashed_password)
return add_and_refresh(db, db_user) return add_and_refresh(db, db_user)
+15 -2
View File
@@ -7,7 +7,7 @@ class Base(SQLModel):
#### User #### User
class UserBase(Base): class UserBase(Base):
name: str = Field(index=True) name: str = Field(index=True, unique=True)
email: str | None = None email: str | None = None
is_admin: bool = False is_admin: bool = False
@@ -84,10 +84,23 @@ class AccessAuthorizationUpdate(Base):
#### Card #### Card
class Card(Base, table=True): class Card(Base, table=True):
id: int | None = Field(default=None, primary_key=True) id: int | None = Field(default=None, primary_key=True)
uuid: str key: str
card_serial: str
enabled: bool = True
name: str = Field(unique=True, max_length=32)
group_id: int | None = Field(default=None, foreign_key="groupdb.id") group_id: int | None = Field(default=None, foreign_key="groupdb.id")
group: GroupDB | None = Relationship(back_populates="cards") group: GroupDB | None = Relationship(back_populates="cards")
class CardCreate(Base):
name: str = Field(unique=True, max_length=32)
enabled: bool = True
group_id: int
class CardUpdate(Base):
name: str | None = None
enabled: bool | None = None
group_id: int | None = None
class TimetableBase(Base): class TimetableBase(Base):
weekday: int = Field(le=6, ge=0) weekday: int = Field(le=6, ge=0)
starttime: time starttime: time
+2 -2
View File
@@ -28,11 +28,11 @@ def closeDoor():
def isDoorOpen(): def isDoorOpen():
return doorIsOpen return doorIsOpen
def checkAccess(uuid: str, db: Session): def checkAccess(key: str, db: Session):
try: try:
current_weekday = datetime.datetime.weekday(datetime.date.today()) current_weekday = datetime.datetime.weekday(datetime.date.today())
current_time = datetime.datetime.now() current_time = datetime.datetime.now()
card = db.exec(select(Card).where(Card.uuid == uuid)).one() card = db.exec(select(Card).where(Card.key == key)).one()
for auth in card.group.accessauths: for auth in card.group.accessauths:
logger.info(f"checking auth: {auth.name}") logger.info(f"checking auth: {auth.name}")
for timetable in auth.timetables: for timetable in auth.timetables:
+1 -1
View File
@@ -233,7 +233,7 @@ def WriteNewCard():
assert rdata == get_list(key) assert rdata == get_list(key)
logger.debug(" - Data written successfully.") logger.debug(" - Data written successfully.")
scannerThread.start() scannerThread.start()
return key return key, to_hex_string(data=uid, separator=":")
except Exception as e: except Exception as e:
logger.error(f"Error in write function: {e}", exc_info=True) logger.error(f"Error in write function: {e}", exc_info=True)
+1 -1
View File
@@ -99,7 +99,7 @@ def test_group(db_session):
@pytest.fixture @pytest.fixture
def test_card(db_session, test_group): def test_card(db_session, test_group):
"""Create a test card.""" """Create a test card."""
card = Card(uuid="test-uuid-123", group_id=test_group.id) card = Card(key="test-key-123", group_id=test_group.id, enabled=True, name="test_card", card_serial="00:00:00:00:00:00:00")
db_session.add(card) db_session.add(card)
db_session.commit() db_session.commit()
db_session.refresh(card) db_session.refresh(card)
+2 -2
View File
@@ -59,8 +59,8 @@ def test_access_authorization_models():
def test_card_model(): def test_card_model():
"""Test card model creation and validation.""" """Test card model creation and validation."""
card = Card(uuid="test-uuid", group_id=1) card = Card(key="test-key", group_id=1)
assert card.uuid == "test-uuid" assert card.key == "test-key"
assert card.group_id == 1 assert card.group_id == 1
+1 -1
View File
@@ -23,7 +23,7 @@ def test_get_cards_for_nonexistent_group(client, auth_headers):
def test_card_operations_by_non_admin(client, test_group, user_auth_headers): def test_card_operations_by_non_admin(client, test_group, user_auth_headers):
"""Test that non-admin users cannot perform card operations.""" """Test that non-admin users cannot perform card operations."""
# Try to add a card # Try to add a card
response = client.post(f"/api/v1/cards/{test_group.id}", headers=user_auth_headers) response = client.post(f"/api/v1/cards/", headers=user_auth_headers)
assert response.status_code == 403 assert response.status_code == 403
# Try to get cards # Try to get cards
+6 -15
View File
@@ -26,25 +26,16 @@ def test_create_db_and_tables():
def test_get_session(db_session): def test_get_session(db_session):
"""Test database session generator.""" """Test database session generator."""
# Test that we can get a session # Test that we can get a session
session_gen = get_session() assert isinstance(db_session, Session)
session = next(session_gen)
assert isinstance(session, Session)
# Test that session works # Test that session works
user = UserDB(name="Test", passwordhash="hash") user = UserDB(name="Test_User", passwordhash="hash")
session.add(user) db_session.add(user)
session.commit() db_session.commit()
retrieved_user = session.get(UserDB, user.id) retrieved_user = db_session.get(UserDB, user.id)
assert retrieved_user is not None assert retrieved_user is not None
assert retrieved_user.name == "Test" assert retrieved_user.name == "Test_User"
# Clean up generator
try:
next(session_gen)
except StopIteration:
pass
def test_add_and_refresh(db_session): def test_add_and_refresh(db_session):
+5 -5
View File
@@ -9,7 +9,7 @@ def test_check_access_with_valid_timetable(db_session):
db_session.add(group) db_session.add(group)
db_session.commit() db_session.commit()
card = Card(uuid="test-uuid-123", group_id=group.id) 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) db_session.add(card)
timetable = Timetable( timetable = Timetable(
@@ -27,7 +27,7 @@ def test_check_access_with_valid_timetable(db_session):
db_session.commit() db_session.commit()
# Test: access should be granted within time window # Test: access should be granted within time window
result = checkAccess("test-uuid-123", db_session) result = checkAccess("test-key-123", db_session)
assert result == True assert result == True
def test_check_access_outside_hours(db_session): def test_check_access_outside_hours(db_session):
@@ -36,7 +36,7 @@ def test_check_access_outside_hours(db_session):
db_session.add(group) db_session.add(group)
db_session.commit() db_session.commit()
card = Card(uuid="test-uuid-123", group_id=group.id) 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) db_session.add(card)
timetable = Timetable( timetable = Timetable(
@@ -52,10 +52,10 @@ def test_check_access_outside_hours(db_session):
group.accessauths = [aa] group.accessauths = [aa]
db_session.commit() db_session.commit()
result = checkAccess("test-uuid-123", db_session) result = checkAccess("test-key-123", db_session)
assert result == False assert result == False
def test_check_access_invalid_card(db_session): def test_check_access_invalid_card(db_session):
# Should raise exception for non-existent card # Should raise exception for non-existent card
with pytest.raises(Exception): with pytest.raises(Exception):
checkAccess("non-existent-uuid", db_session) checkAccess("non-existent-key", db_session)