[Fixes #10482] Upload ISO-19115 xml metadata via the API (#10483) (#1…

…0548)

* ISO-19115 xml metadata file can be uploaded via the API

* Small changes for PEP8 compliance

* Added tests for checking permissions for metadata upload

Co-authored-by: Alessio Fabiani <alessio.fabiani@geosolutionsgroup.com>
(cherry picked from commit 153736b)

Co-authored-by: ahmdthr <116570171+ahmdthr@users.noreply.github.com>

geonode/layers/api/exceptions.py

@@ -31,3 +31,17 @@ class InvalidDatasetException(APIException):
     default_detail = "Input payload is not valid"
     default_code = "invalid_dataset_exception"
     category = "dataset_api"
+
+
+class InvalidMetadataException(APIException):
+    status_code = 500
+    default_detail = "Input payload is not valid"
+    default_code = "invalid_metadata_exception"
+    category = "dataset_api"
+
+
+class MissingMetadataException(APIException):
+    status_code = 400
+    default_detail = "Metadata is missing"
+    default_code = "missing_metadata_exception"
+    category = "dataset_api"

geonode/layers/api/serializers.py

@@ -179,3 +179,18 @@ class Meta:
     xml_file = serializers.CharField(required=False)
     sld_file = serializers.CharField(required=False)
     store_spatial_files = serializers.BooleanField(required=False, default=True)
+
+
+class MetadataFileField(DynamicComputedField):
+
+    def get_attribute(self, instance):
+        return instance.get('metadata_file')
+
+
+class DatasetMetadataSerializer(serializers.Serializer):
+    metadata_file = MetadataFileField(required=True)
+
+    class Meta:
+        fields = (
+            "metadata_file"
+        )

geonode/layers/api/tests.py

@@ -16,6 +16,7 @@
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 #
 #########################################################################
+from io import BytesIO
 import logging
 
 from unittest.mock import patch
@@ -28,6 +29,7 @@
 from geonode.geoserver.createlayer.utils import create_dataset
 from guardian.shortcuts import assign_perm, get_anonymous_user
 
+from django.conf import settings
 from geonode.layers.models import Dataset, Attribute
 from geonode.base.populate_test_data import create_models, create_single_dataset
 from geonode.maps.models import Map, MapLayer
@@ -44,6 +46,7 @@ class DatasetsApiTests(APITestCase):
     ]
 
     def setUp(self):
+        self.exml_path = f"{settings.PROJECT_ROOT}/base/fixtures/test_xml.xml"
         create_models(b'document')
         create_models(b'map')
         create_models(b'dataset')
@@ -312,3 +315,96 @@ def test_layer_replace_should_work(self, _validate_input_source):
         layer.refresh_from_db()
         # evaluate that the number of available layer is not changed
         self.assertEqual(Dataset.objects.count(), cnt)
+
+    def test_metadata_update_for_not_supported_method(self):
+        layer = Dataset.objects.first()
+        url = reverse("datasets-replace-metadata", args=(layer.id,))
+        self.client.login(username="admin", password="admin")
+
+        response = self.client.post(url)
+        self.assertEqual(405, response.status_code)
+
+        response = self.client.get(url)
+        self.assertEqual(405, response.status_code)
+
+    def test_metadata_update_for_not_authorized_user(self):
+        layer = Dataset.objects.first()
+        url = reverse("datasets-replace-metadata", args=(layer.id,))
+
+        response = self.client.put(url)
+        self.assertEqual(403, response.status_code)
+
+    def test_unsupported_file_throws_error(self):
+        layer = Dataset.objects.first()
+        url = reverse("datasets-replace-metadata", args=(layer.id,))
+        self.client.login(username="admin", password="admin")
+
+        data = '<?xml version="1.0" encoding="UTF-8"?><invalid></invalid>'
+        f = BytesIO(bytes(data, encoding='utf-8'))
+        f.name = 'metadata.xml'
+        put_data = {'metadata_file': f}
+        response = self.client.put(url, data=put_data)
+        self.assertEqual(500, response.status_code)
+
+    def test_valid_metadata_file_with_different_uuid(self):
+        layer = Dataset.objects.first()
+        url = reverse("datasets-replace-metadata", args=(layer.id,))
+        self.client.login(username="admin", password="admin")
+
+        f = open(self.exml_path, 'r')
+        put_data = {'metadata_file': f}
+        response = self.client.put(url, data=put_data)
+        self.assertEqual(500, response.status_code)
+
+    def test_permissions_for_not_permitted_user(self):
+        get_user_model().objects.create_user(
+            username="some_user",
+            password="some_password",
+            email="some_user@geonode.org",
+        )
+        layer = Dataset.objects.first()
+        url = reverse("datasets-replace-metadata", args=(layer.id,))
+        self.client.login(username="some_user", password="some_password")
+
+        uuid = layer.uuid
+        data = open(self.exml_path).read()
+        data = data.replace('7cfbc42c-efa7-431c-8daa-1399dff4cd19', uuid)
+        f = BytesIO(bytes(data, encoding='utf-8'))
+        f.name = 'metadata.xml'
+        put_data = {'metadata_file': f}
+        response = self.client.put(url, data=put_data)
+        self.assertEqual(403, response.status_code)
+
+    def test_permissions_for_permitted_user(self):
+        another_non_admin_user = get_user_model().objects.create_user(
+            username="some_other_user",
+            password="some_other_password",
+            email="some_other_user@geonode.org",
+        )
+        layer = Dataset.objects.first()
+        assign_perm("base.change_resourcebase_metadata", another_non_admin_user, layer.get_self_resource())
+        url = reverse("datasets-replace-metadata", args=(layer.id,))
+        self.client.login(username="some_other_user", password="some_other_password")
+
+        uuid = layer.uuid
+        data = open(self.exml_path).read()
+        data = data.replace('7cfbc42c-efa7-431c-8daa-1399dff4cd19', uuid)
+        f = BytesIO(bytes(data, encoding='utf-8'))
+        f.name = 'metadata.xml'
+        put_data = {'metadata_file': f}
+        response = self.client.put(url, data=put_data)
+        self.assertEqual(200, response.status_code)
+
+    def test_valid_metadata_file(self):
+        layer = Dataset.objects.first()
+        url = reverse("datasets-replace-metadata", args=(layer.id,))
+        self.client.login(username="admin", password="admin")
+
+        uuid = layer.uuid
+        data = open(self.exml_path).read()
+        data = data.replace('7cfbc42c-efa7-431c-8daa-1399dff4cd19', uuid)
+        f = BytesIO(bytes(data, encoding='utf-8'))
+        f.name = 'metadata.xml'
+        put_data = {'metadata_file': f}
+        response = self.client.put(url, data=put_data)
+        self.assertEqual(200, response.status_code)

geonode/layers/api/views.py

@@ -23,7 +23,7 @@
 from dynamic_rest.filters import DynamicFilterBackend, DynamicSortingFilter
 
 from rest_framework.decorators import action
-from rest_framework.permissions import IsAuthenticatedOrReadOnly
+from rest_framework.permissions import IsAuthenticatedOrReadOnly, IsAuthenticated
 from rest_framework.authentication import SessionAuthentication, BasicAuthentication
 
 from oauth2_provider.contrib.rest_framework import OAuth2Authentication
@@ -32,16 +32,25 @@
 from geonode.base.api.filters import DynamicSearchFilter, ExtentFilter
 from geonode.base.api.pagination import GeoNodeApiPagination
 from geonode.base.api.permissions import UserHasPerms
-from geonode.layers.api.exceptions import GeneralDatasetException, InvalidDatasetException
+from geonode.layers.api.exceptions import (
+    GeneralDatasetException,
+    InvalidDatasetException,
+    InvalidMetadataException)
+from geonode.layers.metadata import parse_metadata
 from geonode.layers.models import Dataset
 from geonode.layers.utils import validate_input_source
 from geonode.maps.api.serializers import SimpleMapLayerSerializer, SimpleMapSerializer
+from geonode.resource.utils import update_resource
 from rest_framework.exceptions import NotFound
 
 from geonode.storage.manager import StorageManager
 from geonode.resource.manager import resource_manager
 
-from .serializers import DatasetReplaceAppendSerializer, DatasetSerializer, DatasetListSerializer
+from .serializers import (
+    DatasetReplaceAppendSerializer,
+    DatasetSerializer,
+    DatasetListSerializer,
+    DatasetMetadataSerializer)
 from .permissions import DatasetPermissionsFilter
 
 import logging
@@ -72,6 +81,84 @@ def get_serializer_class(self):
             return DatasetListSerializer
         return DatasetSerializer
 
+    @extend_schema(
+        request=DatasetMetadataSerializer,
+        methods=["put"],
+        responses={200},
+        description="API endpoint to upload metadata file.",
+    )
+    @action(
+        detail=False,
+        url_path="(?P<pk>\d+)/metadata",  # noqa
+        url_name="replace-metadata",
+        methods=["put"],
+        serializer_class=DatasetMetadataSerializer,
+        permission_classes=[IsAuthenticated, UserHasPerms(
+            perms_dict={
+                "default": {
+                    "PUT": ['base.change_resourcebase_metadata']
+                }
+            }
+        )]
+    )
+    def metadata(self, request, pk=None):
+        """
+        Endpoint to upload ISO metadata
+        Usage Example:
+
+        import requests
+
+        dataset_id = 1
+        url = f"http://localhost:8080/api/v2/datasets/{dataset_id}/metadata"
+        files=[
+            ('metadata_file',('metadata.xml',open('/home/user/metadata.xml','rb'),'text/xml'))
+        ]
+        headers = {
+            'Authorization': 'Basic dXNlcjpwYXNzd29yZA=='
+        }
+        response = requests.request("PUT", url, payload={}, files=files)
+
+        cURL example:
+        curl --location --request PUT 'http://localhost:8000/api/v2/datasets/{dataset_id}/metadata' \
+        --form 'metadata_file=@/home/user/metadata.xml'
+        """
+        out = {}
+        storage_manager = None
+        if not self.queryset.filter(id=pk).exists():
+            raise NotFound(detail=f"Dataset with ID {pk} is not available")
+        serializer = self.serializer_class(data=request.data)
+        if not serializer.is_valid(raise_exception=False):
+            raise InvalidDatasetException(detail=serializer.errors)
+        try:
+            data = serializer.data.copy()
+            if not data["metadata_file"]:
+                raise InvalidMetadataException(detail="A valid metadata file must be specified")
+            storage_manager = StorageManager(remote_files=data)
+            storage_manager.clone_remote_files()
+            file = storage_manager.get_retrieved_paths()
+            metadata_file = file["metadata_file"]
+            dataset = self.queryset.get(id=pk)
+            try:
+                dataset_uuid, vals, regions, keywords, _ = parse_metadata(
+                    open(metadata_file).read())
+            except Exception:
+                raise InvalidMetadataException(detail="Unsupported metadata format")
+            if dataset_uuid and dataset.uuid != dataset_uuid:
+                raise InvalidMetadataException(detail="The UUID identifier from the XML Metadata, is different from the one saved")
+            try:
+                updated_dataset = update_resource(dataset, metadata_file, regions, keywords, vals)
+                updated_dataset.save()  # This also triggers the recreation of the XML metadata file according to the updated values
+            except Exception:
+                raise GeneralDatasetException(detail="Failed to update metadata")
+            out['success'] = True
+            out['message'] = ['Metadata successfully updated']
+            return Response(out)
+        except Exception as e:
+            raise e
+        finally:
+            if storage_manager:
+                storage_manager.delete_retrieved_paths()
+
     @extend_schema(
         methods=["get"],
         responses={200: SimpleMapLayerSerializer(many=True)},