From 8daf064e4da6b62d4cd140a649708ff4cd4e4ef4 Mon Sep 17 00:00:00 2001 From: Niklas Meinzer Date: Sat, 14 Feb 2026 10:54:40 +0100 Subject: [PATCH] accounts: Add new accounts and add users and balance to accounts Close #5 --- src/allmende_payment_system/api/admin.py | 74 ++++++++++++++++- .../templates/accounts.html.jinja | 83 ++++++++++++++++++- test/test_admin.py | 63 +++++++++++++- 3 files changed, 215 insertions(+), 5 deletions(-) diff --git a/src/allmende_payment_system/api/admin.py b/src/allmende_payment_system/api/admin.py index 10b812f..2ad1020 100644 --- a/src/allmende_payment_system/api/admin.py +++ b/src/allmende_payment_system/api/admin.py @@ -2,6 +2,7 @@ from typing import Annotated from fastapi import APIRouter, File, Form, HTTPException, Request from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from starlette import status from starlette.responses import RedirectResponse @@ -12,6 +13,7 @@ from allmende_payment_system.models import ( Area, Permission, Product, + Transaction, User, UserGroup, ) @@ -296,8 +298,78 @@ async def get_accounts( templates = get_jinja_renderer() accounts = session.scalars(select(Account)).all() + users = session.scalars(select(User)).all() return templates.TemplateResponse( "accounts.html.jinja", - context={"request": request, "accounts": accounts}, + context={"request": request, "accounts": accounts, "users": users}, + ) + + +@admin_router.post("/accounts/new") +async def create_account( + request: Request, + session: SessionDep, + user: UserDep, + account_name: Annotated[str, Form()], +): + if not user.has_permission("account", "edit"): + raise HTTPException(status_code=403, detail="Insufficient permissions") + + account = Account(name=account_name) + session.add(account) + try: + session.flush() + except IntegrityError as e: + session.rollback() + raise HTTPException( + status_code=400, detail="Account with this name already exists" + ) from e + return RedirectResponse( + url="/admin/accounts", status_code=status.HTTP_303_SEE_OTHER + ) + + +@admin_router.post("/accounts/{account_id}/add_user") +async def add_user_to_account( + request: Request, + session: SessionDep, + user: UserDep, + account_id: int, + user_id: Annotated[int, Form()], +): + if not user.has_permission("account", "edit"): + raise HTTPException(status_code=403, detail="Insufficient permissions") + + account = session.execute( + select(Account).where(Account.id == account_id) + ).scalar_one() + user_to_add = session.execute(select(User).where(User.id == user_id)).scalar_one() + + account.users.append(user_to_add) + + return RedirectResponse( + url="/admin/accounts", status_code=status.HTTP_303_SEE_OTHER + ) + + +@admin_router.post("/accounts/{account_id}/add_balance") +async def add_balance_to_account( + request: Request, + session: SessionDep, + user: UserDep, + account_id: int, + amount: Annotated[float, Form()], +): + if not user.has_permission("account", "edit"): + raise HTTPException(status_code=403, detail="Insufficient permissions") + + account = session.execute( + select(Account).where(Account.id == account_id) + ).scalar_one() + + account.transactions.append(Transaction(type="deposit", total_amount=amount)) + + return RedirectResponse( + url="/admin/accounts", status_code=status.HTTP_303_SEE_OTHER ) diff --git a/src/allmende_payment_system/templates/accounts.html.jinja b/src/allmende_payment_system/templates/accounts.html.jinja index e5977b7..5cdbc30 100644 --- a/src/allmende_payment_system/templates/accounts.html.jinja +++ b/src/allmende_payment_system/templates/accounts.html.jinja @@ -2,7 +2,30 @@ {% block content %}

Konten verwalten

- Neues Konto erstellen + +
+ + {% if accounts|length == 0 %} @@ -24,7 +47,63 @@ {{ account.name }} {{ account.users | map(attribute='display_name') | join(", ") }} {{ account.balance | format_number }} € - actions + + + + + {# ADD USER #} + + + {# ADD BALANCE #} + + {% endfor %} diff --git a/test/test_admin.py b/test/test_admin.py index e886c29..744bbc3 100644 --- a/test/test_admin.py +++ b/test/test_admin.py @@ -5,7 +5,7 @@ from sqlalchemy.orm import Session from allmende_payment_system.app import app from allmende_payment_system.database import ensure_user -from allmende_payment_system.models import Permission, User, UserGroup +from allmende_payment_system.models import Account, Permission, User, UserGroup @pytest.fixture @@ -15,6 +15,7 @@ def admin_user(test_db): group = UserGroup(id=1, name="Admins") group.permissions.append(Permission(scope="user", action="edit")) + group.permissions.append(Permission(scope="account", action="edit")) user.user_groups.append(group) test_db.add(group) test_db.flush() @@ -97,6 +98,7 @@ def test_group_add_permission_illegal_format(test_db, client, admin_user): def test_group_remove_permission(test_db, client, admin_user): group = test_db.query(UserGroup).scalar() + num_permissions_before = len(group.permissions) response = client.get( f"/admin/groups/{group.id}/remove_permission/1", user=admin_user, @@ -104,7 +106,7 @@ def test_group_remove_permission(test_db, client, admin_user): ) assert response.status_code == 303 group = test_db.execute(select(UserGroup).where(UserGroup.id == group.id)).scalar() - assert 0 == len(group.permissions) + assert num_permissions_before - 1 == len(group.permissions) def test_create_group(test_db, client, admin_user): @@ -128,3 +130,60 @@ def test_delete_group(test_db, client, admin_user): test_db.execute(select(UserGroup).where(UserGroup.id == group.id)).scalar() is None ) + + +def test_create_account(test_db, client, admin_user): + response = client.post( + "/admin/accounts/new", + data={"account_name": "New Account"}, + user=admin_user, + follow_redirects=False, + ) + assert response.status_code == 303 + assert test_db.query(Account).filter_by(name="New Account").scalar() is not None + + # try to create another account with the same name, should fail + response = client.post( + "/admin/accounts/new", + data={"account_name": "New Account"}, + user=admin_user, + follow_redirects=False, + ) + assert response.status_code == 400 + + +def test_add_user_to_account(test_db, client, admin_user): + user_info = {"username": "test", "display_name": "Display Test"} + user = ensure_user(user_info, test_db) + + account = Account(name="Test Account") + test_db.add(account) + test_db.flush() + + response = client.post( + f"/admin/accounts/{account.id}/add_user", + data={"user_id": user.id}, + user=admin_user, + follow_redirects=False, + ) + assert response.status_code == 303 + + account = test_db.execute(select(Account).where(Account.id == account.id)).scalar() + assert any(u.username == "test" for u in account.users) + + +def test_add_balance_to_account(test_db, client, admin_user): + account = Account(name="Test Account") + test_db.add(account) + test_db.flush() + + response = client.post( + f"/admin/accounts/{account.id}/add_balance", + data={"amount": "100.00"}, + user=admin_user, + follow_redirects=False, + ) + assert response.status_code == 303 + + account = test_db.execute(select(Account).where(Account.id == account.id)).scalar() + assert account.balance == 100.00