Merge pull request 'Card model changes' (#14) from change_card_model into master

Reviewed-on: #14

Fixes #5, #6, #13

app/controllers/cardManager.py

@@ -5,7 +5,7 @@ from sqlmodel import Session, select
 from typing import List
 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.auth import auth_is_admin
 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"])
 
-def register_card(group_id: int):
-    key = WriteNewCard()
+def register_card(cardInput: CardCreate):
+    key, uid = WriteNewCard()
     if key == None:
         logger.info("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
 
-@card_router.post("/{group_id}", response_model=Card)
-def add_card(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)):
-    card = register_card(group_id)
+@card_router.post("/", response_model=Card)
+def add_card(cardInput: CardCreate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
+    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)
 
-@card_router.get("/delete")
+@card_router.delete("/")
 def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
     key = DeleteCard()
     logger.info(key)
     try:
-        card = db.exec(select(Card).where(Card.uuid == key)).one()
+        card = db.exec(select(Card).where(Card.key == key)).one()
     except NoResultFound:
         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.commit()
     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()
     return cards
 
-
-#TODO:
-# -Split Authorisations + Cards
-# -Deactivation
-# -Deleting
+@card_router.patch("/{card_id}", response_model=Card)
+def update_card(card_id: int, cardInput: CardUpdate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)):
+    db_card = db.get(Card, card_id)
+    if db_card is None:
+        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)

app/controllers/debugManager.py

@@ -3,6 +3,7 @@ logger = logging.getLogger(__name__)
 from fastapi import APIRouter, Depends, HTTPException
 from app.services.auth import auth_is_admin
 from sqlmodel import Session, select
+import sqlalchemy.exc as exc
 from app.model.models import *
 from app.services.database import get_session, add_and_refresh
 
@@ -12,19 +13,31 @@ debug_router = APIRouter(
     dependencies=[Depends(auth_is_admin)],
 )
 
-@debug_router.get("/addcard/{groupid}/{card_key}")
-def add_card_manually(groupid: int, card_key: str, db: Session=Depends(get_session)):
+@debug_router.put("/addcard/")
+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)"""
     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)
     return card
 
-@debug_router.get("/rmcard/{card_key}")
-def remove_card_manually(card_key: str, db: Session = Depends(get_session)):
-    logger.critical(f"Manual db change: removing a card with key: {card_key}")
-    card = db.exec(select(Card).where(Card.uuid == card_key)).one()
-    add_and_refresh(db, card)
+@debug_router.get("/rmcard/{card_id}")
+def remove_card_manually(card_id: str, db: Session = Depends(get_session)):
+    try:
+        card = db.exec(select(Card).where(Card.id == card_id)).one()
+    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")
 def list_all_cards(db: Session = Depends(get_session)):
@@ -33,5 +46,5 @@ def list_all_cards(db: Session = Depends(get_session)):
     print(cards)
     out = []
     for i in cards:
-        out.append({i.uuid: i.group.name})
+        out.append({i.key: i.group.name})
     return out

app/controllers/userManager.py

@@ -1,6 +1,6 @@
 import logging
 logger = logging.getLogger(__name__)
-from fastapi import APIRouter, HTTPException, Depends
+from fastapi import APIRouter, HTTPException, Depends, status
 from sqlmodel import Session, select
 from typing import List
 
@@ -12,8 +12,10 @@ user_router = APIRouter(tags=["Users"], prefix="/api/v1/users")
 
 @user_router.post("/", response_model=UserResponse)
 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)}
+    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)
     return add_and_refresh(db, db_user)
 

app/model/models.py

@@ -7,7 +7,7 @@ class Base(SQLModel):
 
 #### User
 class UserBase(Base):
-    name: str = Field(index=True)
+    name: str = Field(index=True, unique=True)
     email: str | None = None
     is_admin: bool = False
 
@@ -84,10 +84,23 @@ class AccessAuthorizationUpdate(Base):
 #### Card
 class Card(Base, table=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: 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):
     weekday: int = Field(le=6, ge=0)
     starttime: time

app/services/door.py

@@ -28,11 +28,11 @@ def closeDoor():
 def isDoorOpen():
     return doorIsOpen
 
-def checkAccess(uuid: str, db: Session):
+def checkAccess(key: str, db: Session):
     try:
         current_weekday = datetime.datetime.weekday(datetime.date.today())
         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:
             logger.info(f"checking auth: {auth.name}")
             for timetable in auth.timetables:

app/services/scanner.py

@@ -233,7 +233,7 @@ def WriteNewCard():
         assert rdata == get_list(key)
         logger.debug("  - Data written successfully.")
         scannerThread.start()
-        return key
+        return key, to_hex_string(data=uid, separator=":") 
 
     except Exception as e:
         logger.error(f"Error in write function: {e}", exc_info=True)

test/conftest.py

@@ -99,7 +99,7 @@ def test_group(db_session):
 @pytest.fixture
 def test_card(db_session, test_group):
     """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.commit()
     db_session.refresh(card)

test/test_models.py

@@ -59,8 +59,8 @@ def test_access_authorization_models():
 
 def test_card_model():
     """Test card model creation and validation."""
-    card = Card(uuid="test-uuid", group_id=1)
-    assert card.uuid == "test-uuid"
+    card = Card(key="test-key", group_id=1)
+    assert card.key == "test-key"
     assert card.group_id == 1
 
 

test/test_services/test_card_manager.py

@@ -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):
     """Test that non-admin users cannot perform card operations."""
     # 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
 
     # Try to get cards

test/test_services/test_database.py

@@ -26,25 +26,16 @@ def test_create_db_and_tables():
 def test_get_session(db_session):
     """Test database session generator."""
     # Test that we can get a session
-    session_gen = get_session()
-    session = next(session_gen)
-
-    assert isinstance(session, Session)
+    assert isinstance(db_session, Session)
 
     # Test that session works
-    user = UserDB(name="Test", passwordhash="hash")
-    session.add(user)
-    session.commit()
+    user = UserDB(name="Test_User", passwordhash="hash")
+    db_session.add(user)
+    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.name == "Test"
-
-    # Clean up generator
-    try:
-        next(session_gen)
-    except StopIteration:
-        pass
+    assert retrieved_user.name == "Test_User"
 
 
 def test_add_and_refresh(db_session):

test/test_services/test_door.py

@@ -9,7 +9,7 @@ def test_check_access_with_valid_timetable(db_session):
     db_session.add(group)
     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)
 
     timetable = Timetable(
@@ -27,7 +27,7 @@ def test_check_access_with_valid_timetable(db_session):
     db_session.commit()
 
     # 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
 
 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.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)
 
     timetable = Timetable(
@@ -52,10 +52,10 @@ def test_check_access_outside_hours(db_session):
     group.accessauths = [aa]
 
     db_session.commit()
-    result = checkAccess("test-uuid-123", db_session)
+    result = checkAccess("test-key-123", db_session)
     assert result == False
 
 def test_check_access_invalid_card(db_session):
     # Should raise exception for non-existent card
     with pytest.raises(Exception):
-        checkAccess("non-existent-uuid", db_session)
+        checkAccess("non-existent-key", db_session)