diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..47649e0 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,13 @@ +version: 2 +updates: + - package-ecosystem: pip + directory: / + schedule: + interval: weekly + groups: + python-development: + dependency-type: development + - package-ecosystem: github-actions + directory: / + schedule: + interval: weekly diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0975347..828f739 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,9 +1,17 @@ name: CI on: [push, pull_request] +permissions: + contents: read + +concurrency: + group: ci-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + jobs: checks: runs-on: ubuntu-latest + timeout-minutes: 20 strategy: matrix: python-version: ['3.13', '3.14'] @@ -16,6 +24,7 @@ jobs: build: runs-on: ubuntu-latest + timeout-minutes: 10 steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 @@ -30,10 +39,34 @@ jobs: dist/*.tar.gz if-no-files-found: error + framework-compatibility: + name: Frameworks ${{ matrix.frameworks }} / Python ${{ matrix.python-version }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.13', '3.14'] + frameworks: [minimum, current] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: {python-version: '${{ matrix.python-version }}'} + - name: Install adapter test dependencies + shell: bash + run: | + python -m pip install -e . pytest pytest-asyncio + if [ '${{ matrix.frameworks }}' = minimum ]; then + python -m pip install 'fastapi>=0.110,<0.111' 'starlette>=0.37,<0.38' + else + python -m pip install 'fastapi>=0.110' 'starlette>=0.37' + fi + - run: python -m pytest tests/test_adapters.py + installation: name: Python ${{ matrix.python-version }} / ${{ matrix.installation }} needs: build runs-on: ubuntu-latest + timeout-minutes: 20 services: redis: image: redis:7-alpine diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..8f3a469 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,54 @@ +name: Release + +on: + push: + tags: ['v*'] + +permissions: + contents: read + +jobs: + build: + if: github.repository == 'Forebase/Eventful' + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + id-token: write + attestations: write + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.13' + - run: python -m pip install build + - name: Require tag and package versions to match + shell: bash + run: | + package_version=$(python -c 'exec(open("src/eventful/__version__.py").read()); print(__version__)') + test "v${package_version}" = "${GITHUB_REF_NAME}" + - run: python -m build + - uses: actions/attest-build-provenance@v2 + with: + subject-path: 'dist/*' + - uses: actions/upload-artifact@v4 + with: + name: release-distributions + path: dist/* + if-no-files-found: error + + publish: + needs: build + runs-on: ubuntu-latest + timeout-minutes: 10 + environment: + name: pypi + url: https://pypi.org/project/eventful/ + permissions: + id-token: write + steps: + - uses: actions/download-artifact@v4 + with: + name: release-distributions + path: dist + - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml new file mode 100644 index 0000000..6c81d0c --- /dev/null +++ b/.github/workflows/security.yml @@ -0,0 +1,41 @@ +name: Security + +on: + push: + branches: [main, dev/alpha] + pull_request: + schedule: + - cron: '17 4 * * 1' + +permissions: + contents: read + +concurrency: + group: security-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + dependency-audit: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.13' + cache: pip + - run: python -m pip install '.[all]' pip-audit + - run: python -m pip_audit + + codeql: + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + security-events: write + steps: + - uses: actions/checkout@v4 + - uses: github/codeql-action/init@v3 + with: + languages: python + - uses: github/codeql-action/analyze@v3 diff --git a/CHANGELOG.md b/CHANGELOG.md index cc724c7..6025c5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,15 @@ # Changelog -## 0.2.0 - unreleased +## 0.3.0 - unreleased + +- Added deterministic ASGI ownership and shutdown behavior, including failed startup. +- Added release gates for supported Python versions, built distributions, optional + integrations, dependency vulnerabilities, and static security analysis. +- Added a tag-only trusted-publishing workflow with artifact attestations. +- Bounded optional dependency compatibility ranges and documented production + guarantees, limitations, support policy, and operator responsibilities. + +## 0.2.0 - 2026-08-23 - Repaired package imports and packaging metadata. - Established provisional v1 topology and contracts. diff --git a/DEPRECATION.md b/DEPRECATION.md index 351eaae..bb4debc 100644 --- a/DEPRECATION.md +++ b/DEPRECATION.md @@ -1,3 +1,7 @@ # Deprecation policy -Provisional APIs should warn for at least one minor release when practical. Experimental APIs may change faster but must be recorded in the changelog. +Before 1.0, public API removals warn for at least one minor release when practical +and are recorded in the changelog. A symbol documented as experimental may change +faster, but the change must still be documented. Patch releases do not knowingly +remove public APIs. Security fixes may override the notice period when retaining an +API would leave users exposed. diff --git a/README.md b/README.md index 1de63d0..23dd6b5 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Eventful -Eventful 0.2.0 is a pre-alpha foundation for a Python event toolkit. The implemented public facade is a local in-memory dispatcher (`Event`, `EventBus`, `InMemoryBus`, `listener`, `emit_sync`, `emit_async`, and the compatibility helper `emit`). v1-oriented packages provide provisional contracts for brokers, durable streams, codecs, stores, middleware, plugins, configuration, schemas, and observability. +Eventful 0.3.0 is a typed event toolkit for Python 3.13 and 3.14. Its supported core is the local dispatcher (`Event`, `EventBus`, `InMemoryBus`, `listener`, `emit_sync`, `emit_async`, and the compatibility helper `emit`), with tested Redis, PostgreSQL, file-persistence, FastAPI, and Starlette integrations. ## Install @@ -35,7 +35,7 @@ assert bus.emit_sync(Event(type="user.created", payload="Ada")) == ["hello Ada"] ## API status -No API is stable before 1.0. The root facade is preserved for 0.1 compatibility and treated as provisional. Experimental packages are importable for architecture work but should not be treated as production integrations. +No API is frozen before 1.0. The root facade remains backward compatible within the documented deprecation policy. Production use must be limited to the guarantees and deployment models in [`docs/production-readiness.md`](docs/production-readiness.md). See `docs/index.md` for the documentation map and `docs/work-register.md` for deferred work. diff --git a/SECURITY.md b/SECURITY.md index 55ca524..9292114 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -1,3 +1,13 @@ # Security policy -Report vulnerabilities privately to the maintainers. Do not file public issues for exploitable behavior until triage is complete. +## Supported versions + +Security fixes are made on the latest released minor version. Users should upgrade +to the newest patch before reporting a suspected vulnerability. + +## Reporting + +Use GitHub's **Security → Report a vulnerability** flow for this repository. Do +not file a public issue for exploitable behavior. Include affected versions, a +minimal reproduction, impact, and any known mitigation. Maintainers will confirm +receipt within five working days and coordinate disclosure after a fix is ready. diff --git a/docs/frameworks.md b/docs/frameworks.md index fc9546c..2776716 100644 --- a/docs/frameworks.md +++ b/docs/frameworks.md @@ -66,11 +66,17 @@ fallback is intentional. - An injected `bus=` remains caller-owned by default. - A bus created by `bus_factory=` or by the middleware default is adapter-owned. - `close_on_shutdown=True` opts an injected bus into adapter ownership. -- On `lifespan.shutdown.complete`, the adapter awaits `bus.close()` when the owned - bus exposes a synchronous or asynchronous close method. -- Cleanup completes before shutdown success is forwarded to the ASGI server. -- Startup failures do not claim that shutdown cleanup occurred; applications should - manage resources created outside the adapter in their own lifespan handler. +- On `lifespan.shutdown.complete` or `lifespan.shutdown.failed`, the adapter awaits + `bus.close()` when the owned bus exposes a synchronous or asynchronous close + method. +- Cleanup completes before the terminal shutdown result is forwarded to the ASGI + server, including when terminal lifespan calls overlap. +- A failed startup (either `lifespan.startup.failed` or an exception escaping the + lifespan application) also closes an adapter-owned bus. Caller-owned buses remain + untouched in every failure path. +- Cleanup is idempotent: repeated startup/shutdown cycles against the same middleware + instance do not close a bus more than once. Create a new application/middleware + instance to obtain a fresh adapter-owned bus after shutdown. The local `EventBus` has no resources to close. The generic close behavior exists for application bus subclasses that coordinate transports, stores, or plugins. @@ -81,3 +87,22 @@ The bus itself is application-scoped. Request state isolates access paths, not b registrations. Use separate application instances or a custom `bus_factory` when tests or tenants require distinct registration state. `state_key=` supports coexistence with another request-state convention. + +Each middleware instance using the default bus or `bus_factory=` creates its own bus, +so those concurrently running application instances are isolated. Injected buses +have caller-defined scope: injecting the same `bus=` into multiple applications +intentionally shares registrations and delivery state, and isolation is the caller's +responsibility. This is process-local isolation only: pre-fork and multi-worker +deployments create one adapter-owned bus per worker. Eventful does not coordinate +registrations or delivery between workers; inject a caller-managed transport-backed +bus when cross-process behavior is required. Do not share one adapter-owned +middleware instance between event loops. + +## Supported versions + +The declared dependency floors are FastAPI 0.110 and Starlette 0.37 on Eventful's +supported Python versions. CI exercises those minor-version ranges through the +public ASGI and dependency APIs, and a separate matrix leg resolves the latest +available releases. Versions between the floor and latest tested releases are +expected to work; compatibility with a future major release is not promised until +that release is separately validated. diff --git a/docs/production-readiness.md b/docs/production-readiness.md new file mode 100644 index 0000000..dfc997c --- /dev/null +++ b/docs/production-readiness.md @@ -0,0 +1,48 @@ +# Production readiness + +Eventful 0.3 is production-capable within the boundaries below. The project is +still pre-1.0: documented deprecations precede compatibility breaks, but the API +is not yet permanently frozen. + +## Supported deployment envelope + +| Component | Supported use | Delivery/durability boundary | +| --- | --- | --- | +| `EventBus` / `InMemoryBus` | One Python process; concurrent threads; sync or async listeners | In-memory, no crash recovery | +| `RedisTransport` | Cross-process live fan-out with Redis 5–6 clients | At-most-once Pub/Sub; no replay | +| `PostgresPersistence` | Durable append/replay with PostgreSQL 16 | Transactional records; caller owns retry and retention policy | +| `FilePersistence` | One process with a local filesystem | Flushes each record; no multi-process coordination or fsync guarantee | +| FastAPI / Starlette adapters | Supported framework extras and ASGI lifespan | Adapter-created buses are isolated and closed exactly once | + +Python 3.13 and 3.14 are tested from both wheels and source distributions. Core +installation does not import or install optional integration dependencies. + +## Release gates + +Every change must pass linting, strict contract typing, the complete unit suite, +service-backed Redis and PostgreSQL tests, documentation validation, wheel/sdist +builds, and clean-install smoke tests on both supported Python versions. Security +automation performs dependency auditing and CodeQL analysis. Version tags are +published only through the protected `pypi` GitHub environment and produce signed +artifact attestations. + +## Operator responsibilities + +- Pin Eventful and integration dependencies in application lock files. +- Set explicit connection URLs; secure Redis/PostgreSQL with network policy, + authentication, TLS, backup, retention, and monitoring appropriate to the system. +- Install an application error handler and observability provider. Alert on + listener, serialization, transport, and persistence failures. +- Use PostgreSQL (or another durable `EventStore`) when loss or replay matters. +- Exercise shutdown, dependency outage, retry, and restore procedures before launch. +- Treat event payloads as application data: Eventful does not encrypt, redact, + authorize, or classify them. + +## Explicit non-guarantees + +Eventful does not provide exactly-once delivery, distributed transactions, +multi-process file locking, schema evolution, dead-letter queues, broker access +control, or automatic retry policy. Applications requiring those properties must +compose them at their own boundary or use a transport/store that supplies them. + +Security reports follow `SECURITY.md`; API removals follow `DEPRECATION.md`. diff --git a/docs/work-register.md b/docs/work-register.md index c127b39..9924b5d 100644 --- a/docs/work-register.md +++ b/docs/work-register.md @@ -4,7 +4,7 @@ All intentional incompleteness must use an annotation listed in `docs/annotation | Module | Status | Deferred work | | --- | --- | --- | -| `eventful.adapters` | provisional ASGI lifecycle | Validate multi-worker ownership and framework-version compatibility (Issue: Forebase/Eventful#1). | +| `eventful.adapters` | validated in-process ASGI lifecycle | Multi-worker delivery remains deployment-managed; major framework versions require separate validation (Issue: Forebase/Eventful#1). | | `eventful.transports.redis` | provisional, service-validated Pub/Sub | Connection loss/restart gaps, cancellation cleanup, malformed input, concurrent publishing, and shutdown limits validated; delivery remains live at-most-once with no Eventful retry/replay or durability. Redis Streams remains deferred (Forebase/Eventful#2). | | `eventful.persistence.postgres_persistence` | provisional durable store | Validate migration upgrades, retention, replication, and operational load behavior (Issue: Forebase/Eventful#3). | | `eventful.contracts` | provisional, reference-validated | Validate async lifecycle, delivery, and durability semantics against real Redis/PostgreSQL integrations (Issue: Forebase/Eventful#4). | diff --git a/mkdocs.yml b/mkdocs.yml index 26f0af1..c600284 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -10,6 +10,7 @@ nav: - Framework Adapters: frameworks.md - Extension Points: extensions.md - Quickstart: quickstart.md + - Production Readiness: production-readiness.md - Runnable Examples: examples.md - Work Register: work-register.md - Annotation Policy: annotation-policy.md diff --git a/pyproject.toml b/pyproject.toml index f91af55..2b3f25e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,14 +5,14 @@ build-backend = "setuptools.build_meta" [project] name = "eventful" dynamic = ["version"] -description = "Provisional event dispatch contracts and local in-memory event bus for Python applications" +description = "Typed event dispatch, transport, and persistence building blocks for Python applications" readme = "README.md" requires-python = ">=3.13,<3.15" license = "MIT" authors = [{name = "Eventful maintainers", email = "team@eventful.org"}] keywords = ["events", "pubsub", "event-bus", "dispatch"] classifiers = [ - "Development Status :: 2 - Pre-Alpha", + "Development Status :: 4 - Beta", "Intended Audience :: Developers", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.13", @@ -21,12 +21,18 @@ classifiers = [ ] dependencies = [] +[project.urls] +Homepage = "https://github.com/Forebase/Eventful" +Documentation = "https://github.com/Forebase/Eventful/tree/main/docs" +Issues = "https://github.com/Forebase/Eventful/issues" +Changelog = "https://github.com/Forebase/Eventful/blob/main/CHANGELOG.md" + [project.optional-dependencies] -redis = ["redis>=5"] -postgres = ["asyncpg>=0.29"] +redis = ["redis>=5,<7"] +postgres = ["asyncpg>=0.29,<1"] file = [] # Dependency-free UTF-8 JSON Lines persistence backend. -fastapi = ["fastapi>=0.110"] -starlette = ["starlette>=0.37"] +fastapi = ["fastapi>=0.110,<1"] +starlette = ["starlette>=0.37,<1"] test = ["asyncpg>=0.29", "build>=1.2", "fastapi>=0.110", "pytest>=8.2", "pytest-asyncio>=0.23", "pytest-cov>=5", "redis>=5", "starlette>=0.37"] docs = ["mkdocs>=1.6"] dev = ["asyncpg>=0.29", "build>=1.2", "fastapi>=0.110", "mkdocs>=1.6", "mypy>=1.10", "pre-commit>=3.7", "pytest>=8.2", "pytest-asyncio>=0.23", "pytest-cov>=5", "redis>=5", "ruff>=0.5", "starlette>=0.37"] diff --git a/scripts/check_annotations.py b/scripts/check_annotations.py index 98ee6de..3b76b84 100644 --- a/scripts/check_annotations.py +++ b/scripts/check_annotations.py @@ -7,7 +7,12 @@ annotation = re.compile(rf"\b({ALLOWED})\b:(.*)") forbidden = re.compile(r"\b(TBD|XXX)\b") failed: list[str] = [] -for path in [p for p in Path('.').rglob('*') if p.is_file() and '.git' not in p.parts and p.resolve() != Path(__file__).resolve()]: +roots = [Path("src"), Path("tests"), Path("docs"), Path(".github")] +paths = [Path("README.md"), Path("CHANGELOG.md"), Path("pyproject.toml")] +paths.extend(path for root in roots for path in root.rglob("*") if path.is_file()) +for path in paths: + if path.resolve() == Path(__file__).resolve(): + continue if path.suffix not in {'.py', '.md', '.toml', '.yml', '.yaml'}: continue for lineno, line in enumerate(path.read_text(errors='ignore').splitlines(), 1): diff --git a/scripts/check_quality.py b/scripts/check_quality.py index a2020a2..7302c20 100644 --- a/scripts/check_quality.py +++ b/scripts/check_quality.py @@ -18,7 +18,13 @@ "src/eventful", "tests/contract_typing.py", ), - (sys.executable, "-m", "pytest", "--cov=eventful"), + ( + sys.executable, + "-m", + "pytest", + "--cov=eventful", + "--cov-fail-under=75", + ), (sys.executable, "scripts/check_annotations.py"), (sys.executable, "scripts/check_docstrings.py"), (sys.executable, "-m", "build"), diff --git a/src/eventful/__version__.py b/src/eventful/__version__.py index 19c2727..810826b 100644 --- a/src/eventful/__version__.py +++ b/src/eventful/__version__.py @@ -1,2 +1,2 @@ """Single version source for Eventful.""" -__version__ = "0.2.0" +__version__ = "0.3.0" diff --git a/src/eventful/adapters/starlette.py b/src/eventful/adapters/starlette.py index 5f6dd00..42a5870 100644 --- a/src/eventful/adapters/starlette.py +++ b/src/eventful/adapters/starlette.py @@ -50,7 +50,7 @@ def __init__( self._close_lock = asyncio.Lock() async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: - """Attach state and wait for application shutdown before closing resources.""" + """Attach state and close owned resources when a lifespan terminates.""" if scope["type"] in {"http", "websocket"}: scope.setdefault("state", {})[self.state_key] = self.bus @@ -59,12 +59,28 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None return async def lifespan_send(message: dict[str, Any]) -> None: - """Close owned resources before reporting successful shutdown.""" - if message["type"] == "lifespan.shutdown.complete": + """Close resources before reporting shutdown or failed startup.""" + if message["type"] in { + "lifespan.startup.failed", + "lifespan.shutdown.complete", + "lifespan.shutdown.failed", + }: await self.close() await send(message) - await self.app(scope, receive, lifespan_send) + try: + await self.app(scope, receive, lifespan_send) + except BaseException as application_error: + # A lifespan exception may prevent the application from sending either + # terminal message. Do not strand a bus that this adapter created. + try: + await self.close() + except BaseException as cleanup_error: + raise BaseExceptionGroup( + "lifespan application and Eventful cleanup failed", + [application_error, cleanup_error], + ) from None + raise async def close(self) -> None: """Close an owned/opted-in bus once when it exposes `close()`.""" @@ -72,14 +88,16 @@ async def close(self) -> None: if self._closed: return self._closed = True - if not self.close_on_shutdown: - return - close = getattr(self.bus, "close", None) - if close is None: - return - result = close() - if inspect.isawaitable(result): - await result + if not self.close_on_shutdown: + return + close = getattr(self.bus, "close", None) + if close is None: + return + result = close() + if inspect.isawaitable(result): + # Keep the lock until cleanup finishes so a concurrent terminal + # lifespan message cannot be forwarded ahead of resource cleanup. + await result def request_event_bus( diff --git a/tests/test_adapters.py b/tests/test_adapters.py index be708f1..9c12cd9 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -8,6 +8,9 @@ from typing import Any from fastapi import Depends, FastAPI, Request +from starlette.applications import Starlette +from starlette.requests import Request as StarletteRequest +from starlette.responses import JSONResponse from eventful import EventBus from eventful.adapters.fastapi import event_bus_dependency, install_eventful @@ -21,10 +24,12 @@ def __init__(self) -> None: """Create an open local bus.""" super().__init__() self.closed = False + self.close_calls = 0 async def close(self) -> None: """Mark the test bus closed.""" self.closed = True + self.close_calls += 1 async def terminal_app(scope: dict[str, Any], receive: Any, send: Any) -> None: @@ -139,6 +144,154 @@ def test_lifespan_respects_external_and_explicit_ownership() -> None: assert owned.closed is True +def test_repeated_lifespan_shutdown_closes_owned_bus_once() -> None: + """Make repeated server lifespan cycles safe and cleanup idempotent.""" + bus = CloseableBus() + middleware = EventfulMiddleware(terminal_app, bus_factory=lambda: bus) + + assert asyncio.run(invoke_lifespan(middleware))[-1] == "lifespan.shutdown.complete" + assert asyncio.run(invoke_lifespan(middleware))[-1] == "lifespan.shutdown.complete" + assert bus.close_calls == 1 + + +def test_concurrent_application_instances_have_distinct_owned_buses() -> None: + """Never share implicitly created buses between application instances.""" + first = EventfulMiddleware(terminal_app, bus_factory=CloseableBus) + second = EventfulMiddleware(terminal_app, bus_factory=CloseableBus) + + async def exercise() -> None: + await asyncio.gather(invoke_lifespan(first), invoke_lifespan(second)) + + asyncio.run(exercise()) + assert first.bus is not second.bus + assert first.bus.closed is True + assert second.bus.closed is True + + +def test_injected_bus_scope_is_shared_by_caller_choice() -> None: + """Attach one caller-owned bus to both apps when it is injected twice.""" + bus = CloseableBus() + first = EventfulMiddleware(terminal_app, bus=bus) + second = EventfulMiddleware(terminal_app, bus=bus) + + async def exercise() -> tuple[dict[str, Any], dict[str, Any]]: + return await asyncio.gather(invoke_http(first), invoke_http(second)) + + first_state, second_state = asyncio.run(exercise()) + assert first_state["eventful_bus"] is bus + assert second_state["eventful_bus"] is bus + + +def test_owned_bus_is_cleaned_up_after_startup_failure() -> None: + """Release adapter resources when startup fails or raises.""" + async def failed_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + await send({"type": "lifespan.startup.failed", "message": "nope"}) + + failed_bus = CloseableBus() + failed = EventfulMiddleware(failed_app, bus_factory=lambda: failed_bus) + assert asyncio.run(invoke_lifespan(failed)) == ["lifespan.startup.failed"] + assert failed_bus.close_calls == 1 + + async def raising_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + raise RuntimeError("startup exploded") + + raised_bus = CloseableBus() + raised = EventfulMiddleware(raising_app, bus_factory=lambda: raised_bus) + try: + asyncio.run(invoke_lifespan(raised)) + except RuntimeError as exc: + assert str(exc) == "startup exploded" + else: + raise AssertionError("startup exception should propagate") + assert raised_bus.close_calls == 1 + + +def test_startup_and_cleanup_failures_are_both_preserved() -> None: + """Report cleanup failure without replacing the startup root cause.""" + class FailingCloseBus(EventBus): + async def close(self) -> None: + raise OSError("cleanup exploded") + + async def raising_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + raise RuntimeError("startup exploded") + + middleware = EventfulMiddleware(raising_app, bus_factory=FailingCloseBus) + try: + asyncio.run(invoke_lifespan(middleware)) + except BaseExceptionGroup as exc: + assert [str(error) for error in exc.exceptions] == [ + "startup exploded", + "cleanup exploded", + ] + assert isinstance(exc.exceptions[0], RuntimeError) + assert isinstance(exc.exceptions[1], OSError) + else: + raise AssertionError("both lifespan failures should propagate") + + +def test_startup_failure_preserves_application_owned_bus() -> None: + """Do not clean up an injected bus merely because application startup fails.""" + async def failed_app(scope: dict[str, Any], receive: Any, send: Any) -> None: + await receive() + await send({"type": "lifespan.startup.failed"}) + + bus = CloseableBus() + middleware = EventfulMiddleware(failed_app, bus=bus) + asyncio.run(invoke_lifespan(middleware)) + assert bus.close_calls == 0 + + +def test_owned_bus_is_cleaned_up_after_shutdown_failure() -> None: + """Treat a failed shutdown message as a terminal lifespan outcome.""" + async def failed_shutdown( + scope: dict[str, Any], receive: Any, send: Any + ) -> None: + await receive() + await send({"type": "lifespan.startup.complete"}) + await receive() + await send({"type": "lifespan.shutdown.failed", "message": "nope"}) + + bus = CloseableBus() + middleware = EventfulMiddleware(failed_shutdown, bus_factory=lambda: bus) + assert asyncio.run(invoke_lifespan(middleware)) == [ + "lifespan.startup.complete", + "lifespan.shutdown.failed", + ] + assert bus.close_calls == 1 + + +def test_concurrent_close_waits_for_cleanup() -> None: + """Do not let a second close return while the first cleanup is in progress.""" + class BlockingCloseBus(EventBus): + def __init__(self) -> None: + super().__init__() + self.started = asyncio.Event() + self.release = asyncio.Event() + self.close_calls = 0 + + async def close(self) -> None: + self.close_calls += 1 + self.started.set() + await self.release.wait() + + async def exercise() -> None: + bus = BlockingCloseBus() + middleware = EventfulMiddleware(terminal_app, bus_factory=lambda: bus) + first = asyncio.create_task(middleware.close()) + await bus.started.wait() + second = asyncio.create_task(middleware.close()) + await asyncio.sleep(0) + assert second.done() is False + bus.release.set() + await asyncio.gather(first, second) + assert bus.close_calls == 1 + + asyncio.run(exercise()) + + def test_fastapi_installs_middleware_and_dependency_uses_request_state() -> None: """Use FastAPI's middleware registry and Request annotation contract.""" app = FastAPI() @@ -157,6 +310,19 @@ def bus_endpoint(resolved: EventBus = Depends(dependency)): assert asyncio.run(invoke_fastapi(app, "/bus")) == (200, {"same": True}) +def test_supported_starlette_application_uses_request_state() -> None: + """Exercise the public middleware API on the supported Starlette release.""" + app = Starlette() + bus = CloseableBus() + app.add_middleware(EventfulMiddleware, bus=bus) + + async def endpoint(request: StarletteRequest) -> JSONResponse: + return JSONResponse({"same": request_event_bus(request) is bus}) + + app.add_route("/bus", endpoint) + assert asyncio.run(invoke_fastapi(app, "/bus")) == (200, {"same": True}) + + def test_request_event_bus_requires_middleware_state() -> None: """Avoid silently leaking the process-global bus into unconfigured requests.""" request = SimpleNamespace(state=SimpleNamespace()) diff --git a/tests/test_import_surface.py b/tests/test_import_surface.py index 5ee2805..a0c0a3b 100644 --- a/tests/test_import_surface.py +++ b/tests/test_import_surface.py @@ -7,7 +7,7 @@ def test_minimal_root_imports() -> None: import eventful - assert eventful.__version__ == "0.2.0" + assert eventful.__version__ == "0.3.0" assert "emit_sync" in eventful.__all__ assert "emit_async" in eventful.__all__ assert "AsyncDispatchRequired" in eventful.__all__