Merge branch 'main' into task/add-translations

app/main/forms.py

@@ -553,7 +553,11 @@ def __init__(self, *args, **kwargs):
 
 
 class SendingDomainForm(StripWhitespaceForm):
-    sending_domain = StringField(_l("Sending domain"), validators=[])
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.sending_domain.choices = kwargs["sending_domain_choices"]
+
+    sending_domain = SelectField(_l("Sending domain"), validators=[])
 
 
 class RenameOrganisationForm(StripWhitespaceForm):

app/main/views/service_settings.py

@@ -72,6 +72,7 @@
     email_safe,
     get_logo_cdn_domain,
     get_new_default_reply_to_address,
+    get_verified_ses_domains,
     user_has_permissions,
     user_is_gov_user,
     user_is_platform_admin,
@@ -549,7 +550,7 @@ def service_set_reply_to_email(service_id):
 @main.route("/services/<service_id>/service-settings/sending-domain", methods=["GET", "POST"])
 @user_is_platform_admin
 def service_sending_domain(service_id):
-    form = SendingDomainForm()
+    form = SendingDomainForm(sending_domain_choices=get_verified_ses_domains())
 
     if request.method == "GET":
         form.sending_domain.data = current_service.sending_domain

app/notify_client/notification_counts_client.py

@@ -1,12 +1,11 @@
-from datetime import datetime
-
 from notifications_utils.clients.redis import (
     email_daily_count_cache_key,
     sms_daily_count_cache_key,
 )
 
 from app import redis_client, service_api_client, template_statistics_client
 from app.models.service import Service
+from app.utils import get_current_financial_year
 
 
 class NotificationCounts:
@@ -83,8 +82,10 @@ def get_limit_stats(self, service: Service):
                 }
         """
 
+        current_financial_year = get_current_financial_year()
         sent_today = self.get_all_notification_counts_for_today(service.id)
-        sent_thisyear = self.get_all_notification_counts_for_year(service.id, datetime.now().year)
+        # We are interested in getting data for the financial year, not the calendar year
+        sent_thisyear = self.get_all_notification_counts_for_year(service.id, current_financial_year)
 
         limit_stats = {
             "email": {

app/templates/views/templates/_template.html

@@ -27,6 +27,10 @@ <h2 class="heading-medium">{{ heading }}</h2>
           <a href="{{ url_for('.send_test', service_id=current_service.id, template_id=template.id) }}" class="button button-secondary">{{ _('No, send yourself this message') }}</a>
         </div>
       {% endif %}
+    {% elif current_user.platform_admin %}
+    <p class="hint">
+      {{ _("You cannot send messages from this service. Only team members can send messages.") }}
+    </p>
     {% endif %}
   {% endif %}
 </div>

app/translations/csv/fr.csv

@@ -2036,4 +2036,5 @@
 "<strong>{}</strong> can only send <strong>{}</strong> more {} until annual limit resets","<strong>{}</strong> ne peut envoyer que <strong>{}</strong> {} d'ici la réinitialisation de la limite annuelle"
 "To send to recipients you removed, wait until <strong>April 1, {}</strong> or contact them some other way.","Pour envoyer des messages aux destinataires, attendez <strong> le 1er avril {}</strong> ou optez pour une autre méthode de contact."
 "For more information, visit <a href={}>usage reports</a>.","Pour plus de détails, <a href={}>consulter les rapports d’utilisation</a>."
-"{} cannot send any more {} until April 1, {}","{} ne peut plus envoyer de {} d’ici le 1<sup>er</sup> avril {}."
+"{} cannot send any more {} until April 1, {}","{} ne peut plus envoyer de {} d’ici le 1<sup>er</sup> avril {}."
+"You cannot send messages from this service. Only team members can send messages.","Vous ne pouvez pas envoyer de messages à partir de ce service. Seuls les membres de l’équipe peuvent envoyer des messages."

app/utils.py

@@ -905,3 +905,24 @@ def get_limit_reset_time_et() -> dict[str, str]:
 
     limit_reset_time_et = {"en": next_midnight_utc_in_et.strftime("%-I%p"), "fr": next_midnight_utc_in_et.strftime("%H")}
     return limit_reset_time_et
+
+
+@cache.memoize(timeout=5 * 60)
+def get_verified_ses_domains():
+    """Query AWS SES for verified domain identities"""
+    ses = boto3.client("ses", region_name="ca-central-1")
+
+    # Get all identities
+    identities = ses.list_identities(
+        IdentityType="Domain"  # Only get domains, not email addresses
+    )
+
+    # Get verification status for each domain
+    verification_attrs = ses.get_identity_verification_attributes(Identities=identities["Identities"])["VerificationAttributes"]
+
+    # Format results
+    domains = [
+        domain for domain in identities["Identities"] if verification_attrs.get(domain, {}).get("VerificationStatus") == "Success"
+    ]
+
+    return domains

tests/app/main/views/test_service_settings.py

@@ -4377,3 +4377,69 @@ def test_should_suspend_service_callback_api(self, client_request, platform_admi
             _data={"updated_by_id": platform_admin_user["id"], "suspend_unsuspend": False},
             _expected_status=200,
         )
+
+
+class TestSendingDomain:
+    def test_sending_domain_page_shows_dropdown_of_verified_domains(
+        self, client_request, platform_admin_user, mock_get_service_settings_page_common, mocker
+    ):
+        client_request.login(platform_admin_user)
+        mock_get_domains = mocker.patch(
+            "app.main.views.service_settings.get_verified_ses_domains", return_value=["domain1.com", "domain2.com"]
+        )
+
+        page = client_request.get("main.service_sending_domain", service_id=SERVICE_ONE_ID, _test_page_title=False)
+
+        assert [option["value"] for option in page.select("select[name=sending_domain] option")] == ["domain1.com", "domain2.com"]
+
+        mock_get_domains.assert_called_once()
+
+    def test_sending_domain_page_populates_with_current_domain(
+        self, client_request, platform_admin_user, mock_get_service_settings_page_common, mocker, service_one
+    ):
+        service_one["sending_domain"] = "domain1.com"
+        mocker.patch("app.main.views.service_settings.get_verified_ses_domains", return_value=["domain1.com", "domain2.com"])
+
+        client_request.login(platform_admin_user)
+        page = client_request.get("main.service_sending_domain", service_id=SERVICE_ONE_ID, _test_page_title=False)
+
+        assert page.select_one("select[name=sending_domain] option[selected]")["value"] == "domain1.com"
+
+    def test_sending_domain_page_updates_domain_and_redirects_when_posted(
+        self, client_request, platform_admin_user, mock_get_service_settings_page_common, mocker, service_one, mock_update_service
+    ):
+        mocker.patch("app.main.views.service_settings.get_verified_ses_domains", return_value=["domain1.com", "domain2.com"])
+
+        client_request.login(platform_admin_user)
+        client_request.post(
+            "main.service_sending_domain",
+            service_id=SERVICE_ONE_ID,
+            _data={"sending_domain": "domain2.com"},
+            _expected_redirect=url_for(
+                "main.service_settings",
+                service_id=SERVICE_ONE_ID,
+            ),
+            _test_page_title=False,
+        )
+
+        mock_update_service.assert_called_once_with(service_one["id"], sending_domain="domain2.com")
+
+    def test_sending_domain_page_doesnt_update_if_domain_not_in_allowed_list(
+        self, client_request, platform_admin_user, mock_get_service_settings_page_common, mocker, service_one, mock_update_service
+    ):
+        mocker.patch("app.main.views.service_settings.get_verified_ses_domains", return_value=["domain1.com", "domain2.com"])
+
+        client_request.login(platform_admin_user)
+        page = client_request.post(
+            "main.service_sending_domain",
+            service_id=SERVICE_ONE_ID,
+            _data={"sending_domain": "domain3.com"},
+            _expected_status=200,
+            _test_page_title=False,
+        )
+
+        assert mock_update_service.called is False
+        assert "Not a valid choice" in page.text
+
+    def test_sending_domain_page_404s_for_non_platform_admin(self, client_request, mock_get_service_settings_page_common, mocker):
+        client_request.get("main.service_sending_domain", service_id=SERVICE_ONE_ID, _expected_status=403)

tests/app/notify_client/test_notification_counts_client.py

@@ -1,9 +1,9 @@
-from datetime import datetime
 from unittest.mock import Mock, patch
 
 import pytest
 
 from app.notify_client.notification_counts_client import NotificationCounts
+from app.utils import get_current_financial_year
 
 
 @pytest.fixture
@@ -199,4 +199,7 @@ def test_get_limit_stats_dependencies_called(self, mocker):
 
         # Assert dependencies called
         mock_today.assert_called_once_with(mock_service.id)
-        mock_year.assert_called_once_with(mock_service.id, datetime.now().year)
+        mock_year.assert_called_once_with(
+            mock_service.id,
+            get_current_financial_year(),
+        )

tests/app/test_utils.py

@@ -2,6 +2,7 @@
 from csv import DictReader
 from io import StringIO
 from pathlib import Path
+from unittest.mock import Mock, patch
 from urllib.parse import unquote
 
 import pytest
@@ -23,6 +24,7 @@
     get_new_default_reply_to_address,
     get_remote_addr,
     get_template,
+    get_verified_ses_domains,
     printing_today_or_tomorrow,
     report_security_finding,
 )
@@ -759,3 +761,52 @@ def test_get_new_default_reply_to_address_returns_none_if_one_reply_to(mocker: M
 
     new_default = get_new_default_reply_to_address(email_reply_tos, reply_to_1)  # type: ignore
     assert new_default is None
+
+
+class TestGetSESDomains:
+    @pytest.fixture
+    def mock_ses_client(self):
+        with patch("boto3.client") as mock_client:
+            mock_ses = Mock()
+            mock_client.return_value = mock_ses
+            yield mock_ses
+
+    @pytest.fixture(autouse=True)
+    def mock_cache_decorator(self):
+        with patch("app.utils.cache.memoize", lambda *args, **kwargs: lambda f: f):
+            yield
+
+    def test_get_verified_ses_domains(self, app_, mock_ses_client):
+        # Setup mock return values
+        mock_ses_client.list_identities.return_value = {"Identities": ["domain1.com", "domain2.com", "domain3.com"]}
+
+        mock_ses_client.get_identity_verification_attributes.return_value = {
+            "VerificationAttributes": {
+                "domain1.com": {"VerificationStatus": "Success"},
+                "domain2.com": {"VerificationStatus": "Failed"},
+                "domain3.com": {"VerificationStatus": "Pending"},
+            }
+        }
+
+        # Execute within app context
+        with app_.test_request_context():
+            result = get_verified_ses_domains()
+
+            # Assert
+            assert result == ["domain1.com"]
+            mock_ses_client.list_identities.assert_called_once_with(IdentityType="Domain")
+            mock_ses_client.get_identity_verification_attributes.assert_called_once_with(
+                Identities=["domain1.com", "domain2.com", "domain3.com"]
+            )
+
+    def test_get_verified_ses_domains_no_domains(self, app_, mock_ses_client):
+        # Setup empty response
+        mock_ses_client.list_identities.return_value = {"Identities": []}
+        mock_ses_client.get_identity_verification_attributes.return_value = {"VerificationAttributes": {}}
+
+        # Execute within app context
+        with app_.test_request_context():
+            result = get_verified_ses_domains()
+
+            # Assert
+            assert result == []