Migration deploy build config

.circleci/config.yml

@@ -42,7 +42,7 @@ jobs:
           path: cypress/screenshots
   docs-build:
     docker:
-      - image: python:3.8
+      - image: python:3.12
     steps:
       - checkout
       - dotenv/source:

.copilot/config.yml

@@ -0,0 +1,4 @@
+repository: emt
+builder:
+  name: paketobuildpacks/builder-jammy-full
+  version: 0.3.339

.copilot/image_build_run.sh

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+
+# Exit early if something goes wrong
+set -e
+
+# Add commands below to run inside the container after all the other buildpacks have been applied
+
+export DJANGO_SECRET_KEY="build-time-secret"
+export DEBUG="False"
+export FEATURE_ENFORCE_STAFF_SSO_ENABLED="False"
+
+python manage.py collectstatic --noinput

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.8
+FROM python:3.12
 
 RUN mkdir /usr/src/app
 WORKDIR /usr/src/app

Procfile

@@ -1,3 +1,3 @@
-web: ./web.sh
+web: bash web.sh
 celery: celery -A app worker -l info
 celerybeat: celery -A app beat -l info

app/pingdom/__init__.py

@@ -0,0 +1 @@
+"""Functionality for monitoring the uptime of the service."""

app/pingdom/services.py

@@ -0,0 +1,55 @@
+from django.conf import settings
+from django.db import DatabaseError
+from redis import Redis
+
+from app.enquiries.celery import app as celery_app
+from app.enquiries.models import Enquiry
+
+
+class CheckDatabase:
+    """Check the database is up and running."""
+
+    name = "database"
+
+    def check(self):
+        """Perform the check."""
+        try:
+            Enquiry.objects.exists()
+            return True, ""
+        except DatabaseError as exception:
+            return False, f"Pingdom check Database: {exception}"
+
+
+class CheckCelery:
+    name = "celery"
+
+    def check(self):
+        try:
+            insp = celery_app.control.inspect()
+            nodes = insp.stats()
+            if not nodes:
+                raise Exception("Celery is not running")
+            return True, ""
+        except Exception as exception:
+            return False, f"Pingdom check Celery: {exception}"
+
+
+class CheckRedis:
+    name = "redis"
+
+    def check(self):
+        try:
+            redis = Redis.from_url(settings.REDIS_BASE_URL)
+            if redis.ping():
+                return True, ""
+            else:
+                return False, "Redis is not connected"
+        except Exception as exception:
+            return False, f"Pingdom check Redis: {exception}"
+
+
+services_to_check = (
+    CheckDatabase,
+    CheckCelery,
+    CheckRedis,
+)

app/pingdom/tests/test_services.py

@@ -0,0 +1,75 @@
+from unittest.mock import patch
+
+from django.db import DatabaseError
+from django.test import TestCase
+from redis import RedisError
+
+from app.pingdom.services import CheckCelery, CheckDatabase, CheckRedis
+
+
+class PingdomServicesTestCase(TestCase):
+    def test_check_database_success(self):
+        check_database = CheckDatabase()
+        result = check_database.check()
+        assert check_database.name == "database"
+        assert result == (True, "")
+
+    def test_check_database_failure(self):
+        check_database = CheckDatabase()
+        with patch(
+            'app.enquiries.models.Enquiry.objects.exists',
+            side_effect=DatabaseError('No database'),
+        ):
+            result = check_database.check()
+
+        assert result == (False, "Pingdom check Database: No database")
+
+    def test_check_celery_success(self):
+        check_celery = CheckCelery()
+        with patch(
+            'app.enquiries.celery.app.control.inspect.stats',
+            return_value=[{}]
+        ):
+            result = check_celery.check()
+        assert check_celery.name == "celery"
+        assert result == (True, "")
+
+    def test_check_celery_failure(self):
+        check_celery = CheckCelery()
+        with patch(
+            'app.enquiries.celery.app.control.inspect.stats',
+            return_value=None
+        ):
+            result = check_celery.check()
+
+        assert result == (False, "Pingdom check Celery: Celery is not running")
+
+    def test_check_redis_success(self):
+        check_redis = CheckRedis()
+        with patch(
+            'redis.Redis.ping',
+            return_value=True,
+        ):
+            result = check_redis.check()
+        assert check_redis.name == "redis"
+        assert result == (True, "")
+
+    def test_check_redis_no_ping(self):
+        check_redis = CheckRedis()
+        with patch(
+            'redis.Redis.ping',
+            return_value=False,
+        ):
+            result = check_redis.check()
+        assert check_redis.name == "redis"
+        assert result == (False, "Redis is not connected")
+
+    def test_check_redis_failure(self):
+        check_redis = CheckRedis()
+        with patch(
+            'redis.Redis.ping',
+            side_effect=RedisError("Redis error"),
+        ):
+            result = check_redis.check()
+
+        assert result == (False, "Pingdom check Redis: Redis error")

app/pingdom/tests/test_views.py

@@ -0,0 +1,89 @@
+import pytest
+from unittest.mock import patch
+
+from app.enquiries.tests import test_dh_utils
+from django.db import DatabaseError
+
+from rest_framework import status
+from rest_framework.reverse import reverse
+from redis.exceptions import RedisError
+
+from app.pingdom.services import CheckCelery, CheckDatabase, CheckRedis
+from app.pingdom.views import ping
+
+
+pytestmark = pytest.mark.django_db
+
+
+class ServiceHealthCheckPingdomTestCase(test_dh_utils.DataHubUtilsTests):
+    @patch.object(CheckDatabase, 'check')
+    @patch.object(CheckCelery, 'check')
+    @patch.object(CheckRedis, 'check')
+    def test_ping_success(self, check_database, check_celery, check_redis):
+        check_database.return_value = (True, "")
+        check_celery.return_value = (True, "")
+        check_redis.return_value = (True, "")
+
+        response = ping({})
+
+        assert response.status_code == status.HTTP_200_OK
+        assert '<status>OK</status>' in str(response.content)
+        assert response.headers['content-type'] == 'text/xml'
+
+    @patch.object(CheckDatabase, 'check')
+    def test_ping_failure(self, check_database):
+        check_database.return_value = (False, "Error message")
+        response = ping({})
+
+        assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+        assert '<status>FALSE</status>' in str(response.content)
+        assert '<!--Error message-->' in str(response.content)
+        assert response.headers['content-type'] == 'text/xml'
+
+    @patch.object(CheckCelery, 'check')
+    def test_all_good(self, check_celery):
+        """Fake Celery for Circle CI"""
+        check_celery.return_value = (True, "")
+        url = reverse('ping')
+        response = self.client.get(url)
+
+        assert response.status_code == status.HTTP_200_OK
+        assert '<status>OK</status>' in str(response.content)
+        assert response.headers['content-type'] == 'text/xml'
+
+    def test_check_database_fail(self):
+        url = reverse('ping')
+
+        with patch(
+            'app.enquiries.models.Enquiry.objects.exists',
+            side_effect=DatabaseError('No database'),
+        ):
+            response = self.client.get(url)
+
+            assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+            assert '<status>FALSE</status>' in str(response.content)
+            assert response.headers['content-type'] == 'text/xml'
+
+    def test_check_celery_fail(self):
+        url = reverse('ping')
+        with patch(
+            'app.enquiries.celery.app.control.inspect.stats',
+            return_value=None
+        ):
+            response = self.client.get(url)
+
+            assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+            assert '<status>FALSE</status>' in str(response.content)
+            assert response.headers['content-type'] == 'text/xml'
+
+    def test_check_redis_fail(self):
+        url = reverse('ping')
+        with patch(
+            'redis.Redis.ping',
+            side_effect=RedisError("Redis error"),
+        ):
+            response = self.client.get(url)
+
+            assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+            assert '<status>FALSE</status>' in str(response.content)
+            assert response.headers['content-type'] == 'text/xml'

app/pingdom/urls.py

@@ -0,0 +1,8 @@
+from django.urls import path
+
+from .views import pingdom
+
+
+urlpatterns = [
+    path("ping.xml", pingdom, name="pingdom"),
+]

app/pingdom/views.py

@@ -0,0 +1,33 @@
+from django.http import HttpResponse
+from rest_framework import status
+
+from .services import services_to_check
+
+PINGDOM_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?>
+<pingdom_http_custom_check>
+    <status>{status}</status>
+</pingdom_http_custom_check>\n"""
+
+COMMENT_TEMPLATE = "<!--{comment}-->\n"
+
+
+def ping(request):
+    """Ping view."""
+    checked = {}
+    for service in services_to_check:
+        checked[service.name] = service().check()
+
+    if all(item[0] for item in checked.values()):
+        return HttpResponse(
+            PINGDOM_TEMPLATE.format(status="OK"),
+            content_type="text/xml",
+        )
+    else:
+        body = PINGDOM_TEMPLATE.format(status="FALSE")
+        for service_result in filter(lambda x: x[0] is False, checked.values()):
+            body += COMMENT_TEMPLATE.format(comment=service_result[1])
+        return HttpResponse(
+            body,
+            status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            content_type="text/xml",
+        )