Merge pull request #2302 from uktrade/uat

Prod release

api/applications/tests/test_finalise_application.py

@@ -1,6 +1,7 @@
 import pytest
 
 from datetime import datetime
+from django.db.utils import IntegrityError
 from django.urls import reverse
 from django.utils import timezone
 from parameterized import parameterized
@@ -791,3 +792,17 @@ def test_approve_quantity_greater_than_applied_for_failure(self):
             response_data,
             {"errors": {f"quantity-{self.good_on_application.id}": [strings.Licence.INVALID_QUANTITY_ERROR]}},
         )
+
+    def test_goods_unique_on_licence(self):
+
+        licence = StandardLicenceFactory(case=self.standard_application, status=LicenceStatus.DRAFT)
+
+        with pytest.raises(IntegrityError) as e:
+            GoodOnLicence.objects.create(
+                good=self.good_on_application, licence=licence, usage=0.0, quantity=5.0, value=100.00
+            )
+            GoodOnLicence.objects.create(
+                good=self.good_on_application, licence=licence, usage=0.0, quantity=5.0, value=100.00
+            )
+
+        self.assertIn(f"({str(licence.id)}, {str(self.good_on_application.id)}) already exists", e.value.args[0])

api/conf/settings.py

@@ -124,8 +124,12 @@
     "api.external_data",
     "api.support",
     "health_check",
+    "health_check.cache",
     "health_check.contrib.celery",
     "health_check.contrib.celery_ping",
+    "health_check.db",
+    "health_check.contrib.migrations",
+    "health_check.storage",
     "django_audit_log_middleware",
     "lite_routing",
     "api.appeals",
@@ -137,14 +141,6 @@
     "drf_spectacular",
 ]
 
-if not IS_ENV_DBT_PLATFORM:
-    INSTALLED_APPS += [
-        "health_check.db",
-        "health_check.cache",
-        "health_check.storage",
-        "health_check.contrib.migrations",
-    ]
-
 MOCK_VIRUS_SCAN_ACTIVATE_ENDPOINTS = env("MOCK_VIRUS_SCAN_ACTIVATE_ENDPOINTS")
 
 if MOCK_VIRUS_SCAN_ACTIVATE_ENDPOINTS:

api/conf/settings_test.py

@@ -20,3 +20,9 @@
 DB_ANONYMISER_AWS_REGION = "eu-west-2"
 DB_ANONYMISER_AWS_STORAGE_BUCKET_NAME = "anonymiser-bucket"
 DB_ANONYMISER_AWS_ENDPOINT_URL = None
+
+try:
+    INSTALLED_APPS.remove("silk")
+    MIDDLEWARE.remove("silk.middleware.SilkyMiddleware")
+except ValueError:
+    pass

api/conf/urls.py

@@ -5,11 +5,12 @@
 from django.conf import settings
 
 import api.core.views
-from api.healthcheck.views import HealthCheckPingdomView
+from api.healthcheck.views import HealthCheckPingdomView, ServiceAvailableHealthCheckView
 
 urlpatterns = [
     path("healthcheck/", include("health_check.urls")),
     path("pingdom/ping.xml", HealthCheckPingdomView.as_view(), name="healthcheck-pingdom"),
+    path("service-available-check/", ServiceAvailableHealthCheckView.as_view(), name="service-available-check"),
     path("applications/", include("api.applications.urls")),
     path("assessments/", include("api.assessments.urls")),
     path("audit-trail/", include("api.audit_trail.urls")),

api/data_workspace/metadata/routers.py

@@ -0,0 +1,145 @@
+import datetime
+import typing
+
+from rest_framework import serializers
+from rest_framework.response import Response
+from rest_framework.reverse import reverse
+from rest_framework.routers import DefaultRouter
+from rest_framework.views import APIView
+
+from django.urls import (
+    NoReverseMatch,
+    path,
+)
+
+
+class TableMetadataView(APIView):
+    _ignore_model_permissions = True
+    schema = None  # exclude from schema
+    metadata = None
+
+    def get(self, request, *args, **kwargs):
+        tables = []
+        namespace = request.resolver_match.namespace
+        for table_metadata in self.metadata:
+            url_name = table_metadata["endpoint"]
+            if namespace:
+                url_name = f"{namespace}:{url_name}"
+            try:
+                url = reverse(
+                    url_name,
+                    args=args,
+                    kwargs=kwargs,
+                    request=request,
+                )
+            except NoReverseMatch:
+                # Don't bail out if eg. no list routes exist, only detail routes.
+                continue
+
+            tables.append(
+                {
+                    "table_name": table_metadata["table_name"],
+                    "endpoint": url,
+                    "indexes": table_metadata["indexes"],
+                    "fields": table_metadata["fields"],
+                }
+            )
+        return Response({"tables": tables})
+
+
+def is_optional(field):
+    return typing.get_origin(field) is typing.Union and type(None) in typing.get_args(field)
+
+
+def get_fields(view):
+    try:
+        serializer = view.get_serializer()
+    except AttributeError:
+        return []
+
+    primary_key_field = getattr(view.DataWorkspace, "primary_key", "id")
+
+    fields = []
+    for field in serializer.fields.values():
+        if isinstance(field, serializers.HiddenField):
+            continue
+
+        field_metadata = {"name": field.field_name}
+        if field.field_name == primary_key_field:
+            field_metadata["primary_key"] = True
+
+        if isinstance(field, serializers.UUIDField):
+            field_metadata["type"] = "UUID"
+            if field.allow_null:
+                field_metadata["nullable"] = True
+
+        elif isinstance(field, serializers.CharField):
+            field_metadata["type"] = "String"
+            if field.allow_null:
+                field_metadata["nullable"] = True
+
+        elif isinstance(field, serializers.SerializerMethodField):
+            method = getattr(field.parent, field.method_name)
+            return_type = method.__annotations__["return"]
+
+            if is_optional(return_type):
+                field_metadata["nullable"] = True
+                return_type, _ = typing.get_args(return_type)
+
+            if return_type is str:
+                field_metadata["type"] = "String"
+            elif return_type is datetime.datetime:
+                field_metadata["type"] = "DateTime"
+            else:  # pragma: no cover
+                raise NotImplementedError(
+                    f"Return type of {return_type} for {serializer.__class__.__name__}.{field.method_name} not handled"
+                )
+
+        else:  # pragma: no cover
+            raise NotImplementedError(f"Annotation not found for {field}")
+
+        fields.append(field_metadata)
+    return fields
+
+
+class TableMetadataRouter(DefaultRouter):
+    def register(self, viewset):
+        if not hasattr(viewset, "DataWorkspace"):  # pragma: no cover
+            raise NotImplementedError(f"No DataWorkspace configuration found for {viewset}")
+
+        prefix = viewset.DataWorkspace.table_name.replace("_", "-")
+        basename = f"dw-{prefix}"
+
+        super().register(prefix, viewset, basename)
+
+    def get_metadata_view(self, urls):
+        metadata = []
+        list_name = self.routes[0].name
+        for _, viewset, basename in self.registry:
+            data_workspace_metadata = viewset.DataWorkspace
+
+            view = viewset()
+            view.args = ()
+            view.kwargs = {}
+            view.format_kwarg = {}
+            view.request = None
+
+            metadata.append(
+                {
+                    "table_name": data_workspace_metadata.table_name,
+                    "endpoint": list_name.format(basename=basename),
+                    "indexes": getattr(data_workspace_metadata, "indexes", []),
+                    "fields": getattr(data_workspace_metadata, "fields", get_fields(view)),
+                }
+            )
+
+        return TableMetadataView.as_view(metadata=metadata)
+
+    def get_urls(self):
+        urls = super().get_urls()
+
+        view = self.get_metadata_view(urls)
+        metadata_url = path("table-metadata/", view, name="table-metadata")
+        urls.append(metadata_url)
+
+        return urls

api/data_workspace/metadata/tests/auto_field_urls.py

@@ -0,0 +1,22 @@
+from django.urls import (
+    include,
+    path,
+)
+
+from ..routers import TableMetadataRouter
+
+from . import views
+
+
+test_router = TableMetadataRouter()
+
+test_router.register(views.HiddenFieldViewSet)
+test_router.register(views.UUIDFieldViewSet)
+test_router.register(views.CharFieldViewSet)
+test_router.register(views.SerializerMethodFieldViewSet)
+test_router.register(views.AutoPrimaryKeyViewSet)
+test_router.register(views.ExplicitPrimaryKeyViewSet)
+
+urlpatterns = [
+    path("endpoints/", include(test_router.urls)),
+]

api/data_workspace/metadata/tests/serializers.py

@@ -0,0 +1,48 @@
+import datetime
+
+from typing import Optional
+
+from rest_framework import serializers
+
+
+class HiddenFieldSerializer(serializers.Serializer):
+    hidden_field = serializers.HiddenField(default="")
+
+
+class UUIDFieldSerializer(serializers.Serializer):
+    uuid_field = serializers.UUIDField()
+    nullable_uuid_field = serializers.UUIDField(allow_null=True)
+
+
+class CharFieldSerializer(serializers.Serializer):
+    char_field = serializers.CharField()
+    nullable_char_field = serializers.CharField(allow_null=True)
+
+
+class SerializerMethodFieldSerializer(serializers.Serializer):
+    returns_string = serializers.SerializerMethodField()
+    returns_optional_string = serializers.SerializerMethodField()
+    returns_datetime = serializers.SerializerMethodField()
+    returns_optional_datetime = serializers.SerializerMethodField()
+
+    def get_returns_string(self, instance) -> str:
+        return "string"
+
+    def get_returns_optional_string(self, instance) -> Optional[str]:
+        return None
+
+    def get_returns_datetime(self, instance) -> datetime.datetime:
+        return datetime.datetime.now()
+
+    def get_returns_optional_datetime(self, instance) -> Optional[datetime.datetime]:
+        return None
+
+
+class AutoPrimaryKeySerializer(serializers.Serializer):
+    id = serializers.UUIDField()
+    not_a_primary_key = serializers.UUIDField()
+
+
+class ExplicitPrimaryKeySerializer(serializers.Serializer):
+    a_different_id = serializers.UUIDField()
+    not_a_primary_key = serializers.UUIDField()