Enh/own assets (#511)

Author: PGijsbers

src/database/review.py

@@ -6,10 +6,10 @@
 from sqlalchemy import Column, select
 from sqlmodel import SQLModel, Field, Relationship, Session
 
-import routers
 from database.model.field_length import NORMAL, LONG
 from database.model.concept.concept import AIoDConcept
 from database.model.helper_functions import non_abstract_subclasses
+from routers.helper_functions import get_all_asset_schemas
 
 REQUIRED_NUMBER_OF_REVIEWS = 1
 
@@ -121,25 +121,9 @@ class Config:
         # ResourceRead classes are only defined at runtime (generated dynamically).
         @staticmethod
         def schema_extra(schema: dict[str, Any], _: type["SubmissionView"]) -> None:
-            available_schemas: list[AIoDConcept] = list(non_abstract_subclasses(AIoDConcept))
-            classes_dict = {
-                clz.__tablename__: clz for clz in available_schemas if clz.__tablename__
-            }
-            resrouters = {
-                route.resource_name: route
-                for route in routers.resource_routers.router_list  # type: ignore
-            }
-            read_classes_dict = {
-                name: resrouters[name].resource_class_read for name in classes_dict
-            }
-
-            responses = [
-                {"$ref": f"#/components/schemas/{clz.__name__}"}
-                for clz in read_classes_dict.values()
-            ]
             schema["properties"]["asset"] = {
                 "title": "Asset under review",
                 "description": "The type of the object can be found in SubmissionView.asset_type.",
                 "type": "object",
-                "anyOf": responses,
+                "anyOf": get_all_asset_schemas(),
             }

src/main.py

@@ -31,6 +31,7 @@
     uploader_routers,
     search_routers,
     review_router,
+    user_router,
 )
 from setup_logger import setup_logger
 
@@ -101,7 +102,7 @@ def counts() -> dict:
         + enum_routers.router_list
         + search_routers.router_list
         + uploader_routers.router_list
-        + [review_router]
+        + [review_router, user_router]
     ):
         app.include_router(router.create(url_prefix))
 

src/routers/helper_functions.py

@@ -0,0 +1,21 @@
+import routers
+from database.model.concept.concept import AIoDConcept
+from database.model.helper_functions import non_abstract_subclasses
+
+
+def get_all_asset_schemas():
+    available_schemas: list[AIoDConcept] = list(non_abstract_subclasses(AIoDConcept))
+    classes_dict = {clz.__tablename__: clz for clz in available_schemas if clz.__tablename__}
+    resrouters = {
+        route.resource_name: route
+        for route in routers.resource_routers.router_list  # type: ignore
+    }
+    read_classes_dict = {
+        name: resrouters[name].resource_class_read
+        for name in classes_dict
+        if name not in ["testresource", "test_object"]
+    }
+    responses = [
+        {"$ref": f"#/components/schemas/{clz.__name__}"} for clz in read_classes_dict.values()
+    ]
+    return responses

src/routers/user_router.py

@@ -0,0 +1,75 @@
+from http import HTTPStatus
+
+from fastapi import APIRouter, Depends
+from sqlalchemy import select
+from sqlmodel import Session
+
+from authentication import KeycloakUser, get_user_or_raise
+from database.authorization import Permission, PermissionType
+from database.session import get_session
+from database.model.concept.aiod_entry import AIoDEntryORM
+from database.model.concept.concept import AIoDConcept
+from database.model.helper_functions import non_abstract_subclasses
+from routers.helper_functions import get_all_asset_schemas
+
+
+def create(url_prefix: str) -> APIRouter:
+    router = APIRouter()
+    version = "v1"
+
+    router.get(
+        f"{url_prefix}/user/resources/{version}",
+        tags=["User"],
+        description="Return all assets for which you have administrator rights",
+        response_model=None,  # Required! Otherwise FastAPI infers it from type annotation.
+        responses={
+            HTTPStatus.OK: {
+                "content": {
+                    "application/json": {
+                        "schema": {
+                            "title": "List of assets owned by the user.",
+                            "type": "array",
+                            "items": {"anyOf": get_all_asset_schemas()},
+                        }
+                    }
+                },
+            }
+        },
+    )(get_resources_for_logged_in_user)
+    return router
+
+
+def get_resources_for_logged_in_user(
+    user: KeycloakUser = Depends(get_user_or_raise),
+    session: Session = Depends(get_session),
+) -> list[AIoDConcept]:
+    return _get_resources_for_user(user, session)
+
+
+def _get_resources_for_user(user: KeycloakUser, session: Session) -> list[AIoDConcept]:
+    # "Ownership" is currently equivalent to having ADMIN permissions
+    stmt = (
+        select(AIoDEntryORM)
+        .join(Permission.aiod_entry)
+        .where(
+            Permission.user_identifier == user._subject_identifier,
+            Permission.type_ == PermissionType.ADMIN,
+        )
+    )
+    entries = session.scalars(stmt).all()
+    assets_to_fetch = [entry.identifier for entry in entries]
+    # We have AIoD entries, but want their respective asset information (e.g. publication).
+    # We lack the information about what the type of the asset is, so unfortunately we
+    # have to check all tables:
+    found_assets = []
+    for asset_type in non_abstract_subclasses(AIoDConcept):
+        query = select(asset_type).where(asset_type.aiod_entry_identifier.in_(assets_to_fetch))
+        assets = session.scalars(query).all()
+        found_assets.extend(assets)
+        if len(found_assets) == len(assets_to_fetch):
+            return found_assets  # minor optimization since queries may be expensive
+
+    raise RuntimeError(
+        f"Expected to find assets for identifiers {assets_to_fetch}, "
+        f"but only found {len(found_assets)} in total: {found_assets}."
+    )

src/tests/test_user_endpoints.py

@@ -0,0 +1,90 @@
+from http import HTTPStatus
+from typing import Callable
+
+from starlette.testclient import TestClient
+
+from database.authorization import set_permission, PermissionType, register_user
+from database.model.knowledge_asset.publication import Publication
+from database.model.concept.aiod_entry import EntryStatus
+from database.session import DbSession
+from database.model.agent.organisation import Organisation
+from tests.testutils.users import register_asset, logged_in_user, ALICE, BOB
+from tests.testutils.default_instances import publication_factory, publication
+
+
+def test_my_resources_can_be_empty(client: TestClient) -> None:
+    with logged_in_user(ALICE):
+        response = client.get("/user/resources/v1", headers={"Authorization": "fake token"})
+    assert response.status_code == HTTPStatus.OK
+    assert response.json() == [], "A user with no resources should get an empty list"
+
+
+def test_my_resources_shows_draft_assets(client: TestClient, publication: Publication) -> None:
+    register_asset(publication, owner=ALICE, status=EntryStatus.DRAFT)
+    with logged_in_user(ALICE):
+        response = client.get("/user/resources/v1", headers={"Authorization": "fake token"})
+    assert response.status_code == HTTPStatus.OK
+    assert len(response.json()) == 1, "Draft assets should be included in this view."
+
+
+def test_my_resources_shows_published_assets(client: TestClient, publication: Publication) -> None:
+    register_asset(publication, owner=ALICE, status=EntryStatus.PUBLISHED)
+    with logged_in_user(ALICE):
+        response = client.get("/user/resources/v1", headers={"Authorization": "fake token"})
+    assert response.status_code == HTTPStatus.OK
+    assert len(response.json()) == 1, "Published assets should be included in this view."
+
+
+def test_my_resources_shows_mixed_assets(client: TestClient, publication: Publication, organisation: Organisation) -> None:
+    register_asset(publication, owner=ALICE, status=EntryStatus.DRAFT)
+    register_asset(organisation, owner=ALICE, status=EntryStatus.PUBLISHED)
+    with logged_in_user(ALICE):
+        response = client.get("/user/resources/v1", headers={"Authorization": "fake token"})
+    assert response.status_code == HTTPStatus.OK
+
+    pub = next(asset for asset in response.json() if asset["aiod_entry_identifier"] == 1)
+    org = next(asset for asset in response.json() if asset["aiod_entry_identifier"] == 2)
+    dataset_property = "legal_name"
+    publication_property = "isbn"
+
+    assert dataset_property in org and publication_property in pub, "Assets should report properties unique to their type"
+    assert dataset_property not in pub and publication_property not in org, "Assets should not report properties they do not have"
+
+def test_my_resources_shows_only_own_resources(client: TestClient, publication_factory: Callable[[], Publication]) -> None:
+    register_asset(publication_factory(), owner=ALICE, status=EntryStatus.DRAFT)
+    register_asset(publication_factory(), owner=ALICE, status=EntryStatus.PUBLISHED)
+    register_asset(publication_factory(), owner=BOB, status=EntryStatus.PUBLISHED)
+
+    with logged_in_user(ALICE):
+        response = client.get("/user/resources/v1", headers={"Authorization": "fake token"})
+        assert len(response.json()) == 2
+
+    with logged_in_user(BOB):
+        response = client.get("/user/resources/v1", headers={"Authorization": "fake token"})
+        assert len(response.json()) == 1
+
+
+def test_my_resources_counts_only_if_admin(client: TestClient, publication_factory: Callable[[], Publication]) -> None:
+    asset_one = publication_factory()
+    identifier_one = register_asset(asset_one, owner=ALICE, status=EntryStatus.PUBLISHED)
+    asset_two = publication_factory()
+    identifier_two = register_asset(asset_two, owner=ALICE, status=EntryStatus.PUBLISHED)
+    asset_three = publication_factory()
+    identifier_three = register_asset(asset_three, owner=ALICE, status=EntryStatus.PUBLISHED)
+
+    with DbSession() as session:
+        register_user(BOB, session)
+        set_permission(BOB, session.get(Publication, identifier_one).aiod_entry, session, type_=PermissionType.READ)
+        set_permission(BOB, session.get(Publication, identifier_two).aiod_entry, session, type_=PermissionType.WRITE)
+        set_permission(BOB, session.get(Publication, identifier_three).aiod_entry, session, type_=PermissionType.ADMIN)
+        session.commit()
+
+    with logged_in_user(BOB):
+        response = client.get("/user/resources/v1", headers={"Authorization": "fake token"})
+        assert len(response.json()) == 1, "Bob has ADMIN permission to one asset."
+        assert response.json()[0]["identifier"] == identifier_three
+
+
+def test_my_resources_must_be_authorized(client: TestClient) -> None:
+    response = client.get("/user/resources/v1")
+    assert response.status_code == HTTPStatus.UNAUTHORIZED

src/tests/testutils/default_instances.py

@@ -6,6 +6,7 @@
 
 import copy
 import json
+import uuid
 from functools import partial
 from typing import Callable
 
@@ -75,12 +76,14 @@ def body_agent(body_resource: dict, load_body_agent: dict) -> dict:
     return copy.deepcopy(body)
 
 
-def make_publication(body_asset: dict) -> Publication:
+def make_publication(body_asset: dict, with_random_platform_identifier: bool = False) -> Publication:
     body = copy.deepcopy(body_asset)
     body["permanent_identifier"] = "http://dx.doi.org/10.1093/ajae/aaq063"
     body["isbn"] = "9783161484100"
     body["issn"] = "20493630"
     body["type"] = "journal"
+    if with_random_platform_identifier:
+        body["platform_resource_identifier"] = uuid.uuid4().hex
     return _create_class_with_body(Publication, body)
 
 
@@ -91,7 +94,7 @@ def publication(body_asset: dict) -> Publication:
 
 @pytest.fixture
 def publication_factory(body_asset: dict) -> Callable[[], Publication]:
-    return partial(make_publication, body_asset)
+    return partial(make_publication, body_asset, with_random_platform_identifier=True)
 
 
 @pytest.fixture