LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Database cleanup: make it synchronous because we work with in-memory data. Defer writes to disk and cleanup to background task. Tests passing.

Browse Source
This commit is contained in:
LeoVasanko
2026-01-23 01:22:47 +00:00
parent 7f3763b46d
commit 02e04da2c4
14 changed files with 548 additions and 556 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_admin.py
+28 -26
View File
@@ -35,7 +35,7 @@ async def second_org(test_db: DB) -> Org:
display_name="Second Organization",
permissions=[],
)
await test_db.create_organization(org)
test_db.create_organization(org)
return org
@@ -50,7 +50,7 @@ async def second_org_role(
display_name="Second Org Admin Role",
permissions=["auth:admin"],
)
await test_db.create_role(role)
test_db.create_role(role)
return role
@@ -64,7 +64,7 @@ async def second_org_user(test_db: DB, second_org_role: Role) -> User:
created_at=datetime.now(timezone.utc),
visits=0,
)
await test_db.create_user(user)
test_db.create_user(user)
return user
@@ -84,7 +84,7 @@ async def second_org_credential(test_db: DB, second_org_user: User) -> Credentia
last_used=datetime.now(timezone.utc),
last_verified=datetime.now(timezone.utc),
)
await test_db.create_credential(credential)
test_db.create_credential(credential)
return credential
@@ -94,7 +94,7 @@ async def second_org_session_token(
) -> str:
"""Create a session for the second org admin user."""
token = create_token()
await test_db.create_session(
test_db.create_session(
user_uuid=second_org_user.uuid,
credential_uuid=second_org_credential.uuid,
key=session_key(token),
@@ -115,7 +115,7 @@ async def org_admin_role(test_db: DB, test_org: Org) -> Role:
display_name="Org Admin Role",
permissions=[f"auth:org:{test_org.uuid}"],
)
await test_db.create_role(role)
test_db.create_role(role)
return role
@@ -130,7 +130,7 @@ async def org_admin_user(test_db: DB, org_admin_role: Role) -> User:
visits=5,
last_seen=datetime.now(timezone.utc),
)
await test_db.create_user(user)
test_db.create_user(user)
return user
@@ -150,7 +150,7 @@ async def org_admin_credential(test_db: DB, org_admin_user: User) -> Credential:
last_used=datetime.now(timezone.utc),
last_verified=None,
)
await test_db.create_credential(credential)
test_db.create_credential(credential)
return credential
@@ -160,7 +160,7 @@ async def org_admin_session_token(
) -> str:
"""Create a session for the org admin user."""
token = create_token()
await test_db.create_session(
test_db.create_session(
user_uuid=org_admin_user.uuid,
credential_uuid=org_admin_credential.uuid,
key=session_key(token),
@@ -176,9 +176,9 @@ async def org_admin_session_token(
async def grantable_permission(test_db: DB, test_org: Org) -> Permission:
"""Create a permission and add it to org's grantable permissions."""
perm = Permission(id="test:grantable:perm", display_name="Grantable Perm")
await test_db.create_permission(perm)
test_db.create_permission(perm)
# Add to org's grantable permissions
await test_db.add_permission_to_organization(str(test_org.uuid), perm.id)
test_db.add_permission_to_organization(str(test_org.uuid), perm.id)
return perm
@@ -375,12 +375,12 @@ class TestAdminOrganizations:
org_admin_perm_id = f"auth:org:{test_org.uuid}"
perm = Permission(id=org_admin_perm_id, display_name="Org Admin")
try:
await test_db.create_permission(perm)
test_db.create_permission(perm)
except Exception:
pass # Permission may already exist
# Add it to the org's permissions
await test_db.add_permission_to_organization(
test_db.add_permission_to_organization(
str(test_org.uuid), org_admin_perm_id
)
@@ -424,13 +424,13 @@ class TestAdminOrganizations:
display_name="Org To Delete",
permissions=[],
)
await test_db.create_organization(org_to_delete)
test_db.create_organization(org_to_delete)
# Create some org-specific permissions to test cleanup
org_perm = Permission(
id=f"test:org:{org_to_delete.uuid}:feature", display_name="Org Feature"
)
await test_db.create_permission(org_perm)
test_db.create_permission(org_perm)
response = await client.delete(
f"/auth/api/admin/orgs/{org_to_delete.uuid}",
@@ -603,7 +603,7 @@ class TestAdminRoles:
"""Creating role with non-grantable permission should fail."""
# Create permission but don't add to org
perm = Permission(id="test:not:grantable", display_name="Not Grantable")
await test_db.create_permission(perm)
test_db.create_permission(perm)
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/roles",
@@ -673,7 +673,7 @@ class TestAdminRoles:
):
"""Adding non-grantable permission to role should fail."""
perm = Permission(id="test:not:grantable:update", display_name="Not Grantable")
await test_db.create_permission(perm)
test_db.create_permission(perm)
response = await client.put(
f"/auth/api/admin/orgs/{test_org.uuid}/roles/{user_role.uuid}",
@@ -1087,7 +1087,7 @@ class TestAdminUsersInOrg:
created_at=datetime.now(timezone.utc),
visits=0,
)
await test_db.create_user(user_no_cred)
test_db.create_user(user_no_cred)
response = await client.post(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{user_no_cred.uuid}/create-link",
@@ -1174,7 +1174,7 @@ class TestAdminSessions:
# Create an additional session to delete
extra_token = create_token()
extra_key = session_key(extra_token)
await test_db.create_session(
test_db.create_session(
user_uuid=test_user.uuid,
credential_uuid=test_credential.uuid,
key=extra_key,
@@ -1301,7 +1301,7 @@ class TestAdminPermissions:
test_org,
grantable_permission,
):
"""Org admin should only see grantable permissions."""
"""Org admin should only see permissions their org can grant."""
response = await client.get(
"/auth/api/admin/permissions",
headers={**auth_headers(org_admin_session_token), "Host": "localhost:4401"},
@@ -1311,8 +1311,10 @@ class TestAdminPermissions:
# Should only see permissions the org can grant
perm_ids = [p["id"] for p in data]
assert grantable_permission.id in perm_ids
# Should NOT see auth:admin (not grantable by org)
assert "auth:admin" not in perm_ids
# test_org CAN grant auth:admin (it's in org.permissions), so org admin sees it
assert "auth:admin" in perm_ids
# Should also see auto-created org admin permission
assert f"auth:org:{test_org.uuid}" in perm_ids
@pytest.mark.asyncio
async def test_create_permission(
@@ -1364,7 +1366,7 @@ class TestAdminPermissions:
"""Admin should be able to update a permission."""
# Create permission first
perm = Permission(id="test:updateable", display_name="Updateable")
await test_db.create_permission(perm)
test_db.create_permission(perm)
response = await client.put(
"/auth/api/admin/permission?permission_id=test:updateable&display_name=Updated%20Name",
@@ -1394,7 +1396,7 @@ class TestAdminPermissions:
"""Admin should be able to rename a permission."""
# Create permission first
perm = Permission(id="test:renameable2", display_name="Renameable")
await test_db.create_permission(perm)
test_db.create_permission(perm)
response = await client.post(
"/auth/api/admin/permission/rename",
@@ -1437,7 +1439,7 @@ class TestAdminPermissions:
):
"""Renaming permission can also update display name."""
perm = Permission(id="test:rename:withname", display_name="Old Name")
await test_db.create_permission(perm)
test_db.create_permission(perm)
response = await client.post(
"/auth/api/admin/permission/rename",
@@ -1457,7 +1459,7 @@ class TestAdminPermissions:
"""Admin should be able to delete a permission."""
# Create permission first
perm = Permission(id="test:deleteable", display_name="Deleteable")
await test_db.create_permission(perm)
test_db.create_permission(perm)
response = await client.delete(
"/auth/api/admin/permission?permission_id=test:deleteable",