test(integration): read the SMTP sink through lock-guarded snapshots and assert the exact deliveries

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -555,7 +555,6 @@ class SlackAlerting(CustomBatchLogger):
         budget_alert_class: Final = get_budget_alert_type(type)
         _id: Final = budget_alert_class.get_id(user_info)
         user_info_str: Final = self._get_user_info_str(user_info)
-        # Team member max-budget alerts ride the key-budget alert type; label them by what they measure.
         event_message = (
             "Team Member Budget: "
             if user_info.event_group == Litellm_EntityType.TEAM_MEMBER

tests/integration/_support/mail.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import socketserver
 import threading
-from collections import deque
 from collections.abc import Generator
 from contextlib import contextmanager
 from dataclasses import dataclass
@@ -30,14 +29,20 @@ class Delivery:
         return ""
 
 
-@dataclass(frozen=True, slots=True)
 class Mailbox:
-    host: str
-    port: int
-    received: deque[Delivery]
+    def __init__(self, host: str, port: int) -> None:
+        self.host: Final = host
+        self.port: Final = port
+        self._lock: Final = threading.Lock()
+        self._deliveries: tuple[Delivery, ...] = ()
 
-    def with_subject(self, fragment: str) -> tuple[Delivery, ...]:
-        return tuple(delivery for delivery in tuple(self.received) if fragment in delivery.subject)
+    def record(self, delivery: Delivery) -> None:
+        with self._lock:
+            self._deliveries = (*self._deliveries, delivery)
+
+    def deliveries(self) -> tuple[Delivery, ...]:
+        with self._lock:
+            return self._deliveries
 
 
 def _address(argument: str) -> str:
@@ -47,7 +52,6 @@ def _address(argument: str) -> str:
 @contextmanager
 def smtp_sink() -> Generator[Mailbox, None, None]:
     """Owned plaintext SMTP peer; deliveries traverse the proxy's real smtplib client."""
-    received: Final[deque[Delivery]] = deque()  # mutable-ok: sink thread appends each delivery
     errors: Final[SimpleQueue[Exception]] = SimpleQueue()
 
     class Handler(socketserver.StreamRequestHandler):
@@ -90,7 +94,7 @@ def smtp_sink() -> Generator[Mailbox, None, None]:
                         if not chunk or chunk == b".\r\n":
                             break
                         body.extend(chunk[1:] if chunk.startswith(b"..") else chunk)
-                    received.append(Delivery(sender, recipients, message_from_bytes(bytes(body))))
+                    mailbox.record(Delivery(sender, recipients, message_from_bytes(bytes(body))))
                     sender, recipients = "", ()
                     self._reply("250 OK queued")
                 elif verb == "RSET":
@@ -109,10 +113,11 @@ def smtp_sink() -> Generator[Mailbox, None, None]:
         daemon_threads = False
 
     with OwnedServer(("127.0.0.1", 0), Handler) as server:
+        mailbox: Final = Mailbox("127.0.0.1", server.server_address[1])
         thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
         thread.start()
         try:
-            yield Mailbox("127.0.0.1", server.server_address[1], received)
+            yield mailbox
         finally:
             server.shutdown()
             thread.join(timeout=6)

tests/integration/spend/test_team_member_budget_alerts.py

@@ -56,7 +56,7 @@ def test_team_member_budget_thresholds_email_member_and_configured_recipients(ga
             eventually(
                 lambda: _membership_spend(user_id, team_id), lambda spend: spend == pytest.approx(CALL_COST), seconds=70
             )
-            assert tuple(mailbox.received) == (), "no threshold is reached before the first call is recorded"
+            assert mailbox.deliveries() == (), "no threshold is reached before the first call is recorded"
 
             second: Final = candidate.request(
                 "POST",
@@ -65,8 +65,9 @@ def test_team_member_budget_thresholds_email_member_and_configured_recipients(ga
                 key=key,
             )
             assert second.status_code == 200, second.text
-            halfway: Final = eventually(lambda: mailbox.with_subject("50%"), lambda found: len(found) >= 1, seconds=30)
+            halfway: Final = eventually(mailbox.deliveries, lambda found: len(found) >= 1, seconds=30)
             assert [delivery.recipients for delivery in halfway] == [(member_email,)], halfway
+            assert "50%" in halfway[0].subject, halfway[0].subject
             assert f"${MEMBER_BUDGET}" in halfway[0].html, halfway[0].html
             eventually(
                 lambda: _membership_spend(user_id, team_id),
@@ -81,10 +82,12 @@ def test_team_member_budget_thresholds_email_member_and_configured_recipients(ga
                 key=key,
             )
             assert third.status_code == 422 and third.json()["error"]["type"] == "budget_exceeded", third.text
-            hundred: Final = eventually(lambda: mailbox.with_subject("100%"), lambda found: len(found) >= 2, seconds=30)
+            capped: Final = eventually(mailbox.deliveries, lambda found: len(found) >= 3, seconds=30)
+            hundred: Final = capped[1:]
+            assert all("100%" in delivery.subject for delivery in hundred), capped
             assert {recipient for delivery in hundred for recipient in delivery.recipients} == {
                 member_email,
                 finance_email,
-            }, hundred
+            }, capped
             assert all(member_email in delivery.html and f"${MEMBER_BUDGET}" in delivery.html for delivery in hundred)
-            assert len(hundred) == 2, hundred
+            assert len(capped) == 3, capped