fix/centralizing anchor rendering in html link message

Author: chintakjoshi

app/config.py

@@ -24,6 +24,8 @@
 _UNSET = object()
 _PENDING_SINGLETON_CLEANUPS: deque[tuple[str, Callable[[Any], Any], Any]] = deque()
 _PENDING_SINGLETON_CLEANUPS_LOCK = Lock()
+_SCHEDULED_SINGLETON_CLEANUPS: set[asyncio.Task[None]] = set()
+_SCHEDULED_SINGLETON_CLEANUPS_LOCK = Lock()
 _REGISTERED_SINGLETON_CLEARERS: list[Callable[[], None]] = []
 _REGISTERED_SINGLETON_CLEARERS_LOCK = Lock()
 _cleanup_logger = logging.getLogger(__name__)
@@ -410,6 +412,18 @@ async def _run_singleton_cleanup(name: str, cleanup: Callable[[Any], Any], value
         _cleanup_logger.exception("reloadable_singleton_cleanup_failed", extra={"name": name})
 
 
+def _track_singleton_cleanup_task(task: asyncio.Task[None]) -> None:
+    """Track one in-loop singleton cleanup task until it finishes."""
+    with _SCHEDULED_SINGLETON_CLEANUPS_LOCK:
+        _SCHEDULED_SINGLETON_CLEANUPS.add(task)
+
+    def _discard(completed_task: asyncio.Task[None]) -> None:
+        with _SCHEDULED_SINGLETON_CLEANUPS_LOCK:
+            _SCHEDULED_SINGLETON_CLEANUPS.discard(completed_task)
+
+    task.add_done_callback(_discard)
+
+
 def _schedule_singleton_cleanup(name: str, cleanup: Callable[[Any], Any], value: Any) -> None:
     """Run cleanup immediately when possible or enqueue it for later draining."""
     try:
@@ -419,21 +433,50 @@ def _schedule_singleton_cleanup(name: str, cleanup: Callable[[Any], Any], value:
             _PENDING_SINGLETON_CLEANUPS.append((name, cleanup, value))
         return
 
-    loop.create_task(_run_singleton_cleanup(name, cleanup, value))
+    task = loop.create_task(_run_singleton_cleanup(name, cleanup, value))
+    _track_singleton_cleanup_task(task)
+
+
+async def _await_scheduled_singleton_cleanups() -> None:
+    """Await tracked cleanup tasks bound to the current running loop."""
+    running_loop = asyncio.get_running_loop()
+
+    while True:
+        with _SCHEDULED_SINGLETON_CLEANUPS_LOCK:
+            tasks = [
+                task
+                for task in _SCHEDULED_SINGLETON_CLEANUPS
+                if not task.done() and task.get_loop() is running_loop
+            ]
+
+        if not tasks:
+            return
+
+        await asyncio.gather(*tasks)
 
 
 async def flush_pending_singleton_cleanups() -> None:
-    """Drain any deferred singleton cleanup work."""
+    """Drain deferred and in-flight singleton cleanup work."""
     while True:
         with _PENDING_SINGLETON_CLEANUPS_LOCK:
-            if not _PENDING_SINGLETON_CLEANUPS:
-                return
             pending = list(_PENDING_SINGLETON_CLEANUPS)
             _PENDING_SINGLETON_CLEANUPS.clear()
 
         for name, cleanup, value in pending:
             await _run_singleton_cleanup(name, cleanup, value)
 
+        await _await_scheduled_singleton_cleanups()
+
+        with _PENDING_SINGLETON_CLEANUPS_LOCK:
+            has_pending = bool(_PENDING_SINGLETON_CLEANUPS)
+        with _SCHEDULED_SINGLETON_CLEANUPS_LOCK:
+            has_scheduled = any(
+                not task.done() and task.get_loop() is asyncio.get_running_loop()
+                for task in _SCHEDULED_SINGLETON_CLEANUPS
+            )
+        if not has_pending and not has_scheduled:
+            return
+
 
 def clear_reloadable_singletons() -> None:
     """Clear all registered reloadable singleton caches."""

app/services/lifecycle_service.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import html
 import secrets
 import smtplib
 from dataclasses import dataclass
@@ -102,9 +103,9 @@ async def send_verification_email(self, to_email: str, verification_link: str) -
             to_email=to_email,
             subject=subject,
             body=f"Open this link to verify your account: {verification_link}",
-            html_body=(
-                "<p>Open this link to verify your account:</p>"
-                f'<p><a href="{verification_link}">{verification_link}</a></p>'
+            html_body=self._html_link_message(
+                intro_text="Open this link to verify your account:",
+                link=verification_link,
             ),
         )
 
@@ -115,9 +116,9 @@ async def send_password_reset_email(self, to_email: str, reset_link: str) -> Non
             to_email=to_email,
             subject="Reset your password",
             body=f"Open this link to reset your password: {reset_link}",
-            html_body=(
-                "<p>Open this link to reset your password:</p>"
-                f'<p><a href="{reset_link}">{reset_link}</a></p>'
+            html_body=self._html_link_message(
+                intro_text="Open this link to reset your password:",
+                link=reset_link,
             ),
         )
 
@@ -149,6 +150,13 @@ def _send_blocking(
         with smtplib.SMTP(self.host, self.port, timeout=10) as smtp:
             smtp.send_message(message)
 
+    @staticmethod
+    def _html_link_message(*, intro_text: str, link: str) -> str:
+        """Build one HTML paragraph pair with a safely escaped anchor."""
+        escaped_intro = html.escape(intro_text, quote=False)
+        escaped_link = html.escape(link, quote=True)
+        return f'<p>{escaped_intro}</p><p><a href="{escaped_link}">{escaped_link}</a></p>'
+
 
 class LifecycleService:
     """Lifecycle orchestration for signup verification and resend flows."""

tests/integration/conftest.py

@@ -96,39 +96,6 @@ def _clear_dependency_caches() -> None:
     get_webhook_service.cache_clear()
 
 
-async def _close_async_client(client: Any) -> None:
-    """Close async client instances regardless of redis-py close API version."""
-    close = getattr(client, "aclose", None)
-    if callable(close):
-        await close()
-        return
-
-    close = getattr(client, "close", None)
-    if callable(close):
-        result = close()
-        if hasattr(result, "__await__"):
-            await result
-
-
-async def _dispose_async_singletons() -> None:
-    """Dispose loop-bound async resources before changing event loops."""
-    from app.core.sessions import get_redis_client
-    from app.db.session import dispose_engine, get_engine
-    from app.middleware.rate_limit import get_rate_limit_redis_client
-
-    redis_client = get_redis_client() if get_redis_client.cache_info().currsize else None
-    rate_limit_client = (
-        get_rate_limit_redis_client() if get_rate_limit_redis_client.cache_info().currsize else None
-    )
-
-    if redis_client is not None:
-        await _close_async_client(redis_client)
-    if rate_limit_client is not None and rate_limit_client is not redis_client:
-        await _close_async_client(rate_limit_client)
-    if get_engine.cache_info().currsize:
-        await dispose_engine()
-
-
 def _redis_connection_url(redis: RedisContainer) -> str:
     """Return a redis:// URL across testcontainers versions."""
     get_url = getattr(redis, "get_connection_url", None)
@@ -307,6 +274,7 @@ async def reset_state(
 ) -> Iterator[None]:
     """Clear DB tables and flush Redis; isolate async singletons per event loop."""
     del integration_env
+    from app.config import shutdown_reloadable_singletons
     from app.core.sessions import get_redis_client
     from app.db.session import get_session_factory
     from app.models.api_key import APIKey
@@ -318,8 +286,7 @@ async def reset_state(
     from app.models.webhook_delivery import WebhookDelivery
     from app.models.webhook_endpoint import WebhookEndpoint
 
-    await _dispose_async_singletons()
-    _clear_dependency_caches()
+    await shutdown_reloadable_singletons()
 
     session_factory = get_session_factory()
     async with session_factory() as session:
@@ -339,8 +306,7 @@ async def reset_state(
     try:
         yield
     finally:
-        await _dispose_async_singletons()
-        _clear_dependency_caches()
+        await shutdown_reloadable_singletons()
 
 
 @pytest.fixture(scope="function")

tests/unit/test_lifecycle_service_edge_cases.py

@@ -237,6 +237,59 @@ def send_message(self, message) -> None:  # type: ignore[no-untyped-def]
     assert messages[1]["Subject"] == "Your password has been reset"
 
 
+@pytest.mark.asyncio
+async def test_mailhog_sender_escapes_links_in_html_email_parts(monkeypatch) -> None:
+    """Lifecycle emails HTML-escape links before interpolating them into anchor tags."""
+    sender = MailhogVerificationEmailSender(
+        host="mailhog", port=1025, email_from="from@example.com"
+    )
+    messages: list[object] = []
+
+    async def _fake_to_thread(func, **kwargs):  # type: ignore[no-untyped-def]
+        return func(**kwargs)
+
+    class _SMTP:
+        def __init__(self, host: str, port: int, timeout: int) -> None:
+            assert host == "mailhog"
+            assert port == 1025
+            assert timeout == 10
+
+        def __enter__(self) -> _SMTP:
+            return self
+
+        def __exit__(self, exc_type, exc, tb) -> None:  # type: ignore[no-untyped-def]
+            return None
+
+        def send_message(self, message) -> None:  # type: ignore[no-untyped-def]
+            messages.append(message)
+
+    monkeypatch.setattr("app.services.lifecycle_service.asyncio.to_thread", _fake_to_thread)
+    monkeypatch.setattr("app.services.lifecycle_service.smtplib.SMTP", _SMTP)
+
+    verification_link = 'https://verify.example/token?value="quoted"&x=<tag>'
+    reset_link = 'https://reset.example/token?value="quoted"&x=<tag>'
+
+    await sender.send_verification_email("user@example.com", verification_link)
+    await sender.send_password_reset_email("user@example.com", reset_link)
+
+    verification_html = messages[0].get_body(preferencelist=("html",)).get_content()
+    reset_html = messages[1].get_body(preferencelist=("html",)).get_content()
+
+    assert (
+        'href="https://verify.example/token?value=&quot;quoted&quot;&amp;x=&lt;tag&gt;"'
+        in verification_html
+    )
+    assert (
+        ">https://verify.example/token?value=&quot;quoted&quot;&amp;x=&lt;tag&gt;<"
+        in verification_html
+    )
+    assert (
+        'href="https://reset.example/token?value=&quot;quoted&quot;&amp;x=&lt;tag&gt;"'
+        in reset_html
+    )
+    assert ">https://reset.example/token?value=&quot;quoted&quot;&amp;x=&lt;tag&gt;<" in reset_html
+
+
 @pytest.mark.asyncio
 async def test_signup_verify_and_resend_cover_rollbacks_and_invalid_paths() -> None:
     """Lifecycle signup/verify/resend cover rollback and invalid-token branches."""

tests/unit/test_reloadable_singletons.py

@@ -9,7 +9,7 @@
 import app.core.sessions as sessions_module
 import app.db.session as db_session_module
 import app.middleware.rate_limit as rate_limit_module
-from app.config import get_settings
+from app.config import flush_pending_singleton_cleanups, get_settings, reloadable_singleton
 
 
 class _AsyncRedisClientStub:
@@ -230,3 +230,33 @@ def _from_url(url: str, **kwargs: object) -> _AsyncRedisClientStub:
     assert second._access_token_ttl_seconds == 450
     assert first._refresh_token_ttl_seconds == 600
     assert second._refresh_token_ttl_seconds == 1200
+
+
+@pytest.mark.asyncio
+async def test_flush_pending_singleton_cleanups_waits_for_in_loop_cleanup_tasks() -> None:
+    """Flushing singleton cleanups should await tasks already scheduled on the active loop."""
+    cleanup_started = asyncio.Event()
+    release_cleanup = asyncio.Event()
+    cleaned_values: list[object] = []
+    value = object()
+
+    async def _cleanup(stale_value: object) -> None:
+        cleanup_started.set()
+        await release_cleanup.wait()
+        cleaned_values.append(stale_value)
+
+    @reloadable_singleton(cleanup=_cleanup)
+    def _get_singleton() -> object:
+        return value
+
+    _get_singleton()
+    _get_singleton.cache_clear()  # type: ignore[attr-defined]
+    await cleanup_started.wait()
+
+    flush_task = asyncio.create_task(flush_pending_singleton_cleanups())
+    await asyncio.sleep(0)
+
+    assert not flush_task.done()
+    release_cleanup.set()
+    await flush_task
+    assert cleaned_values == [value]