Add singleton support with dependency injection

Singletons are expensive resources (e.g. VM pools, resource monitors)
shared across test classes. They implement the async context manager
protocol and are declared on test classes and other singletons via
the singleton descriptor.

Dependencies between singletons are resolved automatically.
SingletonManager handles lifecycle via TaskGroup, tearing down in
reverse creation order.

Co-developed-by: Claude <claude@anthropic.com>
Signed-off-by: Daan De Meyer <daan@amutable.com>

Author: daandemeyer

AGENTS.md

@@ -20,7 +20,7 @@ Barrage is a concurrent async test framework for Python 3.12+ with zero external
 
 - **`runner.py`** — Execution engine. Runs test classes concurrently as asyncio tasks. Uses `contextvars.ContextVar` for per-task stdout/stderr capture so concurrent tests get isolated output.
 - **`case.py`** — `AsyncTestCase` (base class, all hooks/tests are async) and `MonitoredTestCase` (adds background task crash monitoring with auto-cancel/skip).
-- **`singleton.py`** — Singleton lifecycle with dependency injection. `Singleton` base class (async context manager), `singleton[T]` descriptor for declaring dependencies on test classes, `SingletonManager` resolves dependencies from `__init__` type annotations and tears down in reverse creation order.
+- **`singleton.py`** — Singleton lifecycle with dependency injection. `Singleton` base class (async context manager), `singleton[T]` descriptor for declaring dependencies on both test classes and singleton classes, `SingletonManager` resolves dependencies recursively and tears down in reverse creation order.
 - **`discovery.py`** — Test discovery. Supports directories, files, and `File::Class::method` selectors.
 - **`result.py`** — `TestOutcome` dataclass and `AsyncTestResult` collector (asyncio.Lock-protected).
 - **`subprocess.py`** — Async `spawn()` (context manager for long-lived processes) and `run()` helpers with PTY-aware output relaying through the capture system.

README.md

@@ -212,3 +212,70 @@ Because `spawn()` is an async context manager, `monitor_async_context()`
 enters it in a background task and monitors that task. The subprocess is
 killed and cleaned up when the test class finishes or when the background
 task is cancelled due to a crash.
+
+## Singletons
+
+A *singleton* is a resource that is expensive to create and tear down
+(e.g. a pool of virtual machines, a database connection pool) and
+should be shared across many test classes.
+
+Singletons are declared as classes that inherit from `Singleton` and
+implement the async context manager protocol. They are attached to
+test classes using the `singleton` descriptor. The framework creates
+each singleton once, injects it before `setUpClass`, and tears
+everything down in reverse order when the test session ends.
+
+```python
+from typing import Self
+
+from barrage import AsyncTestCase, Singleton, singleton
+
+class VMManager(Singleton):
+    async def __aenter__(self) -> Self:
+        self.pool = await create_vm_pool()
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        await self.pool.shutdown()
+
+class MyTests(AsyncTestCase):
+    manager = singleton(VMManager)
+
+    async def test_something(self) -> None:
+        vm = await self.manager.acquire()
+        result = await vm.run("uname -r")
+        self.assertIn("6.", result)
+```
+
+### Default `__aexit__` behaviour
+
+The base `Singleton` class provides a default `__aexit__` that blocks
+forever (via an unresolved `asyncio.Future`), keeping the singleton's
+background task alive until the framework cancels it during teardown.
+Override `__aexit__` when you need custom cleanup logic.
+
+### Dependency injection
+
+Dependencies between singletons are declared using the same
+`singleton` descriptor. If singleton A has a `singleton(B)` descriptor,
+B is created and injected first. Circular dependencies are detected
+and raise `RuntimeError`.
+
+```python
+class ResourceMonitor(Singleton):
+    async def __aenter__(self) -> Self:
+        ...
+
+class VMManager(Singleton):
+    monitor = singleton(ResourceMonitor)
+
+    async def __aenter__(self) -> Self:
+        ...
+
+class MyTests(AsyncTestCase):
+    monitor = singleton(ResourceMonitor)
+    manager = singleton(VMManager)  # ResourceMonitor is created first
+```
+
+Multiple test classes that reference the same singleton class share a
+single instance.

barrage/__init__.py

@@ -8,9 +8,22 @@
 basis via ``__init_subclass__``, and the overall level of concurrency
 is tuneable via the runner.
 
+Singletons (expensive resources shared across test classes) are
+supported via the :class:`singleton` descriptor and
+:class:`SingletonManager`::
+
+    from barrage import AsyncTestCase, singleton
+
+    class MyTests(AsyncTestCase):
+        manager = singleton(VMManager)
+
+        async def test_example(self) -> None:
+            vm = await self.manager.acquire()
+            self.assertIsNotNone(vm)
+
 Quick start::
 
-    from barrage import AsyncTestCase, AsyncTestRunner
+    from barrage import AsyncTestRunner, AsyncTestCase
 
     class MyTests(AsyncTestCase):
         async def setUp(self) -> None:
@@ -29,6 +42,7 @@ async def test_example(self) -> None:
 from barrage.discovery import discover, discover_module, resolve_tests
 from barrage.result import AsyncTestResult, Outcome, TestOutcome
 from barrage.runner import AsyncTestRunner, AsyncTestSuite
+from barrage.singleton import Singleton, SingletonManager, singleton
 from barrage.subprocess import (
     DEVNULL,
     PIPE,
@@ -44,6 +58,10 @@ async def test_example(self) -> None:
     # Core
     "AsyncTestCase",
     "MonitoredTestCase",
+    # Singletons
+    "Singleton",
+    "singleton",
+    "SingletonManager",
     # Runner
     "AsyncTestRunner",
     "AsyncTestSuite",

barrage/runner.py

@@ -22,6 +22,7 @@
     style,
 )
 from barrage.result import AsyncTestResult, Outcome, TestOutcome
+from barrage.singleton import SingletonManager, discover_singletons
 
 
 def _collect_test_methods(cls: type[AsyncTestCase]) -> list[str]:
@@ -1035,6 +1036,9 @@ async def run_suite_async(self, suite: AsyncTestSuite) -> AsyncTestResult:
         result = AsyncTestResult()
         entries = suite.entries
 
+        # ── singleton injection ───────────────────────────────────────
+        has_singletons = any(discover_singletons(cls) for cls, _methods in entries)
+
         # Create a stop event when failfast is enabled.  It is shared
         # across all test tasks so that a failure anywhere signals the
         # rest to stop.
@@ -1117,10 +1121,16 @@ async def run_suite_async(self, suite: AsyncTestSuite) -> AsyncTestResult:
 
         result.start_time = time.monotonic()
 
-        try:
-            if progress is not None:
-                await progress.start()
+        async def run_tests() -> None:
+            """Run all test classes, with optional singleton injection."""
+            if has_singletons:
+                async with SingletonManager() as sm:
+                    await sm.inject(entries)
+                    await _run_all_classes()
+            else:
+                await _run_all_classes()
 
+        async def _run_all_classes() -> None:
             if self.interactive:
                 # In interactive mode run classes sequentially so that
                 # their output is not interleaved.
@@ -1159,6 +1169,11 @@ async def run_suite_async(self, suite: AsyncTestSuite) -> AsyncTestResult:
                     for cls, methods in entries
                 ]
                 await asyncio.gather(*class_tasks)
+
+        try:
+            if progress is not None:
+                await progress.start()
+            await run_tests()
         finally:
             if progress is not None:
                 await progress.stop()
@@ -1170,6 +1185,7 @@ async def run_suite_async(self, suite: AsyncTestSuite) -> AsyncTestResult:
                 devnull.close()
 
         result.end_time = time.monotonic()
+
         return result
 
     def run_classes(self, *classes: type[AsyncTestCase]) -> AsyncTestResult: