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 %}
+
+
{% 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