[Fixes #14555] Add asset GET Endpoint

docs/src/development/api.md

@@ -978,6 +978,35 @@ response = requests.request("GET", url)
 
 ## Assets
 
+### Get Resource Assets
+
+Retrieve the list of assets attached to a specific resource.
+
+**Endpoint**:
+
+- **API**: `GET /api/v2/resources/{pk}/asset`
+- **Status Code**: `200`
+
+where `{pk}` is the resource ID.
+
+**Permissions**:
+- Users with `view_resourcebase` permission on the resource can view all assets attached to that resource.
+- Users without `view_resourcebase` permission receive `403 Forbidden`.
+
+**Example Request**:
+```python
+import requests
+
+url = "https://development.demo.geonode.org/api/v2/resources/{pk}/asset"
+
+headers = {
+    'Authorization': 'Basic dXNlcjpwYXNzd29yZA=='
+}
+
+response = requests.request("GET", url, headers=headers)
+
+```
+
 ### Assets Upload
 
 Assets can be attached to a resource as a file.

geonode/api/tests.py

@@ -1390,6 +1390,131 @@ def test_delete_asset_no_permission(self):
         self.assertTrue(Asset.objects.filter(pk=self.asset1.pk).exists())
 
 
+class AssetGetApiTests(GeoNodeBaseTestSupport):
+
+    def setUp(self):
+        super().setUp()
+        self.admin_user = get_user_model().objects.get(username="admin")
+        self.owner_user = get_user_model().objects.create_user(
+            username="asset_owner", email="owner@example.com", password="password123"
+        )
+        self.other_user = get_user_model().objects.create_user(
+            username="other_user", email="other@example.com", password="password123"
+        )
+        self.no_perm_user = get_user_model().objects.create_user(
+            username="no_perm_user", email="noperm@example.com", password="password123"
+        )
+
+        self.resource = ResourceBase.objects.create(
+            title="Test Resource for Asset Retrieval",
+            owner=self.owner_user,
+            uuid=str(uuid4()),
+        )
+
+        self.resource.set_permissions(
+            {
+                "users": {
+                    self.owner_user.username: ["view_resourcebase", "change_resourcebase"],
+                    self.other_user.username: ["view_resourcebase"],
+                },
+                "groups": {
+                    "anonymous": ["view_resourcebase"],
+                },
+            }
+        )
+
+        self.asset_owner = Asset.objects.create(
+            title="Owner Asset.pdf",
+            description="Created by owner",
+            type="file",
+            owner=self.owner_user,
+        )
+        self.owner_link = Link.objects.create(
+            resource=self.resource,
+            asset=self.asset_owner,
+            name=self.asset_owner.title,
+            link_type="uploaded",
+            extension="pdf",
+            url="http://example.com/owner_asset.pdf",
+        )
+
+        self.asset_admin = Asset.objects.create(
+            title="Admin Asset.png",
+            description="Created by admin",
+            type="image",
+            owner=self.admin_user,
+        )
+        self.admin_link = Link.objects.create(
+            resource=self.resource,
+            asset=self.asset_admin,
+            name=self.asset_admin.title,
+            link_type="image",
+            extension="png",
+            url="http://example.com/admin_asset.png",
+        )
+
+        self.url = reverse("base-resources-asset", kwargs={"pk": self.resource.pk})
+
+    def assert_asset_response(self, response):
+        self.assertEqual(response.status_code, 200)
+        self.assertIsInstance(response.data, list)
+        self.assertEqual(len(response.data), 2)
+
+        returned_ids = {item["id"] for item in response.data}
+        self.assertEqual(returned_ids, {self.asset_owner.id, self.asset_admin.id})
+
+        returned_links = {item["urls"]["link"] for item in response.data}
+        self.assertEqual(returned_links, {self.owner_link.url, self.admin_link.url})
+
+        for item in response.data:
+            self.assertIn("id", item)
+            self.assertIn("title", item)
+            self.assertIn("description", item)
+            self.assertIn("type", item)
+            self.assertIn("created", item)
+            self.assertIn("deletable", item)
+            self.assertIn("urls", item)
+            self.assertIn("download_url", item["urls"])
+            self.assertIn("link", item["urls"])
+
+    def test_admin_can_get_resource_assets(self):
+        self.client.force_login(self.admin_user)
+
+        response = self.client.get(self.url)
+
+        self.assert_asset_response(response)
+
+    def test_resource_viewer_can_get_resource_assets(self):
+        self.client.force_login(self.other_user)
+
+        response = self.client.get(self.url)
+
+        self.assert_asset_response(response)
+
+    def test_anonymous_with_view_permission_can_get_resource_assets(self):
+        self.client.logout()
+
+        response = self.client.get(self.url)
+
+        self.assert_asset_response(response)
+
+    def test_user_without_view_permission_cannot_get_resource_assets(self):
+        self.resource.set_permissions(
+            {
+                "users": {
+                    self.owner_user.username: ["view_resourcebase", "change_resourcebase"],
+                },
+                "groups": {},
+            }
+        )
+
+        self.client.force_login(self.no_perm_user)
+
+        response = self.client.get(self.url)
+
+        self.assertEqual(response.status_code, 403)
+
+
 class RouterUrlpatternsCompletenessTest(SimpleTestCase):
     """Guards the shared /api/v2/ router against app-loading-order regressions."""
 

geonode/base/api/serializers.py

@@ -28,7 +28,7 @@
 from django.forms.models import model_to_dict
 from django.contrib.auth import get_user_model
 
-from geonode.assets.utils import get_default_asset, is_asset_deletable
+from geonode.assets.utils import get_default_asset
 from geonode.metadata.multilang.serializers import MultiLangOutputMixin
 from geonode.people import Roles
 from django.http import QueryDict
@@ -572,25 +572,9 @@ def to_representation(self, instance):
         links = Link.objects.filter(
             resource_id=instance,  # link_type__in=["OGC:WMS", "OGC:WFS", "OGC:WCS", "image", "metadata"]
         )
-        request = self.context.get("request", None)
         for lnk in links:
             formatted_link = model_to_dict(lnk, fields=link_fields)
             ret.append(formatted_link)
-            if lnk.asset:
-                deletable = is_asset_deletable(lnk.asset)
-                extras = {
-                    "type": "asset",
-                    "deletable": deletable,
-                    "content": model_to_dict(lnk.asset, ["title", "description", "type", "created"]),
-                }
-                if request and permissions_registry.user_has_perm(
-                    request.user, lnk.resource.get_self_resource(), "download_resourcebase", include_virtual=True
-                ):
-                    extras["content"]["download_url"] = asset_handler_registry.get_handler(
-                        lnk.asset
-                    ).create_download_url(lnk.asset)
-                formatted_link["extras"] = extras
-
         return ret
 
 

geonode/base/api/tests.py

@@ -4404,55 +4404,95 @@ def tearDownClass(cls):
 class AssetDownloadPermissionTests(GeoNodeBaseTestSupport):
 
     def setUp(self):
-        self.user_viewer = get_user_model().objects.create_user(username="viewer", password="viewer_password")
-        self.user_downloader = get_user_model().objects.create_user(
+        super().setUp()
+        self.admin_user = get_user_model().objects.get(username="admin")
+        self.viewer_user = get_user_model().objects.create_user(username="viewer", password="viewer_password")
+        self.downloader_user = get_user_model().objects.create_user(
             username="downloader", password="downloader_password"
         )
+
         self.dataset = create_single_dataset(name="test_dataset_for_download_permission")
 
         dataset_files = [
             f"{settings.PROJECT_ROOT}/assets/tests/data/one.json",
         ]
-        create_asset_and_link(
-            self.dataset, get_user_model().objects.get(username="admin"), dataset_files, clone_files=False
+
+        self.owner_asset, self.owner_link = create_asset_and_link(
+            self.dataset,
+            self.viewer_user,
+            dataset_files,
+            title="Owner Asset",
+            clone_files=False,
+        )
+        self.admin_asset, self.admin_link = create_asset_and_link(
+            self.dataset,
+            self.admin_user,
+            dataset_files,
+            title="Admin Asset",
+            clone_files=False,
         )
 
-        perm_spec = {
-            "users": {
-                self.user_viewer.username: [
-                    "view_resourcebase",
-                ],
-                self.user_downloader.username: [
-                    "view_resourcebase",
-                    "download_resourcebase",
-                ],
-            },
-            "groups": {},
-        }
-        self.dataset.set_permissions(perm_spec)
+        self.dataset.set_permissions(
+            {
+                "users": {
+                    self.viewer_user.username: ["view_resourcebase"],
+                    self.downloader_user.username: ["view_resourcebase", "download_resourcebase"],
+                },
+                "groups": {},
+            }
+        )
+
+        self.url = reverse("base-resources-asset", kwargs={"pk": self.dataset.pk})
+
+    def test_admin_sees_all_asset_download_urls(self):
+        self.client.force_login(self.admin_user)
+
+        response = self.client.get(self.url)
 
-    def test_asset_download_link_permission(self):
-        """
-        Ensure that the download link is only visible to users with download permissions.
-        """
-        self.client.login(username="viewer", password="viewer_password")
-        url = f"/api/v2/datasets/{self.dataset.pk}"
-        response = self.client.get(f"{url}", format="json")
         self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 2)
 
-        links = response.data.get("dataset").get("links", [])
-        asset_link = next((link for link in links if link.get("extras", {}).get("type") == "asset"), None)
-        self.assertIsNotNone(asset_link)
-        self.assertNotIn("download_url", asset_link.get("extras", {}).get("content", {}))
+        returned_ids = {item["id"] for item in response.data}
+        self.assertEqual(returned_ids, {self.owner_asset.id, self.admin_asset.id})
+
+        for item in response.data:
+            self.assertIsNotNone(item["urls"]["download_url"])
+            self.assertIsNotNone(item["urls"]["link"])
+
+    def test_viewer_sees_all_assets_without_download_urls(self):
+        self.client.force_login(self.viewer_user)
+
+        response = self.client.get(self.url)
 
-        self.client.login(username="downloader", password="downloader_password")
-        response = self.client.get(f"{url}", format="json")
         self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 2)
 
-        links = response.data.get("dataset").get("links", [])
-        asset_link = next((link for link in links if link.get("extras", {}).get("type") == "asset"), None)
-        self.assertIsNotNone(asset_link)
-        self.assertIn("download_url", asset_link.get("extras", {}).get("content", {}))
+        returned_ids = {item["id"] for item in response.data}
+        self.assertEqual(returned_ids, {self.owner_asset.id, self.admin_asset.id})
+
+        for item in response.data:
+            self.assertIsNone(item["urls"]["download_url"])
+            self.assertIsNotNone(item["urls"]["link"])
+
+    def test_downloader_sees_all_assets_with_download_urls(self):
+        self.client.force_login(self.downloader_user)
+
+        response = self.client.get(self.url)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 2)
+
+        returned_ids = {item["id"] for item in response.data}
+        self.assertEqual(returned_ids, {self.owner_asset.id, self.admin_asset.id})
+
+        for item in response.data:
+            self.assertIsNotNone(item["urls"]["download_url"])
+            self.assertIsNotNone(item["urls"]["link"])
+
+    def test_anonymous_without_view_permission_cannot_get_assets(self):
+        response = self.client.get(self.url)
+
+        self.assertEqual(response.status_code, 401)
 
 
 class ResourceBaseMetadataXMLTest(GeoNodeBaseTestSupport):

geonode/base/api/views.py

@@ -109,7 +109,8 @@
 from geonode.base.utils import patch_perms
 from geonode.base.api.deprecated_extra_metadata import DeprecatedExtraMetadataMixin
 from geonode.assets.models import Asset
-from geonode.assets.utils import create_asset_and_link, unlink_asset
+from geonode.base.models import Link
+from geonode.assets.utils import create_asset_and_link, unlink_asset, is_asset_deletable
 from geonode.assets.handlers import asset_handler_registry
 from geonode.utils import get_supported_datasets_file_types
 from geonode.utils import assert_safe_xml, UnsafeXMLError
@@ -1406,6 +1407,55 @@ def asset(self, request, pk=None, *args, **kwargs):
             logger.exception(e)
             return Response({"message": "Error creating asset."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
+    @action(
+        detail=True,
+        methods=["get"],
+        url_path=r"asset",
+        url_name="asset",
+        permission_classes=[UserHasPerms(perms_dict={"default": {"GET": ["base.view_resourcebase"]}})],
+    )
+    def get_asset(self, request, pk=None, *args, **kwargs):
+        """
+        Returns assets linked to the resource.
+        GET /api/v2/resources/<pk>/asset
+        """
+        resource = self.get_object()
+
+        asset_links = Link.objects.filter(
+            resource=resource,
+            asset__isnull=False,
+        ).select_related("asset")
+
+        can_have_download_url = permissions_registry.user_has_perm(
+            request.user,
+            resource.get_self_resource(),
+            "download_resourcebase",
+            include_virtual=True,
+        )
+
+        data = []
+        for asset_link in asset_links:
+            asset = asset_link.asset.get_real_instance()
+            handler = asset_handler_registry.get_handler(asset)
+            download_url = handler.create_download_url(asset) if handler and can_have_download_url else None
+
+            data.append(
+                {
+                    "id": asset.id,
+                    "title": asset.title,
+                    "description": asset.description,
+                    "type": asset.type,
+                    "created": asset.created,
+                    "deletable": is_asset_deletable(asset),
+                    "urls": {
+                        "download_url": download_url,
+                        "link": asset_link.url,
+                    },
+                }
+            )
+
+        return Response(data, status=status.HTTP_200_OK)
+
     @action(detail=True, methods=["delete"], url_path=r"assets/(?P<asset_id>\d+)", url_name="delete-asset")
     def delete_asset(self, request, pk=None, asset_id=None, *args, **kwargs):
         """Deletes an asset and its link to the resource."""