2 Commits

Author SHA1 Message Date
e871330041 Update flake.lock 2026-06-27 14:59:00 +02:00
2d64edde64 WIP add rpi nixos-config 2026-06-27 14:50:50 +02:00
15 changed files with 306 additions and 200 deletions

View File

@@ -1,39 +1,17 @@
## Gatekeeper - Door access system
#### Status: WIP - getting there o.o
Status: WIP - not prod ready
Start prod server `nix run`<br>
Start dev server `nix run .#dev` or `nix run .#dev -- {args}`<br>
Interactive dev with `nix develop`, then sync deps with `uv sync`<br>
Swagger UI @ http://127.0.0.1:8000/api/v1/docs<br>
OpenApi @ http://127.0.0.1:8000/api/v1/openapi.json<br>
Dev with `nix develop`
sync python deps with `uv sync`
start dev server `uv run fastapi dev`
Swagger UI @ http://127.0.0.1:8000/docs
start prod server `uv run fastapi run`
#### Config:
You need to set services.pcscd.enable = true; for the smartcard reader to work
Nixos config for the card reader:
```
services.pcscd = {
enable = true;
plugins = [ pkgs.acsccid ]; #<-- Needed for the ACE1552U
};
```
This application is configured using env vars. You can use `cp .env.example .env` and then edit `.env` to change the keys to something else.
### A note on keycard scanners
I have tried two scanners while developing this project, the TWN4 MultiTech 2 from elatec and the ACR1552U from acs.<br>
##### TWN4
From the factory the TWN4 comes with firmware that emulates a HID and just inputs the serial of any scanned card. To be usable with pcscd it had to be flashed with the `TWN4_xPx520_S1SC161_Multi_CCID_1Slot_Standard.bix` firmware found in the `TWN4DevPack520` which can be downloaded under https://www.elatec-rfid.com/int/elatec-software (They send you a download link via email).
This software seems to not work under wine. I had to setup a win10 vm and pass the usb to be able to flash.<br>
Range: In my testing the TWN4 has a range of about 22mm when interfacing with our Mifare DESFire v2 cards. This is not enough for our idea of mounting the scanner on the inside of a glass window.
#### ACR1552U
The ACR1552U comes preloaded with PC/SC firmware. You need to install the acsccid package with your package-manager of choice (Nixos user see above) for the scanner to work but after that I had no problems with the hardware interfacing.<br>
Range: The range of the ACR1552U was much better at over 60mm (almost 70mm if you take of the face plate)
#### Issues:
Issues:
- documentation missing
- `nix run` currently broken
- raspberry pi image not working
- no door state
- no door operations

View File

@@ -5,7 +5,7 @@ from sqlmodel import Session, select
from typing import List
from sqlalchemy.exc import NoResultFound
from ..model.models import Card, CardCreate, CardUpdate, GroupDB
from ..model.models import Card
from ..services.database import engine, get_session, add_and_refresh
from ..services.auth import auth_is_admin
import uuid as gen_uuid
@@ -13,40 +13,28 @@ from app.services.scanner import WriteNewCard, DeleteCard
card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"])
def register_card(cardInput: CardCreate):
key, uid = WriteNewCard()
def register_card(group_id: int):
key = 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=cardInput.group_id,
key=key,
name=cardInput.name,
card_serial=uid,
enabled=cardInput.enabled
)
card = Card(group_id=group_id, uuid=key)
return card
@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)
@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)
return add_and_refresh(db, card)
@card_router.delete("/")
@card_router.get("/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.key == key)).one()
card = db.exec(select(Card).where(Card.uuid == key)).one()
except NoResultFound:
logger.info(f"The key:'{key}' was not found in db!")
raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Key on card not found in DB. Please tell an admin about this. KEY={key}")
raise HTTPException(status_code=500, 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"}
@@ -56,15 +44,8 @@ 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
@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)
#TODO:
# -Split Authorisations + Cards
# -Deactivation
# -Deleting

View File

@@ -3,7 +3,6 @@ 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
@@ -13,31 +12,19 @@ debug_router = APIRouter(
dependencies=[Depends(auth_is_admin)],
)
@debug_router.put("/addcard/")
def add_card_manually(groupid: int, card_key: str, name: str, enabled: bool, db: Session=Depends(get_session)):
@debug_router.get("/addcard/{groupid}/{card_key}")
def add_card_manually(groupid: int, card_key: str, 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,
key=card_key,
name=name,
enabled=enabled,
card_serial="00:00:00:00:00:00:00"
)
card = Card(group_id=groupid, uuid=card_key)
add_and_refresh(db, card)
return 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.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.put("/getcards")
def list_all_cards(db: Session = Depends(get_session)):
@@ -46,5 +33,5 @@ def list_all_cards(db: Session = Depends(get_session)):
print(cards)
out = []
for i in cards:
out.append({i.key: i.group.name})
out.append({i.uuid: i.group.name})
return out

View File

@@ -1,6 +1,6 @@
import logging
logger = logging.getLogger(__name__)
from fastapi import APIRouter, HTTPException, Depends, status
from fastapi import APIRouter, HTTPException, Depends
from sqlmodel import Session, select
from typing import List
@@ -12,10 +12,8 @@ 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)

View File

@@ -26,9 +26,6 @@ def checkDeps():
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.critical("-"*63)
logger.critical("---- Documentation is at http://127.0.0.1:8000/api/v1/docs ----")
logger.critical("-"*63)
checkDeps()
create_db_and_tables()
create_first_user(db=get_db_session())

View File

@@ -7,7 +7,7 @@ class Base(SQLModel):
#### User
class UserBase(Base):
name: str = Field(index=True, unique=True)
name: str = Field(index=True)
email: str | None = None
is_admin: bool = False
@@ -84,23 +84,10 @@ class AccessAuthorizationUpdate(Base):
#### Card
class Card(Base, table=True):
id: int | None = Field(default=None, primary_key=True)
key: str
card_serial: str
enabled: bool = True
name: str = Field(unique=True, max_length=32)
uuid: str
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

View File

@@ -28,11 +28,11 @@ def closeDoor():
def isDoorOpen():
return doorIsOpen
def checkAccess(key: str, db: Session):
def checkAccess(uuid: str, db: Session):
try:
current_weekday = datetime.datetime.weekday(datetime.date.today())
current_time = datetime.datetime.now()
card = db.exec(select(Card).where(Card.key == key)).one()
card = db.exec(select(Card).where(Card.uuid == uuid)).one()
for auth in card.group.accessauths:
logger.info(f"checking auth: {auth.name}")
for timetable in auth.timetables:

View File

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

128
flake.lock generated
View File

@@ -1,12 +1,114 @@
{
"nodes": {
"argononed": {
"flake": false,
"locked": {
"lastModified": 1729566243,
"narHash": "sha256-DPNI0Dpk5aym3Baf5UbEe5GENDrSmmXVdriRSWE+rgk=",
"owner": "nvmd",
"repo": "argononed",
"rev": "16dbee54d49b66d5654d228d1061246b440ef7cf",
"type": "github"
},
"original": {
"owner": "nvmd",
"repo": "argononed",
"type": "github"
}
},
"flake-compat": {
"locked": {
"lastModified": 1767039857,
"narHash": "sha256-vNpUSpF5Nuw8xvDLj2KCwwksIbjua2LZCqhV1LNRDns=",
"owner": "edolstra",
"repo": "flake-compat",
"rev": "5edf11c44bc78a0d334f6334cdaf7d60d732daab",
"type": "github"
},
"original": {
"owner": "edolstra",
"repo": "flake-compat",
"type": "github"
}
},
"nixos-hardware": {
"inputs": {
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1782562157,
"narHash": "sha256-a7+T6QSeowynwZ1ZJJbP8T8ntAytvrui8kFGJmIZt2c=",
"owner": "NixOS",
"repo": "nixos-hardware",
"rev": "a9cf7546a938c737b079e738de73934a13de9784",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "master",
"repo": "nixos-hardware",
"type": "github"
}
},
"nixos-images": {
"inputs": {
"nixos-stable": [
"nixos-raspberrypi",
"nixpkgs"
],
"nixos-unstable": [
"nixos-raspberrypi",
"nixpkgs"
]
},
"locked": {
"lastModified": 1747747741,
"narHash": "sha256-LUOH27unNWbGTvZFitHonraNx0JF/55h30r9WxqrznM=",
"owner": "nvmd",
"repo": "nixos-images",
"rev": "cbbd6db325775096680b65e2a32fb6187c09bbb4",
"type": "github"
},
"original": {
"owner": "nvmd",
"ref": "sdimage-installer",
"repo": "nixos-images",
"type": "github"
}
},
"nixos-raspberrypi": {
"inputs": {
"argononed": "argononed",
"flake-compat": "flake-compat",
"nixos-images": "nixos-images",
"nixpkgs": [
"nixpkgs"
]
},
"locked": {
"lastModified": 1779023229,
"narHash": "sha256-MInilg7B/06c34SwOuGSBho4l0H1EZcmvxTkSWCs5pE=",
"owner": "nvmd",
"repo": "nixos-raspberrypi",
"rev": "06c6e3513e1ee64b651913193fc6ac38aa4963f5",
"type": "github"
},
"original": {
"owner": "nvmd",
"ref": "main",
"repo": "nixos-raspberrypi",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1778869304,
"narHash": "sha256-30sZNZoA1cqF5JNO9fVX+wgiQYjB7HJqqJ4ztCDeBZE=",
"lastModified": 1782467914,
"narHash": "sha256-pGvFkM8N0xEkIIXDe5YYfbEAvHrk4IxBrjB/x8OomhE=",
"owner": "nixos",
"repo": "nixpkgs",
"rev": "d233902339c02a9c334e7e593de68855ad26c4cb",
"rev": "e73de5be04e0eff4190a1432b946d469c794e7b4",
"type": "github"
},
"original": {
@@ -29,11 +131,11 @@
]
},
"locked": {
"lastModified": 1776659114,
"narHash": "sha256-qapCOQmR++yZSY43dzrp3wCrkOTLpod+ONtJWBk6iKU=",
"lastModified": 1782093830,
"narHash": "sha256-6gmEVe69+KlRkZD4PEEV5xAlB9CB0Y9TiuEgQjDrKTQ=",
"owner": "pyproject-nix",
"repo": "build-system-pkgs",
"rev": "ffaa2161dd5d63e0e94591f86b54fc239660fb2e",
"rev": "430680a19bc85a3bda55f12e4cc1a1aadcf2e478",
"type": "github"
},
"original": {
@@ -49,11 +151,11 @@
]
},
"locked": {
"lastModified": 1778901413,
"narHash": "sha256-GSKXTAnFqRAMlZkJrIPcQMYf+lpMr66K3i60mB9STvc=",
"lastModified": 1782089418,
"narHash": "sha256-LRD1SuQWr49fGq3A+8GLXfsLE2xqIpQA440YDZwms3M=",
"owner": "pyproject-nix",
"repo": "pyproject.nix",
"rev": "a228447c3e179d477c1b6246ef3efa8cfe3c469a",
"rev": "43f0b40edd0a74c63f66b7b48d969ae6b740d611",
"type": "github"
},
"original": {
@@ -64,6 +166,8 @@
},
"root": {
"inputs": {
"nixos-hardware": "nixos-hardware",
"nixos-raspberrypi": "nixos-raspberrypi",
"nixpkgs": "nixpkgs",
"pyproject-build-systems": "pyproject-build-systems",
"pyproject-nix": "pyproject-nix",
@@ -80,11 +184,11 @@
]
},
"locked": {
"lastModified": 1779269674,
"narHash": "sha256-P1LHCRdYpdtHAEzuEsNHrI6d9mVPl5a2fyFDZGHNVbI=",
"lastModified": 1782100052,
"narHash": "sha256-UfyLY3Hfwb3JqxCcj0953GxTFI5dEL85EKEe6DAFiVs=",
"owner": "pyproject-nix",
"repo": "uv2nix",
"rev": "69aec536f6d1acc415ed2e20299312802aba98c6",
"rev": "920fc6dfaf9f10ec56de93b184055e1a9f380d5e",
"type": "github"
},
"original": {

143
flake.nix
View File

@@ -21,15 +21,24 @@
inputs.uv2nix.follows = "uv2nix";
inputs.nixpkgs.follows = "nixpkgs";
};
nixos-hardware = {
url = "github:NixOS/nixos-hardware/master";
inputs.nixpkgs.follows = "nixpkgs";
};
nixos-raspberrypi = {
url = "github:nvmd/nixos-raspberrypi/main";
inputs.nixpkgs.follows = "nixpkgs";
};
};
outputs =
{
self,
nixpkgs,
pyproject-nix,
uv2nix,
pyproject-build-systems,
nixos-hardware,
...
}:
let
@@ -107,21 +116,121 @@
default = pythonSets.${system}.mkVirtualEnv "gatekeeper" workspace.deps.default;
});
apps = forAllSystems (system: {
default = {
type = "app";
program = toString (nixpkgs.legacyPackages.${system}.writeShellScript "gatekeeper" ''
export LD_LIBRARY_PATH="${lib.getLib nixpkgs.legacyPackages.${system}.pcsclite}/lib''${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
exec ${self.packages.${system}.default}/bin/fastapi run ${self}/app/main.py "$@";
'');
};
dev = {
type = "app";
program = toString (nixpkgs.legacyPackages.${system}.writeShellScript "gatekeeper" ''
export LD_LIBRARY_PATH="${lib.getLib nixpkgs.legacyPackages.${system}.pcsclite}/lib''${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
exec ${self.packages.${system}.default}/bin/fastapi run ${self}/app/main.py "$@";
'');
};
});
nixosConfigurations = {
gatekeeper-rpi = nixpkgs.lib.nixosSystem {
system = "aarch64-linux";
specialArgs = {
inherit workspace overlay;
};
modules = [
"${nixpkgs}/nixos/modules/profiles/headless.nix"
"${nixpkgs}/nixos/modules/profiles/minimal.nix"
"${nixpkgs}/nixos/modules/installer/sd-card/sd-image-aarch64.nix"
nixos-hardware.nixosModules.raspberry-pi-3
({ pkgs, ... }: {
nix.nixPath = [
"nixpkgs=${pkgs.path}"
];
nix.settings = {
trusted-users = [ "root" "@wheel" ];
experimental-features = [ "nix-command" "flakes" ];
auto-optimise-store = true;
};
})
({ config, pkgs, packages, ...}:
let
gatekeeper-app = pkgs.python3Packages.buildPythonPackage {
pname = "gatekeeper-app";
version = "0.0.1";
propagatedBuildInputs = [packages.gatekeeper];
src = ./app;
format = "pyproject";
};
in
{
boot.loader.grub.enable = false;
boot.loader.generic-extlinux-compatible.enable = true;
boot.kernelPackages = pkgs.linuxPackages;
boot.tmp.cleanOnBoot = true;
hardware.enableRedistributableFirmware = true;
networking.hostName = "gatekeeper-rpi";
#minimal
documentation.man.enable = false;
xdg.icons.enable = false;
xdg.mime.enable = false;
xdg.sounds.enable = false;
users.users.root.openssh.authorizedKeys.keys = ["ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAICKaEcGaSKU0xC5qCwzj2oCLLG4PYjWHZ7/CXHw4urVk atlan@nixos"];
users.groups = { gatekeeper = {}; };
users.users.admin = {
isNormalUser = true;
extraGroups = [ "wheel" "networkmanager" "gatekeeper" ];
};
users.users.gatekeeper = {
isNormalUser = true;
extraGroups = ["gatekeeper"];
};
services.openssh = {
enable = true;
ports = [ 22 ];
settings = {
PasswordAuthentication = false;
};
};
environment.systemPackages = with pkgs; [
git
htop
pcsclite
gatekeeper-app
];
environment.etc."gatekeeper-env".source =
pythonSets.aarch64-linux.mkVirtualEnv "gatekeeper" workspace.deps.default;
services.pcscd.enable = true;
systemd.services.gatekeeper = {
enable = true;
description = "Gatekeeper service";
after = [ "network.target" "pcscd.service"];
wantedBy = [ "multi-user.target" ];
environment = {
LD_LIBRARY_PATH = "${lib.getLib pkgs.pcsclite}/lib";
PYTHONPATH = "/etc/gatekeeper-env/lib/python3.x/site-packages";
};
serviceConfig = {
Type = "simple";
User = "gatekeeper";
Group = "gatekeeper";
WorkingDirectory = "/var/lib/gatekeeper";
ExecStart = "${pythonSets.aarch64-linux.mkVirtualEnv "gatekeeper" workspace.deps.default}/bin/python -m uvicorn main:app --host 0.0.0.0 --port 8000";
Restart = "always";
RestartSec = "10";
};
# Create working directory and log directory
preStart = ''
mkdir -p /var/lib/gatekeeper
mkdir -p /var/log/gatekeeper
chown -R gatekeeper:gatekeeper /var/lib/gatekeeper
chown -R gatekeeper:gatekeeper /var/log/gatekeeper
'';
};
networking.firewall.allowedTCPPorts = [ 8000 ];
time.timeZone = "Europe/Berlin";
system.stateVersion = "25.11";
})
];
};
};
};
}

View File

@@ -99,7 +99,7 @@ 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(uuid="test-uuid-123", group_id=test_group.id)
db_session.add(card)
db_session.commit()
db_session.refresh(card)

View File

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

View File

@@ -23,53 +23,9 @@ 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(f"/api/v1/cards/{test_group.id}", 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"
}
response = client.patch(
f"/api/v1/cards/{test_card.id}",
json=update_data,
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "changed_name"
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
)
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"
}
response = client.patch(
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"]

View File

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

View File

@@ -9,7 +9,7 @@ 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(uuid="test-uuid-123", group_id=group.id)
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-key-123", db_session)
result = checkAccess("test-uuid-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(key="test-key-123", group_id=group.id, enabled=True, name="test_card", card_serial="00:00:00:00:00:00:00")
card = Card(uuid="test-uuid-123", group_id=group.id)
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-key-123", db_session)
result = checkAccess("test-uuid-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-key", db_session)
checkAccess("non-existent-uuid", db_session)