diff --git a/.github/scripts/check-idor-model-coverage.py b/.github/scripts/check-idor-model-coverage.py index 00ca07c20a34..80cc9c0bacad 100644 --- a/.github/scripts/check-idor-model-coverage.py +++ b/.github/scripts/check-idor-model-coverage.py @@ -327,6 +327,7 @@ def get_scoped_models() -> tuple[dict[str, set[str]], set[str], set[str], set[st "DevStackImageTarget", "DevStackImageAttempt", "License", + "SandboxSenderTenantState", # instance-global mirror of the shared sandbox SES tenant state # --- Deprecated --- "Prompt", "PromptSequence", diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index f85b1832321d..37d323ec0f5c 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -110,6 +110,8 @@ A reached allowance skips the send instead of delaying it. A limiter error, or a The worker checks members after the claim and gives the claim back when that check skips the send. An SES throttle response also gives the claim back before the delivery retry. Returns are deduplicated, and Valkey ignores a return that arrives more than five minutes after it was issued. Test sends run inline in the CDP API, so a sandbox-enabled CDP API opens the SES Valkey pool (`SES_RATE_LIMITER_VALKEY_*`) for the allowances. Without it, every sandbox test send skips. +Django mirrors the tenant named by `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` from the SES tenant-state webhook and the daily reconciliation, without emailing project admins. Set the same `SES_SANDBOX_TENANT_NAME` on Django web, Celery, and the worker. +A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Each delivered send emits a `workflows sandbox email sent` event. Test sends never record `email_sent`, so they do not count toward the sending tier. Live workflow sends still count as `email_sent`, which the sending tier reads, until sandbox sends get their own metric. diff --git a/posthog/plugins/plugin_server_api.py b/posthog/plugins/plugin_server_api.py index 071ff89c4d5f..c8f46f8585db 100644 --- a/posthog/plugins/plugin_server_api.py +++ b/posthog/plugins/plugin_server_api.py @@ -82,6 +82,11 @@ def reload_team_workflows_config_on_workers(team_id: int): publish_message("reload-team-workflows-config", {"teamId": team_id}) +def reload_sandbox_sender_state_on_workers(tenant_name: str) -> None: + logger.info(f"Reloading sandbox sender state for tenant {tenant_name} on workers") + publish_message("reload-sandbox-sender-state", {"tenantName": tenant_name}) + + def populate_plugin_capabilities_on_workers(plugin_id: str): logger.info(f"Populating plugin capabilities for plugin {plugin_id} on workers") publish_message("populate-plugin-capabilities", {"pluginId": plugin_id}) diff --git a/products/workflows/backend/migrations/0040_sandbox_sender_tenant_state.py b/products/workflows/backend/migrations/0040_sandbox_sender_tenant_state.py new file mode 100644 index 000000000000..6a00259f3096 --- /dev/null +++ b/products/workflows/backend/migrations/0040_sandbox_sender_tenant_state.py @@ -0,0 +1,32 @@ +# Generated by Django 5.2.18 on 2026-10-27 07:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("workflows", "0039_workflowimport"), + ] + + operations = [ + migrations.CreateModel( + name="SandboxSenderTenantState", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("tenant_name", models.CharField(max_length=64, unique=True)), + ( + "sending_status", + models.CharField(blank=True, db_default="", default="", max_length=32), + ), + ("synced_at", models.DateTimeField(blank=True, null=True)), + ], + ), + ] diff --git a/products/workflows/backend/migrations/max_migration.txt b/products/workflows/backend/migrations/max_migration.txt index b8a83e817ac3..bce2c1cd8817 100644 --- a/products/workflows/backend/migrations/max_migration.txt +++ b/products/workflows/backend/migrations/max_migration.txt @@ -1 +1 @@ -0039_workflowimport +0040_sandbox_sender_tenant_state diff --git a/products/workflows/backend/models/__init__.py b/products/workflows/backend/models/__init__.py index 7bfb7e8871b6..a0c363b3abf0 100644 --- a/products/workflows/backend/models/__init__.py +++ b/products/workflows/backend/models/__init__.py @@ -4,6 +4,7 @@ from .hog_flow_optimization import HogFlowOptimization from .hog_flow_revision import HogFlowRevision from .hog_flow_schedule.hog_flow_schedule import HogFlowSchedule +from .sandbox_sender_tenant_state import SandboxSenderTenantState from .team_workflows_config import TeamWorkflowsConfig from .workflow_idea import WorkflowIdea from .workflow_import import WorkflowImport @@ -16,6 +17,7 @@ "HogFlowRevision", "HogFlowSchedule", "HogFlowTemplate", + "SandboxSenderTenantState", "TeamWorkflowsConfig", "WorkflowIdea", "WorkflowImport", diff --git a/products/workflows/backend/models/sandbox_sender_tenant_state.py b/products/workflows/backend/models/sandbox_sender_tenant_state.py new file mode 100644 index 000000000000..f6b9172850bc --- /dev/null +++ b/products/workflows/backend/models/sandbox_sender_tenant_state.py @@ -0,0 +1,9 @@ +from django.db import models + + +class SandboxSenderTenantState(models.Model): + """Instance-level mirror of the shared sandbox SES tenant; platform state, not tenant data, so no team_id.""" + + tenant_name = models.CharField(max_length=64, unique=True) + sending_status = models.CharField(max_length=32, blank=True, default="", db_default="") + synced_at = models.DateTimeField(null=True, blank=True) diff --git a/products/workflows/backend/providers/ses.py b/products/workflows/backend/providers/ses.py index 4ecb6490fda4..763abd7a8495 100644 --- a/products/workflows/backend/providers/ses.py +++ b/products/workflows/backend/providers/ses.py @@ -395,6 +395,13 @@ def _is_throttling(error: Exception) -> bool: return isinstance(error, ClientError) and error.response["Error"]["Code"] in ("Throttling", "ThrottlingException") +@frozen +class TenantSendingState: + sending_status: str + reputation_impact: str | None + tenant_arn: str | None + + class SESProvider: ses_client: "SESClient" ses_v2_client: "SESV2Client" @@ -469,7 +476,17 @@ def get_tenant_reputation(self, team_id: int) -> dict[str, Any] | None: provider feedback loops, third-party listings), so this is the authoritative health source; our own app metrics only provide the per-workflow diagnosis. """ - tenant_name = self._tenant_name_for_team(team_id) + state = self.get_tenant_sending_state(self._tenant_name_for_team(team_id)) + if state is None: + return None + return { + "sending_status": state.sending_status, + "reputation_impact": state.reputation_impact, + "findings": self._open_findings(state.tenant_arn) if state.tenant_arn else [], + } + + def get_tenant_sending_state(self, tenant_name: str) -> TenantSendingState | None: + """Sending status and reputation impact of the named SES tenant, or None when it doesn't exist.""" try: tenant = self.ses_v2_client.get_tenant(TenantName=tenant_name)["Tenant"] except ClientError as e: @@ -479,40 +496,38 @@ def get_tenant_reputation(self, team_id: int) -> dict[str, Any] | None: sending_status: str = tenant.get("SendingStatus", "ENABLED") tenant_arn = tenant.get("TenantArn") - reputation_impact: str | None = None - findings: list[dict[str, Any]] = [] + if not tenant_arn: + return TenantSendingState(sending_status=sending_status, reputation_impact=None, tenant_arn=None) - if tenant_arn: - try: - entity = self.ses_v2_client.get_reputation_entity( - ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE" - )["ReputationEntity"] - except ClientError as e: - # A tenant with no attributed sends yet has no reputation entity. - if e.response["Error"]["Code"] != "NotFoundException": - raise - entity = {} - reputation_impact = entity.get("ReputationImpact") + entity = self._reputation_entity(tenant_arn) + return TenantSendingState( # The aggregate folds in both AWS-managed and customer-managed pauses. - sending_status = entity.get("SendingStatusAggregate", sending_status) - - # RESOURCE_ARN is the only filter key AWS documents as usable on its own with this - # scoping (see _iter_open_recommendations for why STATUS is filtered locally). - findings.extend( - { - "finding_type": recommendation.get("Type", ""), - "impact": recommendation.get("Impact", "LOW"), - "description": recommendation.get("Description", ""), - "last_updated_at": recommendation.get("LastUpdatedTimestamp"), - } - for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn}) - ) + sending_status=entity.get("SendingStatusAggregate", sending_status), + reputation_impact=entity.get("ReputationImpact"), + tenant_arn=tenant_arn, + ) - return { - "sending_status": sending_status, - "reputation_impact": reputation_impact, - "findings": findings, - } + def _reputation_entity(self, tenant_arn: str) -> Mapping[str, Any]: + try: + return self.ses_v2_client.get_reputation_entity( + ReputationEntityReference=tenant_arn, ReputationEntityType="RESOURCE" + )["ReputationEntity"] + except ClientError as e: + # A tenant with no attributed sends yet has no reputation entity. + if e.response["Error"]["Code"] != "NotFoundException": + raise + return {} + + def _open_findings(self, tenant_arn: str) -> list[dict[str, Any]]: + return [ + { + "finding_type": recommendation.get("Type", ""), + "impact": recommendation.get("Impact", "LOW"), + "description": recommendation.get("Description", ""), + "last_updated_at": recommendation.get("LastUpdatedTimestamp"), + } + for recommendation in self._iter_open_recommendations({"RESOURCE_ARN": tenant_arn}) + ] def _iter_open_recommendations( self, finding_filter: dict[Any, str] | None, client: "SESV2Client | None" = None diff --git a/products/workflows/backend/services/ses_tenant_events.py b/products/workflows/backend/services/ses_tenant_events.py index 77672ddc31cf..423cd5837aa5 100644 --- a/products/workflows/backend/services/ses_tenant_events.py +++ b/products/workflows/backend/services/ses_tenant_events.py @@ -15,21 +15,22 @@ from collections.abc import Mapping from typing import Any +from django.conf import settings + import requests import structlog from posthog.ingress.contracts import WebhookDelivery from posthog.ingress.verify.sns_signature import is_valid_sns_url -from products.workflows.backend.tasks.ses_tenant_state import sync_ses_tenant_state_task +from products.workflows.backend.tasks.ses_tenant_state import sync_sandbox_tenant_state_task, sync_ses_tenant_state_task logger = structlog.get_logger(__name__) _TENANT_TEAM_RE = re.compile(r"\bteam-(\d+)\b") -def _extract_team_id(event: dict[str, Any]) -> int | None: - """Pull the `team-` tenant name out of an EventBridge event, wherever AWS put it.""" +def _tenant_candidates(event: dict[str, Any]) -> list[str]: detail = event.get("detail") or {} candidates: list[Any] = [ detail.get("tenantName"), @@ -39,8 +40,24 @@ def _extract_team_id(event: dict[str, Any]) -> int | None: resources = event.get("resources") if isinstance(resources, list): candidates.extend(resources) - for candidate in candidates: - if isinstance(candidate, str) and (match := _TENANT_TEAM_RE.search(candidate)): + return [candidate for candidate in candidates if isinstance(candidate, str)] + + +def _names_sandbox_tenant(event: dict[str, Any]) -> bool: + tenant_name = settings.SES_SANDBOX_TENANT_NAME + if not tenant_name: + return False + tenant_path = re.compile(rf"(?:/|:)tenant/{re.escape(tenant_name)}(?:/|$)") + return any( + candidate == tenant_name or (candidate.startswith("arn:") and tenant_path.search(candidate) is not None) + for candidate in _tenant_candidates(event) + ) + + +def _extract_team_id(event: dict[str, Any]) -> int | None: + """Pull the `team-` tenant name out of an EventBridge event, wherever AWS put it.""" + for candidate in _tenant_candidates(event): + if match := _TENANT_TEAM_RE.search(candidate): return int(match.group(1)) return None @@ -61,7 +78,7 @@ def _confirm_subscription(message: Mapping[str, Any]) -> None: def _handle_notification(message: Mapping[str, Any]) -> None: - """Enqueue a tenant-state sync for the team the EventBridge event names. + """Enqueue a tenant-state sync for the sandbox tenant or the team the EventBridge event names. Every unusable payload is dropped rather than raised, because a redelivery would carry the same unusable payload. @@ -73,6 +90,11 @@ def _handle_notification(message: Mapping[str, Any]) -> None: if not isinstance(event, dict) or event.get("source") != "aws.ses": return + if _names_sandbox_tenant(event): + logger.info("ses_tenant_events_webhook_accepted", tenant="sandbox", detail_type=event.get("detail-type")) + sync_sandbox_tenant_state_task.delay() + return + team_id = _extract_team_id(event) if team_id is None: logger.warning("ses_tenant_events_webhook_no_tenant", detail_type=event.get("detail-type")) diff --git a/products/workflows/backend/services/ses_tenant_state.py b/products/workflows/backend/services/ses_tenant_state.py index 99f9f5c3c3de..d713c4671204 100644 --- a/products/workflows/backend/services/ses_tenant_state.py +++ b/products/workflows/backend/services/ses_tenant_state.py @@ -1,21 +1,26 @@ from collections.abc import Callable from typing import Any +from django.conf import settings from django.db import transaction from django.utils import timezone import structlog from posthog.models.team import Team -from posthog.plugins.plugin_server_api import reload_team_workflows_config_on_workers +from posthog.plugins.plugin_server_api import ( + reload_sandbox_sender_state_on_workers, + reload_team_workflows_config_on_workers, +) from posthog.tasks.email import ( send_email_sending_reputation_finding, send_email_sending_suspended, send_email_sending_unsuspended, ) +from products.workflows.backend.models.sandbox_sender_tenant_state import SandboxSenderTenantState from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig -from products.workflows.backend.providers.ses import SESProvider +from products.workflows.backend.providers.ses import SESProvider, TenantSendingState logger = structlog.get_logger(__name__) @@ -58,6 +63,39 @@ def _announce_state_change(team_id: int) -> None: logger.exception("Failed to announce SES tenant state change to workers", team_id=team_id) +def _apply_sandbox_tenant_state(state: SandboxSenderTenantState, tenant: TenantSendingState) -> None: + previous_status = state.sending_status + state.synced_at = timezone.now() + if previous_status == tenant.sending_status: + state.save(update_fields=["synced_at"]) + logger.info("Sandbox SES tenant state unchanged", tenant_name=state.tenant_name, status=previous_status) + return + + state.sending_status = tenant.sending_status + state.save(update_fields=["sending_status", "synced_at"]) + logger.info( + "Sandbox SES tenant state changed", + tenant_name=state.tenant_name, + from_status=previous_status, + to_status=tenant.sending_status, + ) + tenant_name = state.tenant_name + transaction.on_commit(lambda: reload_sandbox_sender_state_on_workers(tenant_name), robust=True) + + +def sync_sandbox_tenant_state(provider: SESProvider | None = None) -> None: + # Serialize SES reads so a stale sweep cannot overwrite a newer pause. + tenant_name = settings.SES_SANDBOX_TENANT_NAME + if not tenant_name: + return + SandboxSenderTenantState.objects.get_or_create(tenant_name=tenant_name) + with transaction.atomic(): + state = SandboxSenderTenantState.objects.select_for_update().get(tenant_name=tenant_name) + tenant = (provider or SESProvider()).get_tenant_sending_state(tenant_name) + if tenant is not None: + _apply_sandbox_tenant_state(state, tenant) + + def sync_ses_tenant_state(team_id: int, provider: SESProvider | None = None, *, verify_team: bool = True) -> None: """ Fetch the authoritative AWS SES tenant state for a team and apply it. Called from the diff --git a/products/workflows/backend/tasks/ses_tenant_state.py b/products/workflows/backend/tasks/ses_tenant_state.py index 55b41d59978f..d8c01d8c355f 100644 --- a/products/workflows/backend/tasks/ses_tenant_state.py +++ b/products/workflows/backend/tasks/ses_tenant_state.py @@ -6,7 +6,7 @@ from posthog.tasks.utils import CeleryQueue from products.workflows.backend.providers.ses import SESProvider -from products.workflows.backend.services.ses_tenant_state import sync_ses_tenant_state +from products.workflows.backend.services.ses_tenant_state import sync_sandbox_tenant_state, sync_ses_tenant_state logger = get_logger(__name__) @@ -27,6 +27,19 @@ def sync_ses_tenant_state_task(team_id: int) -> None: sync_ses_tenant_state(team_id) +@shared_task( + ignore_result=True, + queue=CeleryQueue.DEFAULT.value, + autoretry_for=(Exception,), + retry_backoff=60, + retry_backoff_max=600, + max_retries=5, + retry_jitter=True, +) +def sync_sandbox_tenant_state_task() -> None: + sync_sandbox_tenant_state() + + @shared_task(ignore_result=True, queue=CeleryQueue.LONG_RUNNING.value) @skip_team_scope_audit def reconcile_ses_tenant_states() -> None: @@ -44,6 +57,10 @@ def reconcile_ses_tenant_states() -> None: # One provider for the whole sweep: a fresh SESProvider per team would rebuild its boto3 # clients (and re-resolve credentials) thousands of times. provider = SESProvider() + try: + sync_sandbox_tenant_state(provider=provider) + except Exception: + logger.exception("SES sandbox tenant reconciliation failed") synced = 0 failed = 0 for team_id in team_ids.iterator(chunk_size=500): diff --git a/products/workflows/backend/test/test_ses_tenant_events.py b/products/workflows/backend/test/test_ses_tenant_events.py index 8b4f7aa2d9e8..fb4fe7f6aa6a 100644 --- a/products/workflows/backend/test/test_ses_tenant_events.py +++ b/products/workflows/backend/test/test_ses_tenant_events.py @@ -23,6 +23,7 @@ from products.workflows.backend.facade.api import accept_ses_event TOPIC = "arn:aws:sns:us-east-1:123456789012:ses-tenant-events" +SANDBOX_TENANT = "sandbox-team-1" WEBHOOK_PATH = "/webhooks/workflows/ses-events" SUBSCRIBE_URL = "https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription&Token=tok" _CERT_URL = "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-01d088a6f77103d0fe307c0069e40ed6.pem" @@ -106,6 +107,9 @@ def setUp(self) -> None: sync = patch("products.workflows.backend.services.ses_tenant_events.sync_ses_tenant_state_task") self.sync_mock = sync.start() self.addCleanup(sync.stop) + sandbox_sync = patch("products.workflows.backend.services.ses_tenant_events.sync_sandbox_tenant_state_task") + self.sandbox_sync_mock = sandbox_sync.start() + self.addCleanup(sandbox_sync.stop) def test_enqueues_a_sync_for_the_tenant_named_in_the_event(self) -> None: accept_ses_event(_delivery(_notification(_eventbridge_event()))) @@ -119,6 +123,31 @@ def test_finds_the_tenant_in_resource_arns_when_detail_has_no_name(self) -> None self.sync_mock.delay.assert_called_once_with(7) + @parameterized.expand( + [ + ("sandbox_tenant_by_name", {"tenantName": SANDBOX_TENANT}, [], "sandbox"), + ( + "sandbox_tenant_by_arn", + {}, + [f"arn:aws:ses:us-east-1:123456789012:tenant/{SANDBOX_TENANT}/abc"], + "sandbox", + ), + ("team_tenant", {"tenantName": "team-42"}, [], "team"), + ] + ) + @override_settings(SES_SANDBOX_TENANT_NAME=SANDBOX_TENANT) + def test_routes_the_sandbox_tenant_to_its_own_sync( + self, _name: str, detail: dict[str, Any], resources: list[str], expected_sync: str + ) -> None: + accept_ses_event(_delivery(_notification(_eventbridge_event(detail=detail, resources=resources)))) + + if expected_sync == "sandbox": + self.sandbox_sync_mock.delay.assert_called_once_with() + self.sync_mock.delay.assert_not_called() + else: + self.sync_mock.delay.assert_called_once_with(42) + self.sandbox_sync_mock.delay.assert_not_called() + def test_ignores_events_from_other_sources(self) -> None: accept_ses_event(_delivery(_notification(_eventbridge_event(source="aws.health")))) diff --git a/products/workflows/backend/test/test_ses_tenant_state.py b/products/workflows/backend/test/test_ses_tenant_state.py index e00b17a1df33..214aa4f1f8fb 100644 --- a/products/workflows/backend/test/test_ses_tenant_state.py +++ b/products/workflows/backend/test/test_ses_tenant_state.py @@ -1,10 +1,18 @@ +from datetime import UTC, datetime + +import time_machine from posthog.test.base import BaseTest from unittest.mock import MagicMock, patch +from django.test import override_settings from django.utils import timezone +from botocore.exceptions import ClientError from parameterized import parameterized +from posthog.models.integration import Integration + +from products.workflows.backend.models.sandbox_sender_tenant_state import SandboxSenderTenantState from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig from products.workflows.backend.services.ses_tenant_state import ( PROVIDER_PAUSE_REASON, @@ -12,6 +20,10 @@ apply_ses_tenant_state, sync_ses_tenant_state, ) +from products.workflows.backend.tasks.ses_tenant_state import ( + reconcile_ses_tenant_states, + sync_sandbox_tenant_state_task, +) class TestApplySesTenantState(BaseTest): @@ -183,3 +195,114 @@ def test_unknown_team_is_skipped_without_calling_ses(self) -> None: provider.get_tenant_reputation.assert_not_called() assert not TeamWorkflowsConfig.objects.filter(team_id=self.team.id + 99_999).exists() + + +SANDBOX_TENANT = "workflows-sandbox" +SANDBOX_TENANT_ARN = f"arn:aws:ses:us-east-1:123456789012:tenant/{SANDBOX_TENANT}/abc" + + +@override_settings(SES_SANDBOX_TENANT_NAME=SANDBOX_TENANT) +class TestSandboxTenantStateSync(BaseTest): + def setUp(self) -> None: + super().setUp() + boto3_client = patch("products.workflows.backend.providers.ses.boto3.client") + self.ses = boto3_client.start().return_value + self.addCleanup(boto3_client.stop) + publish = patch("posthog.plugins.plugin_server_api.publish_message") + self.publish = publish.start() + self.addCleanup(publish.stop) + self.ses.list_recommendations.return_value = {"Recommendations": []} + + def _aws_tenant(self, sending_status: str, impact: str) -> None: + self.ses.get_tenant.return_value = {"Tenant": {"TenantName": SANDBOX_TENANT, "TenantArn": SANDBOX_TENANT_ARN}} + self.ses.get_reputation_entity.return_value = { + "ReputationEntity": {"ReputationImpact": impact, "SendingStatusAggregate": sending_status} + } + + def _sync(self) -> None: + with self.captureOnCommitCallbacks(execute=True): + sync_sandbox_tenant_state_task() + + def test_a_changed_state_is_stored_and_announced_after_commit(self) -> None: + self._aws_tenant("ENABLED", "NONE") + self._sync() + self.publish.reset_mock() + self._aws_tenant("DISABLED", "HIGH") + + with self.captureOnCommitCallbacks(execute=True): + sync_sandbox_tenant_state_task() + self.publish.assert_not_called() + + state = SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT) + assert state.sending_status == "DISABLED" + self.ses.get_tenant.assert_called_with(TenantName=SANDBOX_TENANT) + self.publish.assert_called_once_with("reload-sandbox-sender-state", {"tenantName": SANDBOX_TENANT}) + + @parameterized.expand([("same_impact", "HIGH"), ("only_the_impact_changed", "LOW")]) + def test_an_unchanged_sending_status_refreshes_synced_at_without_announcing(self, _name: str, impact: str) -> None: + self._aws_tenant("DISABLED", "HIGH") + with time_machine.travel("2026-10-05T10:00:00Z", tick=False): + self._sync() + self.publish.reset_mock() + self._aws_tenant("DISABLED", impact) + + with time_machine.travel("2026-10-05T11:00:00Z", tick=False): + self._sync() + + state = SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT) + assert state.sending_status == "DISABLED" + assert state.synced_at == datetime(2026, 10, 5, 11, tzinfo=UTC) + self.publish.assert_not_called() + + def test_a_pause_is_stored_even_when_findings_cannot_be_listed(self) -> None: + self._aws_tenant("DISABLED", "HIGH") + self.ses.list_recommendations.side_effect = ClientError( + {"Error": {"Code": "TooManyRequestsException"}}, "ListRecommendations" + ) + + self._sync() + + assert SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT).sending_status == "DISABLED" + + def test_an_unknown_sandbox_tenant_is_never_paused(self) -> None: + self.ses.get_tenant.side_effect = ClientError({"Error": {"Code": "NotFoundException"}}, "GetTenant") + + self._sync() + + state = SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT) + assert (state.sending_status, state.synced_at) == ("", None) + self.publish.assert_not_called() + + def test_the_daily_sweep_syncs_the_sandbox_tenant_once_beside_team_tenants(self) -> None: + Integration.objects.create(team=self.team, kind="email", config={"provider": "ses"}) + self._aws_tenant("ENABLED", "LOW") + + with self.captureOnCommitCallbacks(execute=True): + reconcile_ses_tenant_states() + + tenant_names = [call.kwargs["TenantName"] for call in self.ses.get_tenant.call_args_list] + assert sorted(tenant_names) == sorted([SANDBOX_TENANT, f"team-{self.team.id}"]) + assert SandboxSenderTenantState.objects.get(tenant_name=SANDBOX_TENANT).sending_status == "ENABLED" + + def test_a_failed_sandbox_sync_does_not_stop_the_team_sweep(self) -> None: + Integration.objects.create(team=self.team, kind="email", config={"provider": "ses"}) + + def get_tenant(TenantName: str) -> dict: + if TenantName == SANDBOX_TENANT: + raise ClientError({"Error": {"Code": "TooManyRequestsException"}}, "GetTenant") + return {"Tenant": {"TenantName": TenantName, "SendingStatus": "ENABLED"}} + + self.ses.get_tenant.side_effect = get_tenant + + with self.captureOnCommitCallbacks(execute=True): + reconcile_ses_tenant_states() + + assert TeamWorkflowsConfig.objects.get(team=self.team).ses_tenant_sending_status == "ENABLED" + + @override_settings(SES_SANDBOX_TENANT_NAME="") + def test_nothing_is_synced_while_no_sandbox_tenant_is_configured(self) -> None: + sync_sandbox_tenant_state_task() + reconcile_ses_tenant_states() + + self.ses.get_tenant.assert_not_called() + assert not SandboxSenderTenantState.objects.exists()