aboutsummaryrefslogtreecommitdiff
path: root/gn_auth
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2024-06-20 15:34:02 -0500
committerFrederick Muriuki Muriithi2024-06-20 15:34:02 -0500
commitc0f5b9d646487e035f2d2e5370041b317b81baf6 (patch)
treee1cce317553a2379aa1717d700ded55ca9418440 /gn_auth
parent8e460b05da4d419aa1b53b1c639d3e370143de4f (diff)
downloadgn-auth-c0f5b9d646487e035f2d2e5370041b317b81baf6.tar.gz
Reorganise test fixtures. Fix tests and issues caught.
Reorganise test fixtures to more closely follow the design of the auth system. Fix the broken tests due to refactors and fix all issues caught by the running tests.
Diffstat (limited to 'gn_auth')
-rw-r--r--gn_auth/auth/authorisation/resources/groups/models.py2
-rw-r--r--gn_auth/auth/authorisation/resources/models.py50
2 files changed, 27 insertions, 25 deletions
diff --git a/gn_auth/auth/authorisation/resources/groups/models.py b/gn_auth/auth/authorisation/resources/groups/models.py
index 03a93b6..ee77654 100644
--- a/gn_auth/auth/authorisation/resources/groups/models.py
+++ b/gn_auth/auth/authorisation/resources/groups/models.py
@@ -72,7 +72,7 @@ def user_membership(conn: db.DbConnection, user: User) -> Sequence[Group]:
"WHERE group_users.user_id=?")
with db.cursor(conn) as cursor:
cursor.execute(query, (str(user.user_id),))
- groups = tuple(Group(row[0], row[1], json.loads(row[2]))
+ groups = tuple(Group(row[0], row[1], json.loads(row[2] or "{}"))
for row in cursor.fetchall())
return groups
diff --git a/gn_auth/auth/authorisation/resources/models.py b/gn_auth/auth/authorisation/resources/models.py
index c6c2e9e..94e817d 100644
--- a/gn_auth/auth/authorisation/resources/models.py
+++ b/gn_auth/auth/authorisation/resources/models.py
@@ -4,6 +4,8 @@ from uuid import UUID, uuid4
from functools import reduce, partial
from typing import Dict, Sequence, Optional
+import sqlite3
+
from gn_auth.auth.db import sqlite3 as db
from gn_auth.auth.authentication.users import User
from gn_auth.auth.db.sqlite3 import with_db_connection
@@ -48,6 +50,19 @@ def __assign_resource_owner_role__(cursor, resource, user):
"resource_id": str(resource.resource_id)
})
+
+def resource_from_dbrow(row: sqlite3.Row):
+ """Convert an SQLite3 resultset row into a resource."""
+ return Resource(
+ resource_id=UUID(row["resource_id"]),
+ resource_name=row["resource_name"],
+ resource_category=ResourceCategory(
+ UUID(row["resource_category_id"]),
+ row["resource_category_key"],
+ row["resource_category_description"]),
+ public=bool(int(row["public"])))
+
+
@authorised_p(("group:resource:create-resource",),
error_description="Insufficient privileges to create a resource",
oauth2_scope="profile resource")
@@ -135,32 +150,19 @@ def group_leader_resources(
def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]:
"""List the resources available to the user"""
- categories = { # Repeated in `public_resources` function
- cat.resource_category_id: cat for cat in resource_categories(conn)
- }
with db.cursor(conn) as cursor:
- def __all_resources__(group) -> Sequence[Resource]:
- gl_resources = group_leader_resources(conn, user, group, categories)
+ cursor.execute(
+ ("SELECT r.*, rc.resource_category_key, "
+ "rc.resource_category_description FROM user_roles AS ur "
+ "INNER JOIN resources AS r ON ur.resource_id=r.resource_id "
+ "INNER JOIN resource_categories AS rc "
+ "ON r.resource_category_id=rc.resource_category_id "
+ "WHERE ur.user_id=?"),
+ (str(user.user_id),))
+ rows = cursor.fetchall() or []
+
+ return tuple(resource_from_dbrow(row) for row in rows)
- cursor.execute(
- ("SELECT resources.* FROM user_roles LEFT JOIN resources "
- "ON user_roles.resource_id=resources.resource_id "
- "WHERE user_roles.user_id=?"),
- (str(user.user_id),))
- rows = cursor.fetchall()
- private_res = tuple(
- Resource(UUID(row[0]), row[1], categories[UUID(row[2])],
- bool(row[3]))
- for row in rows)
- return tuple({
- res.resource_id: res
- for res in
- (private_res + gl_resources + public_resources(conn))# type: ignore[operator]
- }.values())
-
- # Fix the typing here
- return user_group(conn, user).map(__all_resources__).maybe(# type: ignore[arg-type,misc]
- public_resources(conn), lambda res: res)# type: ignore[arg-type,return-value]
def resource_data(conn, resource, offset: int = 0, limit: Optional[int] = None) -> tuple[dict, ...]:
"""