diff --git a/app/controllers/cardManager.py b/app/controllers/cardManager.py index 74c86f3..0e9f527 100644 --- a/app/controllers/cardManager.py +++ b/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 \ No newline at end of file +@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) \ No newline at end of file diff --git a/app/controllers/debugManager.py b/app/controllers/debugManager.py index 282c9bd..e8502e4 100644 --- a/app/controllers/debugManager.py +++ b/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 \ No newline at end of file diff --git a/app/controllers/userManager.py b/app/controllers/userManager.py index 7314272..22bfbb4 100644 --- a/app/controllers/userManager.py +++ b/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) diff --git a/app/model/models.py b/app/model/models.py index b465640..7157b94 100644 --- a/app/model/models.py +++ b/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 diff --git a/app/services/door.py b/app/services/door.py index afdeaff..c4d1faf 100644 --- a/app/services/door.py +++ b/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: diff --git a/app/services/scanner.py b/app/services/scanner.py index b22da7c..56bbf49 100644 --- a/app/services/scanner.py +++ b/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) diff --git a/test/conftest.py b/test/conftest.py index cc1a953..70bf024 100644 --- a/test/conftest.py +++ b/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) diff --git a/test/test_models.py b/test/test_models.py index 6dff3aa..85e3ac9 100644 --- a/test/test_models.py +++ b/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 diff --git a/test/test_services/test_card_manager.py b/test/test_services/test_card_manager.py index dc68cbb..4e9e15c 100644 --- a/test/test_services/test_card_manager.py +++ b/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 diff --git a/test/test_services/test_database.py b/test/test_services/test_database.py index 8496032..c4a305d 100644 --- a/test/test_services/test_database.py +++ b/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): diff --git a/test/test_services/test_door.py b/test/test_services/test_door.py index b90e9ef..166f3e0 100644 --- a/test/test_services/test_door.py +++ b/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)