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

[lint] lint + format tests

Browse Source
This commit is contained in:
ahtlon
2026-07-29 02:40:46 +02:00
parent 506076cc1e
commit 8aa17c1b84
11 changed files with 208 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
test.py
View File

@@ -15,21 +15,32 @@ It performs the following steps:
import logging
import os
from desfire import (
DESFire,
DESFireKey,
PCSCDevice,
diversify_key,
get_list,
to_hex_string,
)
from desfire.enums import (
DESFireCommunicationMode,
DESFireFileType,
DESFireKeySettings,
DESFireKeyType,
)
from desfire.schemas import FilePermissions, FileSettings, KeySettings
from dotenv import load_dotenv
from smartcard.CardRequest import CardRequest
from smartcard.CardType import AnyCardType
from smartcard.Exceptions import CardRequestTimeoutException
from desfire import DESFire, DESFireKey, PCSCDevice, diversify_key, get_list, to_hex_string
from desfire.enums import DESFireCommunicationMode, DESFireFileType, DESFireKeySettings, DESFireKeyType
from desfire.schemas import FilePermissions, FileSettings, KeySettings
from dotenv import load_dotenv
# Please make sure to yet your own keys here before running this script
load_dotenv()
MIFARE_APP_MASTER_KEY = os.getenv('MIFARE_APP_MASTER_KEY')
MIFARE_ACL_READ_BASE_KEY = os.getenv('MIFARE_ACL_READ_BASE_KEY')
MIFARE_ACL_WRITE_BASE_KEY = os.getenv('MIFARE_ACL_WRITE_BASE_KEY')
MIFARE_APP_MASTER_KEY = os.getenv("MIFARE_APP_MASTER_KEY")
MIFARE_ACL_READ_BASE_KEY = os.getenv("MIFARE_ACL_READ_BASE_KEY")
MIFARE_ACL_WRITE_BASE_KEY = os.getenv("MIFARE_ACL_WRITE_BASE_KEY")
# Constants
MIFARE_APP_ID = "DEAFFE" # 7 bytes
@@ -118,8 +129,12 @@ desfire.authenticate(0x0, aes_app_mk)
# Change file read and write keys (diversified)
diversification_data = [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID)
read_div_key_bytes = diversify_key(get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False)
write_div_key_bytes = diversify_key(get_list(MIFARE_ACL_WRITE_BASE_KEY), diversification_data, pad_to_32=False)
read_div_key_bytes = diversify_key(
get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False
)
write_div_key_bytes = diversify_key(
get_list(MIFARE_ACL_WRITE_BASE_KEY), diversification_data, pad_to_32=False
)
print("Changing file read key...")
aes_file_read_key = DESFireKey(aes_keysettings, read_div_key_bytes)
@@ -154,11 +169,13 @@ print(" - File created successfully.")
print("Writing UID to encrypted file...")
data = [0x0] + uid
assert len(data) == 8
desfire.write_file_data(MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(data))
desfire.write_file_data(
MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(data)
)
print("Reading from encrypted file...")
rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data)
assert rdata == data
print(" - Data written successfully.")
print("Personalization finished.")
print("Personalization finished.")

61
test/conftest.py
View File

@@ -1,18 +1,29 @@
import pytest
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
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine
from app.main import app
from app.model.models import UserDB, Card, GroupDB, AccessAuthorizationDB, Timetable, AaGroupLink
from app.model.models import (
AccessAuthorizationDB,
Card,
GroupDB,
Timetable,
UserDB,
)
from app.services.database import get_session
# Use in-memory SQLite for testing
TEST_SQLALCHEMY_DATABASE_URL = "sqlite://"
engine = create_engine(TEST_SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool)
engine = create_engine(
TEST_SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
@pytest.fixture(scope="function")
def db_session():
@@ -26,6 +37,7 @@ def db_session():
@pytest.fixture(scope="function")
def client(db_session):
"""Create a test client with a database session override."""
def override_get_session():
yield db_session
@@ -39,10 +51,9 @@ def client(db_session):
def admin_user(db_session):
"""Create an admin user for testing."""
from app.services.auth import get_password_hash
admin = UserDB(
name="admin",
passwordhash=get_password_hash("admin123"),
is_admin=True
name="admin", passwordhash=get_password_hash("admin123"), is_admin=True
)
db_session.add(admin)
db_session.commit()
@@ -54,10 +65,9 @@ def admin_user(db_session):
def regular_user(db_session):
"""Create a regular user for testing."""
from app.services.auth import get_password_hash
user = UserDB(
name="user",
passwordhash=get_password_hash("user123"),
is_admin=False
name="user", passwordhash=get_password_hash("user123"), is_admin=False
)
db_session.add(user)
db_session.commit()
@@ -69,8 +79,7 @@ def regular_user(db_session):
def auth_headers(client, admin_user):
"""Get authentication headers for admin user."""
response = client.post(
"/api/v1/token",
data={"username": admin_user.name, "password": "admin123"}
"/api/v1/token", data={"username": admin_user.name, "password": "admin123"}
)
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@@ -80,8 +89,7 @@ def auth_headers(client, admin_user):
def user_auth_headers(client, regular_user):
"""Get authentication headers for regular user."""
response = client.post(
"/api/v1/token",
data={"username": regular_user.name, "password": "user123"}
"/api/v1/token", data={"username": regular_user.name, "password": "user123"}
)
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@@ -100,7 +108,13 @@ def test_group(db_session):
@pytest.fixture
def test_card(db_session, test_group):
"""Create a test card."""
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")
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)
@@ -110,16 +124,9 @@ def test_card(db_session, test_group):
@pytest.fixture
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
)
tt = Timetable(weekday=1, starttime=time(1, 0, 0, 0), duration=50)
aa = AccessAuthorizationDB(
name="Test AA",
is_active=True,
type="timetable",
timetables=[tt]
name="Test AA", is_active=True, type="timetable", timetables=[tt]
)
db_session.add(aa)
db_session.commit()

2
test/test_main.py
View File

@@ -4,9 +4,11 @@ def test_app_startup(client):
# Application should respond (even if it's a 404)
assert response.status_code in [404, 200]
def test_router_includes():
"""Test that all routers are included in the app."""
from app.main import app
routes = [route.path for route in app.routes]
# Check that router prefixes are present

28
test/test_models.py
View File

@@ -1,11 +1,18 @@
import pytest
import datetime
from app.model.models import (
UserBase, UserResponse, UserCreate, UserDB, UserUpdate,
GroupBase, GroupCreate, GroupDB, GroupResponse,
AccessAuthorizationBase, AccessAuthorizationCreate,
AccessAuthorizationDB, AccessAuthorizationResponse, AccessAuthorizationUpdate,
Card, Timetable, TimetableCreate, Token, TokenData, AaGroupLink
AaGroupLink,
AccessAuthorizationBase,
AccessAuthorizationCreate,
Card,
GroupBase,
GroupCreate,
TimetableCreate,
Token,
TokenData,
UserBase,
UserCreate,
UserUpdate,
)
@@ -18,7 +25,9 @@ def test_user_models():
assert user_base.is_admin is False
# Test UserCreate
user_create = UserCreate(name="New User", email="new@example.com", password="secret123")
user_create = UserCreate(
name="New User", email="new@example.com", password="secret123"
)
assert user_create.password == "secret123"
# Test UserUpdate
@@ -48,10 +57,7 @@ def test_access_authorization_models():
# Test AccessAuthorizationCreate with timetables
timetable_create = TimetableCreate(weekday=1, starttime="08:00", duration=60)
aa_create = AccessAuthorizationCreate(
name="New AA",
is_active=False,
type="timetable",
timetables=[timetable_create]
name="New AA", is_active=False, type="timetable", timetables=[timetable_create]
)
assert aa_create.name == "New AA"
assert aa_create.is_active is False

75
test/test_services/test_aa_manager.py
View File

@@ -1,7 +1,3 @@
import pytest
from fastapi import status
def test_create_access_auth_tt(client, auth_headers):
"""Test creating a new access authorization."""
aa_data = {
@@ -10,8 +6,8 @@ def test_create_access_auth_tt(client, auth_headers):
"is_active": True,
"timetables": [
{"weekday": 1, "starttime": "08:00", "duration": 60},
{"weekday": 2, "starttime": "09:00", "duration": 90}
]
{"weekday": 2, "starttime": "09:00", "duration": 90},
],
}
response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
@@ -24,16 +20,14 @@ def test_create_access_auth_tt(client, auth_headers):
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"
}
"oneshot": {"uses": 1, "ends_at": "2029-07-27"},
}
response = client.post("/api/v1/aa/", json=aa_data, headers=auth_headers)
@@ -46,6 +40,7 @@ def test_create_access_auth_os(client, auth_headers):
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 = {
@@ -57,6 +52,7 @@ def test_create_wrong_aa_type(client, auth_headers):
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)
@@ -88,8 +84,7 @@ def test_get_nonexistent_access_auth(client, auth_headers):
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_tt.id}",
headers=auth_headers
f"/api/v1/aa/assign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
)
assert response.status_code == 200
@@ -99,15 +94,18 @@ def test_assign_access_auth_to_group(client, auth_headers, test_group, test_aa_t
# Note: The response model might not include the full relationship
def test_assign_already_assigned_access_auth(client, auth_headers, test_group, test_aa_tt):
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_tt.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_tt.id}",
headers=auth_headers
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
assert response.status_code == 409
@@ -117,12 +115,13 @@ def test_assign_already_assigned_access_auth(client, auth_headers, test_group, t
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_tt.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_tt.id}",
headers=auth_headers
f"/api/v1/aa/unassign/{test_group.id}/{test_aa_tt.id}", headers=auth_headers
)
assert response.status_code == 200
@@ -130,35 +129,33 @@ def test_unassign_access_auth_from_group(client, auth_headers, test_group, test_
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_tt.id}",
headers=auth_headers
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_tt):
"""Test assigning an AA to a non-existent group."""
response = client.put(f"/api/v1/aa/assign/99999/{test_aa_tt.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
def test_assign_nonexistent_aa(client, auth_headers, test_group):
"""Test assigning a non-existent AA to a group."""
response = client.put(f"/api/v1/aa/assign/{test_group.id}/99999", headers=auth_headers)
response = client.put(
f"/api/v1/aa/assign/{test_group.id}/99999", headers=auth_headers
)
assert response.status_code == 404
def test_update_access_auth(client, auth_headers, test_aa_tt):
"""Test updating an access authorization."""
update_data = {
"name": "Updated AA",
"is_active": False
}
update_data = {"name": "Updated AA", "is_active": False}
response = client.patch(
f"/api/v1/aa/{test_aa_tt.id}",
json=update_data,
headers=auth_headers
f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
@@ -170,15 +167,11 @@ def test_update_access_auth(client, auth_headers, test_aa_tt):
def test_update_access_auth_with_timetables(client, auth_headers, test_aa_tt):
"""Test updating an access authorization with new timetables."""
update_data = {
"timetables": [
{"weekday": 5, "starttime": "10:00", "duration": 120}
]
"timetables": [{"weekday": 5, "starttime": "10:00", "duration": 120}]
}
response = client.patch(
f"/api/v1/aa/{test_aa_tt.id}",
json=update_data,
headers=auth_headers
f"/api/v1/aa/{test_aa_tt.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
jresponse = response.json()
@@ -216,9 +209,9 @@ 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(
"/api/v1/aa/",
json={"name": "test", "is_active": True, "timetables": []},
headers=user_auth_headers
"/api/v1/aa/",
json={"name": "test", "is_active": True, "timetables": []},
headers=user_auth_headers,
)
assert response.status_code == 403
@@ -227,5 +220,7 @@ def test_aa_tt_operations_by_non_admin(client, test_aa_tt, user_auth_headers):
assert response.status_code == 403
# Try to assign AA
response = client.put(f"/api/v1/aa/assign/1/{test_aa_tt.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

37
test/test_services/test_auth.py
View File

@@ -1,12 +1,19 @@
from datetime import timedelta
import pytest
from datetime import datetime, timedelta, timezone
from fastapi import HTTPException, status
from app.services.auth import (
verify_password, get_password_hash, get_user, authenticate_user,
create_access_token, get_current_user, auth_is_admin, create_first_user
)
from app.model.models import UserDB
from jwt.exceptions import InvalidTokenError
from app.services.auth import (
auth_is_admin,
authenticate_user,
create_access_token,
create_first_user,
get_current_user,
get_password_hash,
get_user,
verify_password,
)
def test_password_hashing():
@@ -84,7 +91,7 @@ def test_create_access_token():
def test_get_current_user(db_session, admin_user):
"""Test getting current user from token."""
from app.services.auth import create_access_token, get_current_user
from app.services.auth import create_access_token
# Create token for admin user
token = create_access_token(data={"sub": admin_user.name})
@@ -102,7 +109,9 @@ def test_get_current_user(db_session, admin_user):
# Test expired token (create token with past expiration)
past_expire = timedelta(minutes=-100)
expired_token = create_access_token(data={"sub": admin_user.name}, expires_delta=past_expire)
expired_token = create_access_token(
data={"sub": admin_user.name}, expires_delta=past_expire
)
with pytest.raises(HTTPException) as exc_info:
get_current_user(token=expired_token)
@@ -111,7 +120,7 @@ def test_get_current_user(db_session, admin_user):
def test_auth_is_admin(db_session, admin_user, regular_user):
"""Test admin authorization check."""
from app.services.auth import create_access_token, auth_is_admin
from app.services.auth import create_access_token
# Create token for admin user
admin_token = create_access_token(data={"sub": admin_user.name})
@@ -133,6 +142,7 @@ def test_create_first_user(db_session):
"""Test automatic creation of first admin user."""
# Clear any existing users
from sqlmodel import select
db_session.exec(select(UserDB)).all()
for user in db_session.exec(select(UserDB)).all():
db_session.delete(user)
@@ -158,8 +168,7 @@ def test_token_endpoint(client, admin_user):
"""Test the token endpoint for login."""
# Test successful login
response = client.post(
"/api/v1/token",
data={"username": admin_user.name, "password": "admin123"}
"/api/v1/token", data={"username": admin_user.name, "password": "admin123"}
)
assert response.status_code == 200
data = response.json()
@@ -168,15 +177,13 @@ def test_token_endpoint(client, admin_user):
# Test failed login with wrong password
response = client.post(
"/api/v1/token",
data={"username": admin_user.name, "password": "wrongpassword"}
"/api/v1/token", data={"username": admin_user.name, "password": "wrongpassword"}
)
assert response.status_code == 401
# Test failed login with non-existent user
response = client.post(
"/api/v1/token",
data={"username": "nonexistent", "password": "password"}
"/api/v1/token", data={"username": "nonexistent", "password": "password"}
)
assert response.status_code == 401

33
test/test_services/test_card_manager.py
View File

@@ -1,6 +1,3 @@
import pytest
from fastapi import status
def test_get_cards_for_group(client, auth_headers, test_group, test_card):
"""Test getting all cards for a group."""
response = client.get(f"/api/v1/cards/{test_group.id}", headers=auth_headers)
@@ -23,24 +20,20 @@ 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/", headers=user_auth_headers)
response = client.post("/api/v1/cards/", headers=user_auth_headers)
assert response.status_code == 403
# Try to get cards
response = client.get(f"/api/v1/cards/{test_group.id}", headers=user_auth_headers)
assert response.status_code == 403
def test_update_card(client, auth_headers, test_group, test_card):
"""Test Patching a card entity"""
update_data = {
"name": "changed_name",
"enabled": "False"
}
update_data = {"name": "changed_name", "enabled": "False"}
response = client.patch(
f"/api/v1/cards/{test_card.id}",
json=update_data,
headers=auth_headers
f"/api/v1/cards/{test_card.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
@@ -49,27 +42,21 @@ def test_update_card(client, auth_headers, test_group, test_card):
assert data["enabled"] == False
assert data["group_id"] == test_card.group_id
def test_update_wrong_card(client, auth_headers, test_group, test_card):
"""Test Patching a card entity"""
response = client.patch(
f"/api/v1/cards/9999",
json={},
headers=auth_headers
)
response = client.patch("/api/v1/cards/9999", json={}, headers=auth_headers)
assert response.status_code == 404
assert "Card not found" in response.json()["detail"]
def test_update_card_with_wrong_group(client, auth_headers, test_group, test_card):
"""Test Patching a card entity with wrong group"""
update_data = {
"group_id": "9999"
}
update_data = {"group_id": "9999"}
response = client.patch(
f"/api/v1/cards/{test_card.id}",
json=update_data,
headers=auth_headers
f"/api/v1/cards/{test_card.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 404
assert "GroupID not found" in response.json()["detail"]
assert "GroupID not found" in response.json()["detail"]

9
test/test_services/test_database.py
View File

@@ -1,13 +1,14 @@
import pytest
from sqlmodel import Session, select
from app.services.database import create_db_and_tables, get_session, add_and_refresh
from app.model.models import UserDB, GroupDB, Card
from sqlmodel import Session
from app.model.models import UserDB
from app.services.database import add_and_refresh, create_db_and_tables
def test_create_db_and_tables():
"""Test database and tables creation."""
# This is primarily an integration test
from sqlalchemy import inspect
from app.services.database import engine
create_db_and_tables()

47
test/test_services/test_door.py
View File

@@ -1,7 +1,16 @@
import pytest
import datetime
import pytest
from app.model.models import (
AccessAuthorizationDB,
Card,
GroupDB,
OneShotAccess,
Timetable,
)
from app.services.door import checkAccess
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
@@ -9,13 +18,19 @@ def test_check_access_with_valid_timetable(db_session):
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")
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(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.datetime.now().time(),
duration=120 # 2 hours
duration=120, # 2 hours
)
db_session.add(timetable)
@@ -30,19 +45,26 @@ def test_check_access_with_valid_timetable(db_session):
result = checkAccess("test-key-123", db_session)
assert result == True
def test_check_access_outside_hours(db_session):
# Test when current time is outside valid hours
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")
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(
weekday=datetime.datetime.weekday(datetime.date.today()),
starttime=datetime.time(1, 0),
duration=1 # 2 hours
duration=1, # 2 hours
)
db_session.add(timetable)
@@ -55,18 +77,24 @@ 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")
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)
uses=1, ends_at=datetime.datetime.now() + datetime.timedelta(days=1)
)
db_session.add(oneshot)
@@ -82,6 +110,7 @@ def test_check_access_with_valid_oneshot(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):

8
test/test_services/test_group_manager.py
View File

@@ -1,7 +1,3 @@
import pytest
from fastapi import status
def test_create_group(client, auth_headers):
"""Test creating a new group."""
group_data = {"name": "New Test Group"}
@@ -57,9 +53,7 @@ def test_group_operations_by_non_admin(client, user_auth_headers):
"""Test that non-admin users cannot perform group operations."""
# Try to create a group
response = client.post(
"/api/v1/groups/",
json={"name": "test"},
headers=user_auth_headers
"/api/v1/groups/", json={"name": "test"}, headers=user_auth_headers
)
assert response.status_code == 403

39
test/test_services/test_user_manager.py
View File

@@ -1,14 +1,10 @@
import pytest
from fastapi import status
def test_create_user(client, auth_headers):
"""Test creating a new user."""
user_data = {
"name": "newuser",
"email": "newuser@example.com",
"is_admin": False,
"password": "newpassword123"
"password": "newpassword123",
}
response = client.post("/api/v1/users/", json=user_data, headers=auth_headers)
@@ -27,7 +23,7 @@ def test_create_user_unauthorized(client):
user_data = {
"name": "unauthorized_user",
"email": "unauthorized@example.com",
"password": "password123"
"password": "password123",
}
response = client.post("/api/v1/users/", json=user_data)
@@ -56,6 +52,7 @@ def test_get_user_by_id(client, auth_headers, regular_user):
assert data["id"] == regular_user.id
assert data["name"] == regular_user.name
def test_get_current_user(client, auth_headers, admin_user):
"""Test getting the special url current"""
response = client.get("/api/v1/users/current", headers=auth_headers)
@@ -66,6 +63,7 @@ def test_get_current_user(client, auth_headers, admin_user):
assert data["name"] == admin_user.name
assert data["is_admin"] == admin_user.is_admin
def test_get_nonexistent_user(client, auth_headers):
"""Test retrieving a non-existent user."""
response = client.get("/api/v1/users/99999", headers=auth_headers)
@@ -75,15 +73,10 @@ def test_get_nonexistent_user(client, auth_headers):
def test_update_user(client, auth_headers, regular_user):
"""Test updating a user."""
update_data = {
"name": "updated_name",
"email": "updated@example.com"
}
update_data = {"name": "updated_name", "email": "updated@example.com"}
response = client.patch(
f"/api/v1/users/{regular_user.id}",
json=update_data,
headers=auth_headers
f"/api/v1/users/{regular_user.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
@@ -96,21 +89,17 @@ def test_update_user(client, auth_headers, regular_user):
def test_update_user_password(client, auth_headers, regular_user):
"""Test updating a user's password."""
update_data = {
"password": "new_password_456"
}
update_data = {"password": "new_password_456"}
response = client.patch(
f"/api/v1/users/{regular_user.id}",
json=update_data,
headers=auth_headers
f"/api/v1/users/{regular_user.id}", json=update_data, headers=auth_headers
)
assert response.status_code == 200
# Verify password can be used for login
login_response = client.post(
"/api/v1/token",
data={"username": regular_user.name, "password": "new_password_456"}
data={"username": regular_user.name, "password": "new_password_456"},
)
assert login_response.status_code == 200
@@ -118,7 +107,9 @@ def test_update_user_password(client, auth_headers, regular_user):
def test_update_nonexistent_user(client, auth_headers):
"""Test updating a non-existent user."""
update_data = {"name": "updated"}
response = client.patch("/api/v1/users/99999", json=update_data, headers=auth_headers)
response = client.patch(
"/api/v1/users/99999", json=update_data, headers=auth_headers
)
assert response.status_code == 404
@@ -143,9 +134,9 @@ def test_user_operations_by_non_admin(client, user_auth_headers):
"""Test that non-admin users cannot perform admin operations."""
# Try to create a user
response = client.post(
"/api/v1/users/",
json={"name": "test", "password": "pass"},
headers=user_auth_headers
"/api/v1/users/",
json={"name": "test", "password": "pass"},
headers=user_auth_headers,
)
assert response.status_code == 403