diff --git a/docker/integration-tests/test_stack.py b/docker/integration-tests/test_integration_qgis.py similarity index 100% rename from docker/integration-tests/test_stack.py rename to docker/integration-tests/test_integration_qgis.py diff --git a/src/django_oapif/decorators.py b/src/django_oapif/decorators.py index 36e3213e..d8e60178 100644 --- a/src/django_oapif/decorators.py +++ b/src/django_oapif/decorators.py @@ -1,9 +1,10 @@ from typing import Any, Callable, Dict, Optional from django.db.models import Model -from rest_framework import viewsets +from rest_framework import response, viewsets from rest_framework_gis.serializers import GeoFeatureModelSerializer +from django_oapif.metadata import OAPIFMetadata from django_oapif.mixins import OAPIFDescribeModelViewSetMixin from django_oapif.urls import oapif_router @@ -76,6 +77,15 @@ class Viewset(OAPIFDescribeModelViewSetMixin, viewsets.ModelViewSet): # Allowing '.' and '-' in urls lookup_value_regex = r"[\w.-]+" + # Metadata + metadata_class = OAPIFMetadata + + def options(self, request, *args, **kwargs) -> response.Response: + allowed_actions = self.metadata_class().determine_actions(request, self) + allowed_actions = ", ".join(allowed_actions.keys()) + data = self.metadata_class().determine_metadata(request, self) + return response.Response(data, headers={"Allow": allowed_actions}) + # ON HOLD, WAITING ON GeoFeatureModelSerializer to admit of null geometries """ # Apply custom serializer attributes diff --git a/src/django_oapif/metadata.py b/src/django_oapif/metadata.py new file mode 100644 index 00000000..ce14a6b9 --- /dev/null +++ b/src/django_oapif/metadata.py @@ -0,0 +1,27 @@ +from django.core.exceptions import PermissionDenied +from django.http import Http404 +from rest_framework import exceptions, metadata +from rest_framework.request import clone_request + + +class OAPIFMetadata(metadata.SimpleMetadata): + def determine_actions(self, request, view) -> dict: + actions = {} + for method in set(view.allowed_methods): + view.request = clone_request(request, method) + try: + # Test global permissions + if hasattr(view, "check_permissions"): + view.check_permissions(view.request) + # Test object permissions + if hasattr(view, "get_object") and method in {"PUT", "PATCH", "DELETE"}: + view.get_object() + except (exceptions.APIException, PermissionDenied, Http404): + pass + else: + serializer = view.get_serializer() + actions[method] = self.get_serializer_info(serializer) + finally: + view.request = request + + return actions diff --git a/src/signalo/core/tests.py b/src/signalo/core/tests.py index 3a1ca3f5..05094227 100644 --- a/src/signalo/core/tests.py +++ b/src/signalo/core/tests.py @@ -15,6 +15,8 @@ logger = logging.getLogger(__name__) +collections_url = "/oapif/collections" + def is_dense_partial_order(sorted_it: Iterable[int]) -> bool: prev = 0 @@ -111,7 +113,7 @@ def serialize_with_profile( class SpeedTestSerialization(APITestCase): @classmethod - def setUpClass(cls): + def setUpTestData(cls): call_command("populate_vl") call_command("populate_signs_poles", magnitude=30) cls.poles = Pole.objects.all() @@ -142,34 +144,45 @@ def setUpTestData(cls): cls.demo_viewer = User.objects.get(username="demo_viewer") cls.admin = User.objects.get(username="admin") + cls.collection_url = collections_url + "/signalo_core.pole" + cls.items_url = cls.collection_url + "/items" def tearDown(self): self.client.force_authenticate(user=None) - def test_auth_as_demo_viewer(self): + def test_get_as_viewer(self): collections_from_anonymous = self.client.get( - "/oapif/collections", format="json" + collections_url, format="json" ).json() self.client.force_authenticate(user=self.demo_viewer) - collection_response = self.client.get("/oapif/collections", format="json") - oapif_response_code = self.client.get("/oapif/").status_code + collection_response = self.client.get(collections_url, format="json") - self.assertEqual(oapif_response_code, 200) self.assertEqual(collection_response.status_code, 200) self.assertEqual( len(collection_response.json()), len(collections_from_anonymous) ) - def test_auth_as_admin(self): - collections_from_anonymous = self.client.get( - "/oapif/collections", format="json" - ).json() + def test_post_as_admin(self): self.client.force_authenticate(user=self.admin) - collection_response = self.client.get("/oapif/collections", format="json") - oapif_response_code = self.client.get("/oapif/").status_code - - self.assertEqual(oapif_response_code, 200) - self.assertEqual(collection_response.status_code, 200) - self.assertEqual( - len(collection_response.json()), len(collections_from_anonymous) - ) + data = {"geom": "Point(1300000 600000)", "name": "test123"} + post_to_items = self.client.post(self.items_url, data, format="json") + self.assertIn(post_to_items.status_code, (200, 201)) + + def test_anonymous_items_options(self): + # Anonymous client + expected = {"GET", "OPTIONS", "HEAD"} + response = self.client.options(self.items_url) + allowed_headers = map(lambda x: x.strip(), response.headers["Allow"].split(",")) + allowed_body = response.json()["actions"].keys() + self.assertEqual(set(allowed_body), expected) + self.assertEqual(set(allowed_headers), set(allowed_body)) + + def test_admin_items_options(self): + # Authenticated client with write permissions + expected = {"POST", "GET", "OPTIONS", "HEAD"} + self.client.force_authenticate(user=self.admin) + response = self.client.options(self.items_url) + allowed_headers = map(lambda x: x.strip(), response.headers["Allow"].split(",")) + allowed_body = response.json()["actions"].keys() + self.assertEqual(set(allowed_body), expected) + self.assertEqual(set(allowed_headers), set(allowed_body))