Translate report values (#2506)

This change:
- updates the sql query used to generate the report so that the notification_type and status are translated into the appropriate display values. These values should match what is in current report feature
- updates the date time in the report to be in eastern time, to match the current report feature
- some light refactoring in models.py so I can import the email and sms statuses into utils.py
- an integration test

Author: smcmurtry

app/models.py

@@ -73,6 +73,33 @@
 sms_sending_vehicles = db.Enum(*[vehicle.value for vehicle in SmsSendingVehicles], name="sms_sending_vehicles")
 
 
+EMAIL_STATUS_FORMATTED = {
+    "failed": "Failed",
+    "technical-failure": "Tech issue",
+    "temporary-failure": "Content or inbox issue",
+    "virus-scan-failed": "Attachment has virus",
+    "delivered": "Delivered",
+    "sending": "In transit",
+    "created": "In transit",
+    "sent": "Delivered",
+    "pending": "In transit",
+    "pending-virus-check": "In transit",
+    "pii-check-failed": "Exceeds Protected A",
+}
+
+SMS_STATUS_FORMATTED = {
+    "failed": "Failed",
+    "technical-failure": "Tech issue",
+    "temporary-failure": "Carrier issue",
+    "permanent-failure": "No such number",
+    "delivered": "Delivered",
+    "sending": "In transit",
+    "created": "In transit",
+    "pending": "In transit",
+    "sent": "Sent",
+}
+
+
 def filter_null_value_fields(obj):
     return dict(filter(lambda x: x[1] is not None, obj.items()))
 
@@ -1885,6 +1912,7 @@ def subject(self):
     def formatted_status(self):
         def _getStatusByBounceSubtype():
             """Return the status of a notification based on the bounce sub type"""
+            # note: if this function changes, update the report query in app/report/utils.py::build_notifications_query
             if self.feedback_subtype:
                 return {
                     "suppressed": "Blocked",
@@ -1895,6 +1923,7 @@ def _getStatusByBounceSubtype():
 
         def _get_sms_status_by_feedback_reason():
             """Return the status of a notification based on the feedback reason"""
+            # note: if this function changes, update the report query in app/report/utils.py::build_notifications_query
             if self.feedback_reason:
                 return {
                     "NO_ORIGINATION_IDENTITIES_FOUND": "Can't send to this international number",
@@ -1905,30 +1934,12 @@ def _get_sms_status_by_feedback_reason():
 
         return {
             "email": {
-                "failed": "Failed",
-                "technical-failure": "Tech issue",
-                "temporary-failure": "Content or inbox issue",
+                **EMAIL_STATUS_FORMATTED,
                 "permanent-failure": _getStatusByBounceSubtype(),
-                "virus-scan-failed": "Attachment has virus",
-                "delivered": "Delivered",
-                "sending": "In transit",
-                "created": "In transit",
-                "sent": "Delivered",
-                "pending": "In transit",
-                "pending-virus-check": "In transit",
-                "pii-check-failed": "Exceeds Protected A",
             },
             "sms": {
-                "failed": "Failed",
-                "technical-failure": "Tech issue",
-                "temporary-failure": "Carrier issue",
-                "permanent-failure": "No such number",
+                **SMS_STATUS_FORMATTED,
                 "provider-failure": _get_sms_status_by_feedback_reason(),
-                "delivered": "Delivered",
-                "sending": "In transit",
-                "created": "In transit",
-                "pending": "In transit",
-                "sent": "Sent",
             },
             "letter": {
                 "technical-failure": "Technical failure",

app/report/utils.py

@@ -1,9 +1,9 @@
-from sqlalchemy import func, text
+from sqlalchemy import case, func, text
 from sqlalchemy.orm import aliased
 
 from app import db
 from app.aws.s3 import stream_to_s3
-from app.models import Job, Notification, Template, User
+from app.models import EMAIL_STATUS_FORMATTED, SMS_STATUS_FORMATTED, Job, Notification, Template, User
 
 FR_TRANSLATIONS = {
     "Recipient": "Destinataire",
@@ -14,6 +14,23 @@
     "Job": "Tâche",
     "Status": "État",
     "Sent Time": "Heure d’envoi",
+    # notification types
+    "email": "courriel",
+    "sms": "sms",
+    # notification statuses
+    "Failed": "Échec",
+    "Tech issue": "Problème technique",
+    "Content or inbox issue": "Problème de contenu ou de boîte de réception",
+    "Attachment has virus": "La pièce jointe contient un virus",
+    "Delivered": "Livraison réussie",
+    "In transit": "Envoi en cours",
+    "Exceeds Protected A": "Niveau supérieur à Protégé A",
+    "Carrier issue": "Problème du fournisseur",
+    "No such number": "Numéro inexistant",
+    "Sent": "Envoyé",
+    "Blocked": "Message bloqué",
+    "No such address": "Adresse inexistante",
+    # "Can't send to this international number": "" # no translation exists for this yet
 }
 
 
@@ -51,19 +68,19 @@ def build_notifications_query(service_id, notification_type, language, days_limi
     j = aliased(Job)
     u = aliased(User)
 
-    translate = Translate(language).translate
-
-    # Build the query using SQLAlchemy
-    return (
+    # Build the inner subquery (returns enum values, cast as text for notification_type)
+    inner_query = (
         db.session.query(
-            n.to.label(translate("Recipient")),
-            t.name.label(translate("Template")),
-            n.notification_type.label(translate("Type")),
-            func.coalesce(u.name, "").label(translate("Sent by")),
-            func.coalesce(u.email_address, "").label(translate("Sent by email")),
-            func.coalesce(j.original_file_name, "").label(translate("Job")),
-            n.status.label(translate("Status")),
-            func.to_char(n.created_at, "YYYY-MM-DD HH24:MI:SS").label(translate("Sent Time")),
+            n.to.label("to"),
+            t.name.label("template_name"),
+            n.notification_type.cast(db.String).label("notification_type"),
+            u.name.label("user_name"),
+            u.email_address.label("user_email"),
+            j.original_file_name.label("job_name"),
+            n.status.label("status"),
+            n.created_at.label("created_at"),
+            n.feedback_subtype.label("feedback_subtype"),
+            n.feedback_reason.label("feedback_reason"),
         )
         .join(t, t.id == n.template_id)
         .outerjoin(j, j.id == n.job_id)
@@ -74,6 +91,63 @@ def build_notifications_query(service_id, notification_type, language, days_limi
             n.created_at > func.now() - text(f"interval '{days_limit} days'"),
         )
         .order_by(n.created_at.desc())
+        .subquery()
+    )
+
+    # Map statuses for translation
+    translate = Translate(language).translate
+    # Provider-failure logic for email
+    provider_failure_email = case(
+        [(inner_query.c.feedback_subtype.in_(["suppressed", "on-account-suppression-list"]), "Blocked")], else_="No such address"
+    )
+    # Provider-failure logic for sms
+    provider_failure_sms = case(
+        [
+            (
+                inner_query.c.feedback_reason.in_(["NO_ORIGINATION_IDENTITIES_FOUND", "DESTINATION_COUNTRY_BLOCKED"]),
+                "Can't send to this international number",
+            )
+        ],
+        else_="No such number",
+    )
+
+    email_status_cases = [(inner_query.c.status == k, translate(v)) for k, v in EMAIL_STATUS_FORMATTED.items()]
+    sms_status_cases = [(inner_query.c.status == k, translate(v)) for k, v in SMS_STATUS_FORMATTED.items()]
+    # Add provider-failure logic
+    if notification_type == "email":
+        email_status_cases.append((inner_query.c.status == "provider-failure", translate(provider_failure_email)))
+        status_expr = case(email_status_cases, else_=inner_query.c.status)
+    elif notification_type == "sms":
+        sms_status_cases.append((inner_query.c.status == "provider-failure", translate(provider_failure_sms)))
+        status_expr = case(sms_status_cases, else_=inner_query.c.status)
+    else:
+        status_expr = inner_query.c.status
+    if language == "fr":
+        status_expr = func.coalesce(func.nullif(status_expr, ""), "").label(translate("Status"))
+    else:
+        status_expr = status_expr.label(translate("Status"))
+
+    # Outer query: translate notification_type for display
+    notification_type_translated = case(
+        [
+            (inner_query.c.notification_type == "email", translate("email")),
+            (inner_query.c.notification_type == "sms", translate("sms")),
+        ],
+        else_=inner_query.c.notification_type,
+    ).label(translate("Type"))
+
+    return db.session.query(
+        inner_query.c.to.label(translate("Recipient")),
+        inner_query.c.template_name.label(translate("Template")),
+        notification_type_translated,
+        func.coalesce(inner_query.c.user_name, "").label(translate("Sent by")),
+        func.coalesce(inner_query.c.user_email, "").label(translate("Sent by email")),
+        func.coalesce(inner_query.c.job_name, "").label(translate("Job")),
+        status_expr,
+        # Explicitly cast created_at to UTC, then to America/Toronto
+        func.to_char(
+            func.timezone("America/Toronto", func.timezone("UTC", inner_query.c.created_at)), "YYYY-MM-DD HH24:MI:SS"
+        ).label(translate("Sent Time")),
     )
 
 

tests/app/report/test_utils.py

@@ -1,9 +1,14 @@
+import csv
+import io
+from datetime import datetime, timedelta
 from unittest.mock import patch
 
 from app.report.utils import (
     Translate,
+    build_notifications_query,
     generate_csv_from_notifications,
 )
+from tests.app.conftest import create_sample_email_template, create_sample_notification, create_sample_service
 
 
 def test_translate_en():
@@ -42,3 +47,62 @@ def test_calls_helper_functions_with_correct_parameters(self):
                     mock_build_query.assert_called_once_with(service_id, notification_type, language, days_limit)
                     mock_compile_query.assert_called_once_with("mock query")
                     mock_stream.assert_called_once_with("mock copy command", s3_bucket, s3_key)
+
+
+class TestNotificationReportIntegration:
+    def test_generate_csv_from_notifications_integration(self, notify_db, notify_db_session, sample_user):
+        service = create_sample_service(notify_db, notify_db_session, user=sample_user)
+        template = create_sample_email_template(notify_db, notify_db_session, service=service)
+
+        now = datetime.utcnow()
+        notification_data = [
+            {"to_field": "[email protected]", "status": "delivered", "personalisation": {"name": "User1"}, "created_at": now},
+            {
+                "to_field": "[email protected]",
+                "status": "failed",
+                "personalisation": {"name": "User2"},
+                "created_at": now - timedelta(minutes=1),
+            },
+        ]
+        for data in notification_data:
+            create_sample_notification(
+                notify_db,
+                notify_db_session,
+                service=service,
+                template=template,
+                to_field=data["to_field"],
+                status=data["status"],
+                personalisation=data["personalisation"],
+                created_at=data["created_at"],
+            )
+
+        # Patch stream_query_to_s3 to write to a buffer instead of S3
+        csv_buffer = io.StringIO()
+
+        def fake_stream_query_to_s3(copy_command, s3_bucket, s3_key):
+            # Actually run the query and write CSV to the buffer
+            query = build_notifications_query(str(service.id), "email", "en", 7)
+            result = query.all()
+            fieldnames = [col["name"] for col in query.column_descriptions]
+            writer = csv.DictWriter(csv_buffer, fieldnames=fieldnames)
+            writer.writeheader()
+            for row in result:
+                writer.writerow(dict(zip(fieldnames, row)))
+            csv_buffer.seek(0)
+
+        with patch("app.report.utils.stream_query_to_s3", fake_stream_query_to_s3):
+            generate_csv_from_notifications(
+                str(service.id),
+                "email",
+                "en",
+                days_limit=7,
+                s3_bucket="test-bucket",
+                s3_key="test-key.csv",
+            )
+
+        # Check the buffer content
+        rows = list(csv.DictReader(csv_buffer.getvalue().splitlines()))
+        assert len(rows) == 2
+        assert rows[0]["Recipient"] == "[email protected]"
+        assert rows[1]["Recipient"] == "[email protected]"
+        assert set(r["Status"] for r in rows) == {"Delivered", "Failed"}