ci: allow manual publishing of release tags

.github/workflows/publish.yml

@@ -2,6 +2,12 @@ name: Publish
 on:
   push:
     tags: ["*"]
+  workflow_dispatch:
+    inputs:
+      tag:
+        description: Release tag to publish
+        required: true
+        type: string
 permissions:
   id-token: write
   contents: read
@@ -11,6 +17,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v6
+        with:
+          ref: ${{ inputs.tag && format('refs/tags/{0}', inputs.tag) || github.ref }}
       - name: Install uv
         uses: astral-sh/setup-uv@v8.2.0
       - name: Build

.github/workflows/tests.yml

@@ -36,6 +36,7 @@ jobs:
   tests:
     name: Run Tests (${{ matrix.os.name }}, ${{ matrix.label }})
     runs-on: ${{ matrix.os.name }}
+    timeout-minutes: 20
     needs: pre-commit
     strategy:
       fail-fast: false
@@ -64,6 +65,7 @@ jobs:
           - asyncfast-avro
           - asyncfast-cli
           - pytest-amgi
+          - test-utils
     steps:
       - uses: actions/checkout@v6
         with:

packages/amgi-aiobotocore/tests_amgi_aiobotocore/test_sqs_message.py

@@ -1,4 +1,5 @@
 from collections.abc import AsyncGenerator
+from contextlib import AsyncExitStack
 from typing import Any
 from typing import cast
 from typing import Generator
@@ -385,6 +386,30 @@ async def test_lifespan(
             }
 
 
+async def test_lifespan_finishes_redelivered_message(
+    queue_url: str, queue_name: str, moto_endpoint: str, sqs_client: Any
+) -> None:
+    app = MockApp()
+    server = Server(
+        app,
+        queue_name,
+        region_name="us-east-1",
+        endpoint_url=moto_endpoint,
+        aws_access_key_id="test",
+        aws_secret_access_key="test",
+    )
+    async with AsyncExitStack() as calls:
+        async with app.lifespan(server=server):
+            await sqs_client.send_message(QueueUrl=queue_url, MessageBody="value")
+            async with app.call():
+                pass
+            # Hold the redelivery open until after server shutdown. Fixture cleanup
+            # must release it even though the test never completes the call itself.
+            scope, _, _ = await calls.enter_async_context(app.call())
+            assert scope["type"] == "message"
+            assert scope["payload"] == b"value"
+
+
 def test_run(queue_name: str, moto_endpoint: str) -> None:
     assert_run_can_terminate(
         run,

packages/test-utils/pyproject.toml

@@ -11,3 +11,15 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
   "Programming Language :: Python :: 3.14",
 ]
+
+[dependency-groups]
+dev = [
+  "amgi-common==0.50.0",
+  "pytest>=8.4.1",
+  "pytest-asyncio>=1.3.0",
+  "pytest-cov>=7.0.0",
+  "pytest-timeout>=2.4.0",
+]
+
+[tool.uv]
+sources.amgi-common.workspace = true

packages/test-utils/src/test_utils/__init__.py

@@ -6,6 +6,7 @@
 from asyncio import Queue
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
+from contextlib import suppress
 from functools import partial
 from threading import Thread
 from typing import Any
@@ -32,6 +33,8 @@ def __init__(self) -> None:
         self._call_queue = Queue[
             tuple[Scope, AMGIReceiveCallable, AMGISendCallable, Event]
         ]()
+        self._message_return_events: set[Event] = set()
+        self._shutting_down = False
 
     @asynccontextmanager
     async def call(
@@ -50,33 +53,70 @@ async def lifespan(
         loop = asyncio.get_running_loop()
         serve_task = loop.create_task(server.serve()) if server else None
 
-        async with self.call() as (scope, receive, send):
-            assert scope == {
-                "amgi": {"version": "2.0", "spec_version": "2.0"},
-                "type": "lifespan",
-                "state": {},
-            }
-            if state is not None:
-                scope["state"].update(state)
-            lifespan_startup = await receive()
-            assert lifespan_startup == {"type": "lifespan.startup"}
-            await send({"type": "lifespan.startup.complete"})
-            yield
-            if server:
-                server.stop()
-
-            lifespan_shutdown = await receive()
-            assert lifespan_shutdown == {"type": "lifespan.shutdown"}
-            await send({"type": "lifespan.shutdown.complete"})
-        if serve_task:
-            await serve_task
+        try:
+            async with self.call() as (scope, receive, send):
+                assert scope == {
+                    "amgi": {"version": "2.0", "spec_version": "2.0"},
+                    "type": "lifespan",
+                    "state": {},
+                }
+                if state is not None:
+                    scope["state"].update(state)
+                lifespan_startup = await receive()
+                assert lifespan_startup == {"type": "lifespan.startup"}
+                await send({"type": "lifespan.startup.complete"})
+                try:
+                    yield
+                finally:
+                    self._shutting_down = True
+                    # Tests may leave redelivered messages unconsumed. Release
+                    # those calls so the server can finish before lifespan shutdown.
+                    for return_event in self._message_return_events:
+                        return_event.set()
+                    while not self._call_queue.empty():
+                        self._call_queue.get_nowait()[3].set()
+                    if server:
+                        server.stop()
+
+                    shutdown_task = asyncio.ensure_future(receive())
+                    try:
+                        if serve_task:
+                            await asyncio.wait(
+                                (shutdown_task, serve_task),
+                                return_when=asyncio.FIRST_COMPLETED,
+                            )
+                            if not shutdown_task.done():
+                                await serve_task
+                                raise RuntimeError(
+                                    "Server exited without sending lifespan.shutdown"
+                                )
+                        lifespan_shutdown = await shutdown_task
+                        assert lifespan_shutdown == {"type": "lifespan.shutdown"}
+                        await send({"type": "lifespan.shutdown.complete"})
+                    finally:
+                        shutdown_task.cancel()
+                        with suppress(asyncio.CancelledError):
+                            await shutdown_task
+        finally:
+            try:
+                if serve_task:
+                    await serve_task
+            finally:
+                self._shutting_down = False
 
     async def __call__(
         self, scope: Scope, receive: AMGIReceiveCallable, send: AMGISendCallable
     ) -> None:
+        if scope["type"] == "message" and self._shutting_down:
+            return
         return_event = Event()
+        if scope["type"] == "message":
+            self._message_return_events.add(return_event)
         self._call_queue.put_nowait((scope, receive, send, return_event))
-        await return_event.wait()
+        try:
+            await return_event.wait()
+        finally:
+            self._message_return_events.discard(return_event)
 
 
 async def _app(

packages/test-utils/tests_test_utils/test_mock_app.py

@@ -0,0 +1,113 @@
+import asyncio
+from contextlib import suppress
+from unittest.mock import AsyncMock
+from unittest.mock import Mock
+
+import pytest
+from amgi_common import Lifespan
+from amgi_types import AMGIReceiveEvent
+from test_utils import MockApp
+
+
+@pytest.mark.parametrize("fails", [False, True])
+async def test_lifespan_reports_server_exit_without_shutdown(fails: bool) -> None:
+    app = MockApp()
+    stopped = asyncio.Event()
+
+    async def serve() -> None:
+        receive_queue = asyncio.Queue[AMGIReceiveEvent]()
+        receive_queue.put_nowait({"type": "lifespan.startup"})
+        app_task = asyncio.create_task(
+            app(
+                {
+                    "type": "lifespan",
+                    "amgi": {"version": "2.0", "spec_version": "2.0"},
+                    "state": {},
+                },
+                receive_queue.get,
+                AsyncMock(),
+            )
+        )
+        try:
+            await stopped.wait()
+            if fails:
+                raise ValueError("Server failed")
+        finally:
+            app_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await app_task
+
+    server = Mock(serve=AsyncMock(side_effect=serve), stop=stopped.set)
+    exception = ValueError if fails else RuntimeError
+    message = (
+        "Server failed" if fails else "Server exited without sending lifespan.shutdown"
+    )
+    with pytest.raises(exception, match=message):
+        async with app.lifespan(server=server):
+            pass
+    server.serve.assert_awaited_once()
+
+
+async def test_lifespan_shuts_down_server_when_test_fails() -> None:
+    app = MockApp()
+    stopped = asyncio.Event()
+    finished = asyncio.Event()
+
+    async def serve() -> None:
+        async with Lifespan(app):
+            await stopped.wait()
+        finished.set()
+
+    server = Mock(serve=AsyncMock(side_effect=serve), stop=stopped.set)
+    with pytest.raises(ValueError, match="Test failed"):
+        async with app.lifespan(server=server):
+            raise ValueError("Test failed")
+    assert finished.is_set()
+
+
+@pytest.mark.parametrize("message_after_stop", [False, True])
+async def test_lifespan_finishes_unconsumed_messages(message_after_stop: bool) -> None:
+    app = MockApp()
+    stopped = asyncio.Event()
+    ready = asyncio.Event()
+    finished = asyncio.Event()
+
+    async def serve() -> None:
+        async with Lifespan(app):
+            ready.set()
+            if message_after_stop:
+                await stopped.wait()
+            await app(
+                {
+                    "type": "message",
+                    "amgi": {"version": "2.0", "spec_version": "2.0"},
+                    "address": "test",
+                    "headers": [],
+                    "payload": b"test",
+                    "bindings": {},
+                    "state": {},
+                },
+                AsyncMock(),
+                AsyncMock(),
+            )
+            await stopped.wait()
+        finished.set()
+
+    server = Mock(serve=AsyncMock(side_effect=serve), stop=stopped.set)
+
+    async def run() -> None:
+        async with app.lifespan(server=server):
+            await ready.wait()
+
+    task = asyncio.create_task(run())
+    try:
+        done, _ = await asyncio.wait((task,), timeout=1)
+        assert task in done, "Shutdown blocked on an unconsumed message"
+        await task
+        assert finished.is_set()
+    finally:
+        # Release queued calls even if the regression returns, so the test can fail
+        # without leaving fixture teardown stuck as well.
+        while not app._call_queue.empty():  # pragma: no cover
+            app._call_queue.get_nowait()[3].set()
+        await task

pyproject.toml

@@ -42,8 +42,9 @@ workspace.members = [
 
 [tool.pytest]
 ini_options.asyncio_mode = "auto"
-ini_options.timeout = 10
-ini_options.timeout_func_only = true
+ini_options.timeout = 60
+ini_options.timeout_func_only = false
+ini_options.timeout_method = "thread"
 ini_options.filterwarnings = [
   "ignore:^The wait_for_logs function with string or callable predicates is deprecated:DeprecationWarning",
   "ignore:^The @wait_container_is_ready decorator is deprecated:DeprecationWarning",

tox.ini

@@ -5,6 +5,7 @@ env_list =
     py314-asyncfast-pydantic212
     py313-asyncfast-pydantic2{8-12}
     clean
+    py3{10-14}-test-utils
     py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-pulsar, amgi-redis, amgi-sqs-event-source-mapping, asyncfast-avro, asyncfast-cli, pytest-amgi}
     py3{10-12}-asyncfast-pydantic2{0-12}
     py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-pulsar, amgi-redis, amgi-sqs-event-source-mapping, amgi-types, asyncfast, asyncfast-avro, asyncfast-cli, pytest-amgi}-import
@@ -48,6 +49,12 @@ commands =
 labels = clean
 uv_sync_flags = --all-packages
 
+[testenv:py3{10-14}-test-utils]
+commands =
+    {[testenv]commands} packages/test-utils
+labels = test-utils
+uv_sync_flags = --package=test-utils
+
 [testenv:py3{10-12}-asyncfast-pydantic2{0-12}]
 commands_pre =
     {[pydantic2]commands_pre}