Release workflow feat test

backend/requirements/base.txt

@@ -26,8 +26,8 @@ django-storages==1.14
 django-tenants==3.6.1
 djangorestframework==3.15.2
 flake8==6.1.0
-
 flower==2.0.1
+GitPython==3.1.43
 loguru==0.7.2
 opentelemetry-api==1.22.0
 opentelemetry-distro==0.43b0

backend/src/zango/api/platform/releases/v1/serializers.py

@@ -0,0 +1,24 @@
+from rest_framework import serializers
+
+from zango.apps.release.models import AppRelease
+from zango.core.utils import get_datetime_str_in_tenant_timezone
+
+
+class AppReleaseSerializer(serializers.ModelSerializer):
+    created_at = serializers.SerializerMethodField()
+
+    class Meta:
+        model = AppRelease
+        fields = [
+            "id",
+            "version",
+            "description",
+            "status",
+            "last_git_hash",
+            "created_at",
+        ]
+
+    def get_created_at(self, obj):
+        return get_datetime_str_in_tenant_timezone(
+            obj.created_at, self.context["tenant"]
+        )

backend/src/zango/api/platform/releases/v1/urls.py

@@ -0,0 +1,8 @@
+from django.urls import path
+
+from .views import AppReleaseView
+
+
+urlpatterns = [
+    path("", AppReleaseView.as_view()),
+]

backend/src/zango/api/platform/releases/v1/views.py

@@ -0,0 +1,156 @@
+import json
+
+from datetime import datetime
+
+import pytz
+
+from django.conf import settings
+from django.db.models import Q
+from django.utils.decorators import method_decorator
+
+from zango.apps.release.models import AppRelease
+from zango.apps.shared.tenancy.models import TenantModel
+from zango.core.api import ZangoGenericPlatformAPIView, get_api_response
+from zango.core.api.utils import ZangoAPIPagination
+from zango.core.common_utils import set_app_schema_path
+from zango.core.utils import get_search_columns
+
+from .serializers import AppReleaseSerializer
+
+
+@method_decorator(set_app_schema_path, name="dispatch")
+class AppReleaseView(ZangoGenericPlatformAPIView, ZangoAPIPagination):
+    pagination_class = ZangoAPIPagination
+
+    def process_timestamp(self, timestamp, timezone):
+        try:
+            ts = json.loads(timestamp)
+            if not timezone:
+                timezone = settings.TIME_ZONE
+            tz = pytz.timezone(timezone)
+            ts["start"] = tz.localize(
+                datetime.strptime(ts["start"] + "-" + "00:00", "%Y-%m-%d-%H:%M"),
+                is_dst=None,
+            )
+            ts["end"] = tz.localize(
+                datetime.strptime(ts["end"] + "-" + "23:59", "%Y-%m-%d-%H:%M"),
+                is_dst=None,
+            )
+            return ts
+        except Exception:
+            import traceback
+
+            traceback.print_exc()
+            return None
+
+    def process_id(self, id):
+        try:
+            return int(id)
+        except ValueError:
+            return None
+
+    def get_queryset(self, search, tenant, columns={}):
+        field_name_query_mapping = {
+            "version": "version__icontains",
+            "description": "description__icontains",
+            "status": "status",
+            "id": "id__icontains",
+        }
+        search_filters = {
+            "id": self.process_id,
+            "Release_date": self.process_timestamp,
+        }
+        if columns.get("status") == "" or columns.get("status") is None:
+            field_name_query_mapping.pop("status")
+        records = AppRelease.objects.all().order_by("-id")
+        if search == "" and columns == {}:
+            return records
+        filters = Q()
+        for field_name, query in field_name_query_mapping.items():
+            if search:
+                if search_filters.get(field_name, None):
+                    value = search_filters[field_name](search)
+                    if value is not None:
+                        filters |= Q(**{query: value})
+                else:
+                    filters |= Q(**{query: search})
+        records = records.filter(filters).distinct()
+        if columns.get("Release_date"):
+            processed = self.process_timestamp(
+                columns.get("Release_date"), tenant.timezone
+            )
+            if processed is not None:
+                records = records.filter(
+                    created_at__gte=processed["start"], created_at__lte=processed["end"]
+                )
+        if columns.get("status"):
+            records = records.filter(status=columns["status"])
+        if columns.get("version"):
+            records = records.filter(version=columns["version"])
+        return records
+
+    def get_dropdown_options(self):
+        options = {}
+        options["version"] = [
+            {
+                "id": version,
+                "label": version,
+            }
+            for version in AppRelease.objects.values_list(
+                "version", flat=True
+            ).distinct()
+        ]
+        options["status"] = [
+            {
+                "id": "initiated",
+                "label": "Initiated",
+            },
+            {
+                "id": "in_progress",
+                "label": "In Progress",
+            },
+            {
+                "id": "released",
+                "label": "Released",
+            },
+            {
+                "id": "failed",
+                "label": "Failed",
+            },
+            {
+                "id": "archived",
+                "label": "Archived",
+            },
+        ]
+        return options
+
+    def get(self, request, app_uuid, *args, **kwargs):
+        try:
+            tenant = TenantModel.objects.get(uuid=app_uuid)
+            include_dropdown_options = request.GET.get("include_dropdown_options")
+            search = request.GET.get("search", None)
+            columns = get_search_columns(request)
+            app_releases = self.get_queryset(search, tenant, columns)
+            paginated_releases = self.paginate_queryset(
+                app_releases, request, view=self
+            )
+            serializer = AppReleaseSerializer(
+                paginated_releases, many=True, context={"tenant": tenant}
+            )
+            paginated_app_releases = self.get_paginated_response_data(serializer.data)
+            success = True
+            response = {
+                "releases": paginated_app_releases,
+                "message": "All app releases fetched successfully",
+            }
+            if include_dropdown_options:
+                response["dropdown_options"] = self.get_dropdown_options()
+            status = 200
+        except Exception as e:
+            import traceback
+
+            traceback.print_exc()
+            success = False
+            response = {"message": str(e)}
+            status = 500
+        return get_api_response(success, response, status)

backend/src/zango/api/platform/tenancy/v1/serializers.py

@@ -39,6 +39,18 @@ def get_date_format_display(self, obj):
         return obj.get_date_format_display()
 
     def update(self, instance, validated_data):
+        request = self.context["request"]
+        extra_config_str = request.data.get("extra_config")
+        # Convert extra_config from string to JSON if it exists
+        if extra_config_str:
+            try:
+                extra_config_json = json.loads(extra_config_str)
+                validated_data["extra_config"] = extra_config_json
+            except json.JSONDecodeError:
+                raise serializers.ValidationError(
+                    {"extra_config": "Invalid JSON format"}
+                )
+
         instance = super(TenantSerializerModel, self).update(instance, validated_data)
         request = self.context["request"]
         domains = request.data.getlist("domains")

backend/src/zango/api/platform/tenancy/v1/urls.py

@@ -5,6 +5,7 @@
 from zango.api.platform.codeassist.v1 import urls as codeassist_v1_urls
 from zango.api.platform.packages.v1 import urls as packages_v1_urls
 from zango.api.platform.permissions.v1 import urls as permissions_v1_urls
+from zango.api.platform.releases.v1 import urls as releases_v1_urls
 from zango.api.platform.tasks.v1 import urls as tasks_v1_urls
 
 from .views import (
@@ -61,5 +62,6 @@
     re_path(r"^(?P<app_uuid>[\w-]+)/code-assist/", include(codeassist_v1_urls)),
     re_path(r"^(?P<app_uuid>[\w-]+)/auditlog/", include(auditlog_v1_urls)),
     re_path(r"^(?P<app_uuid>[\w-]+)/access-logs/", include(accesslog_v1_urls)),
+    re_path(r"^(?P<app_uuid>[\w-]+)/releases/", include(releases_v1_urls)),
     path("", include(permissions_v1_urls)),
 ]

backend/src/zango/api/platform/tenancy/v1/utils.py

@@ -0,0 +1,63 @@
+import json
+import os
+import zipfile
+
+
+def extract_app_details_from_zip(template_zip):
+    settings_filename = "settings.json"
+    try:
+        with zipfile.ZipFile(template_zip, "r") as zip_file:
+            # Check if the settings file exists in the zip
+            zip_name = str(template_zip).split(".")[0]
+
+            # List all files in the zip
+            all_files = zip_file.namelist()
+
+            # Find the settings file path
+            settings_path = None
+            for file_path in all_files:
+                if file_path == os.path.join(zip_name, settings_filename):
+                    settings_path = file_path
+                    break
+
+            if not settings_path:
+                raise FileNotFoundError(
+                    f"{settings_filename} not found in the zip file."
+                )
+
+            # Read the contents of the settings file
+            with zip_file.open(settings_path) as settings_file:
+                settings_content = settings_file.read()
+
+            # Parse the JSON content
+            settings = json.loads(settings_content)
+            return (
+                settings["version"],
+                settings["app_name"],
+            )
+
+    except zipfile.BadZipFile:
+        raise Exception(f"Error: {template_zip} is not a valid zip file.")
+    except json.JSONDecodeError:
+        raise Exception(f"Error: {settings_filename} is not a valid JSON file.")
+    except KeyError:
+        raise Exception(f"Error: version or app_name not found in {settings_filename}.")
+    except Exception as e:
+        raise Exception(f"An error occurred: {str(e)}")
+
+
+def extract_zip_to_temp_dir(app_template):
+    # Create a temporary directory
+    temp_dir = os.getcwd()
+
+    try:
+        # Extract the zip file to the temporary directory
+        with zipfile.ZipFile(app_template, "r") as zip_ref:
+            zip_ref.extractall(temp_dir)
+
+        # Return the path of the temporary directory
+        return temp_dir
+    except Exception as e:
+        # If an error occurs, remove the temporary directory and re-raise the exception
+        os.rmdir(temp_dir)
+        raise e