Implement CRUD for azure storage

.env

@@ -28,3 +28,6 @@ DSO_CLIENT_ID=test
 DSO_CLIENT_SECRET=test
 DSO_AUTH_URL=https://
 DSO_API_URL=https://
+AZURE_ACCOUNT_NAME=devstoreaccount1
+AZURE_CONTAINER=devcontainer
+AZURE_CONNECTION_STRING=DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite:10000/devstoreaccount1;

app/Dockerfile

@@ -4,7 +4,7 @@ LABEL maintainer="[email protected]"
 RUN apt-get update && apt-get install -y
 RUN apt-get install libpq-dev -y
 RUN apt-get install gcc -y
-
+RUN apt install libmagic1 -y
 RUN pip install --upgrade pip
 RUN pip install uwsgi
 WORKDIR /app

app/apps/cases/admin.py

@@ -1,5 +1,5 @@
 from django.contrib import admin
-from .models import Case
+from .models import Case, CaseDocument
 
 
 @admin.register(Case)
@@ -13,3 +13,9 @@ class CaseAdmin(admin.ModelAdmin):
         "updated",
     )
     search_fields = ("id",)
+
+
+@admin.register(CaseDocument)
+class CaseDocumentAdmin(admin.ModelAdmin):
+    list_display = ("id", "case", "document", "name")
+    search_fields = ("id",)

app/apps/cases/migrations/0005_casedocument.py

@@ -0,0 +1,43 @@
+# Generated by Django 5.0.8 on 2024-10-28 09:21
+
+import apps.cases.models
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("cases", "0004_case_created_case_updated"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="CaseDocument",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("name", models.CharField(max_length=100)),
+                (
+                    "document",
+                    models.FileField(upload_to=apps.cases.models.get_upload_path),
+                ),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "case",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.PROTECT,
+                        related_name="documents",
+                        to="cases.case",
+                    ),
+                ),
+            ],
+        ),
+    ]

app/apps/cases/models.py

@@ -2,6 +2,9 @@
 from apps.homeownerassociation.models import HomeownerAssociation
 from apps.events.models import CaseEvent, ModelEventEmitter
 from enum import Enum
+import os
+from apps.events.models import CaseEvent, ModelEventEmitter
+from django.core.files.storage import default_storage
 
 
 class AdviceType(Enum):
@@ -45,3 +48,21 @@ def __str__(self):
 
     class Meta:
         ordering = ["name"]
+
+
+def get_upload_path(instance, filename):
+    return os.path.join("uploads", "cases", "%s" % instance.case.id, filename)
+
+
+class CaseDocument(models.Model):
+    name = models.CharField(max_length=100)
+    case = models.ForeignKey(Case, on_delete=models.PROTECT, related_name="documents")
+    document = models.FileField(upload_to=get_upload_path)
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    def __str__(self):
+        return f"Document: {self.id}"
+
+    def delete(self):
+        default_storage.delete(self.document.name)
+        return super().delete()

app/apps/cases/serializers.py

@@ -1,7 +1,10 @@
 from apps.homeownerassociation.serializers import ContactSerializer
 from apps.cases.models import Case
+from apps.cases.models import Case, CaseDocument
 from apps.workflow.serializers import CaseWorkflowSerializer
 from rest_framework import serializers
+import magic
+import os
 
 
 class CaseSerializer(serializers.ModelSerializer):
@@ -49,3 +52,26 @@ def get_homeowner_association(self, obj):
     class Meta:
         model = Case
         fields = ("id", "homeowner_association", "created")
+
+
+class CaseDocumentSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = CaseDocument
+        fields = ("id", "case", "document", "name")
+
+    def validate_document(self, value):
+        ext = os.path.splitext(value.name)[1].lower()
+        if ext not in [".pdf", ".docx", ".txt"]:
+            raise serializers.ValidationError("File extension not allowed")
+
+        mime = magic.Magic(mime=True)
+        file_mime_type = mime.from_buffer(value.read(2048))
+        value.seek(0)
+        if file_mime_type not in [
+            "application/pdf",
+            "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+            "text/plain",
+        ]:
+            raise serializers.ValidationError("File type not allowed")
+
+        return value

app/apps/cases/tests/tests_api.py

@@ -2,7 +2,7 @@
 from django.urls import reverse
 from rest_framework import status
 from rest_framework.test import APITestCase
-
+from django.core.files.uploadedfile import SimpleUploadedFile
 from utils.test_utils import get_authenticated_client, get_unauthenticated_client
 
 
@@ -11,6 +11,7 @@ def setUp(self):
         management.call_command("flush", verbosity=0, interactive=False)
         super().setUp()
         self.client = get_authenticated_client()
+        self.case = self._create_case()
 
     def test_unauthenticated_get(self):
         url = reverse("cases-list")
@@ -23,23 +24,112 @@ def test_authenticated_get(self):
         response = self.client.get(url)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
-    def test_authenticated_get_empty(self):
-        url = reverse("cases-list")
-        response = self.client.get(url)
-        data = response.json()
-
-        self.assertEqual(data, [])
-
     def test_retrieve_case(self):
         self._create_case()
-        case_id = 1
+        case_id = self.case
         url = reverse("cases-detail", args=[case_id])
         response = self.client.get(url)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
+    def test_create_document_success(self):
+        url = reverse("cases-create-document")
+        document_data = {
+            "document": SimpleUploadedFile(
+                "test_document.pdf", b"file_content", content_type="application/pdf"
+            ),
+            "case": self.case,
+            "name": "document_name",
+        }
+        response = self.client.post(url, data=document_data, format="multipart")
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertIn("id", response.data)
+
+    def test_create_document_blocked_file_extension(self):
+        url = reverse("cases-create-document")
+        document_data = {
+            "document": SimpleUploadedFile(
+                "test_document.exe", b"file_content", content_type="application/pdf"
+            ),
+            "case": self.case,
+            "name": "document_name",
+        }
+        response = self.client.post(url, data=document_data, format="multipart")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_create_document_invalid_case(self):
+        url = reverse("cases-create-document")
+        document_data = {
+            "document": SimpleUploadedFile(
+                "test_document.pdf", b"file_content", content_type="application/pdf"
+            ),
+            "case": 1231321,
+            "name": "document_name",
+        }
+        response = self.client.post(url, data=document_data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_create_document_invalid_data(self):
+        url = reverse("cases-create-document")
+        document_data = {}
+        response = self.client.post(url, data=document_data, format="json")
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_get_documents_success(self):
+        self._create_sample_document()
+        url = reverse("cases-get-documents", args=[self.case])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIsInstance(response.data, list)
+
+    def test_get_documents_empty(self):
+        url = reverse("cases-get-documents", args=[self.case])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data, [])
+
+    def test_download_document_success(self):
+        document = (
+            self._create_sample_document()
+        )  # Create and retrieve the document instance
+        url = reverse("cases-download-document", args=[self.case, document["id"]])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertIn("Content-Disposition", response)
+
+    def test_download_document_not_found(self):
+        non_existent_doc_id = 999
+        url = reverse("cases-download-document", args=[self.case, non_existent_doc_id])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_delete_document_success(self):
+        document = self._create_sample_document()
+        url = reverse("cases-delete-document", args=[self.case, document["id"]])
+        response = self.client.delete(url)
+        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+
+    def test_delete_document_not_found(self):
+        non_existent_doc_id = 999
+        url = reverse("cases-delete-document", args=[self.case, non_existent_doc_id])
+        response = self.client.delete(url)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def _create_sample_document(self):
+        url = reverse("cases-create-document")
+        document_data = {
+            "document": SimpleUploadedFile(
+                "test_document.pdf", b"file_content", content_type="application/pdf"
+            ),
+            "case": self.case,
+            "name": "document_name",
+        }
+        response = self.client.post(url, data=document_data, format="multipart")
+        return response.data
+
     def _create_case(self):
         url = reverse("cases-list")
         data = {"description": "Test case description"}
 
         response = self.client.post(url, data, format="json")
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        return response.data["id"]

app/apps/cases/views.py

@@ -9,6 +9,11 @@
 
 from .models import Case
 from .serializers import CaseCreateSerializer, CaseListSerializer, CaseSerializer
+from rest_framework import status
+from .models import Case, CaseDocument
+from .serializers import CaseCreateSerializer, CaseDocumentSerializer, CaseSerializer
+from django.shortcuts import get_object_or_404
+from django.core.files.storage import default_storage
 
 
 class CaseViewSet(
@@ -22,12 +27,15 @@ class CaseViewSet(
     serializer_class = CaseSerializer
 
     def get_serializer_class(self):
-        if self.action == "create":
+        if self.action == "create_document" or self.action == "get_documents":
+            return CaseDocumentSerializer
+        elif self.action == "create":
             return CaseCreateSerializer
         elif self.action == "list":
             return CaseListSerializer
         elif self.action == "events":
             return CaseEventSerializer
+
         return CaseSerializer
 
     @action(detail=True, methods=["get"], url_path="workflows")
@@ -46,3 +54,41 @@ def create(self, request):
         case = Case.objects.create(**validated_data)
         Contact.process_contacts(case, contacts_data)
         return Response(CaseSerializer(case).data, status=201)
+
+    @action(
+        detail=False, methods=["post"], url_path="documents", name="cases-documents"
+    )
+    def create_document(self, request):
+        serializer = CaseDocumentSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+    @action(detail=True, methods=["get"], url_path="documents")
+    def get_documents(self, request, pk=None):
+        case = self.get_object()
+        documents = CaseDocument.objects.filter(case=case)
+        serializer = CaseDocumentSerializer(documents, many=True)
+        return Response(serializer.data)
+
+    @action(
+        detail=True, methods=["get"], url_path="documents/download/(?P<doc_id>[^/.]+)"
+    )
+    def download_document(self, request, pk=None, doc_id=None):
+        case = self.get_object()
+        case_document = get_object_or_404(CaseDocument, case=case, id=doc_id)
+        with default_storage.open(case_document.document.name, "rb") as file:
+            response = Response(file.read(), content_type="application/octet-stream")
+            response["Content-Disposition"] = (
+                f'attachment; filename="{case_document.document.name}"'
+            )
+            return response
+
+    @action(detail=True, methods=["delete"], url_path="documents/(?P<doc_id>[^/.]+)")
+    def delete_document(self, request, pk=None, doc_id=None):
+        case = self.get_object()
+        case_document = get_object_or_404(CaseDocument, case=case, id=doc_id)
+        case_document.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)

app/apps/homeownerassociation/views.py

@@ -15,10 +15,14 @@ class HomeOwnerAssociationView(
     queryset = HomeownerAssociation.objects.all()
     serializer_class = HomeownerAssociationSerializer
 
+    def get_serializer_class(self):
+        if self.action == "cases":
+            return CaseListSerializer
+        return super().get_serializer_class()
+
     @action(detail=True, methods=["get"])
     def cases(self, request, pk=None):
         hoa = self.get_object()
         cases = Case.objects.filter(homeowner_association=hoa)
-        print(cases)
         serializer = CaseListSerializer(cases, many=True)
         return Response(serializer.data)