Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion examples/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ def build_server(port: str = "127.0.0.1:0") -> tuple[grpc.Server, int, Container
return server, bound_port, container


if __name__ == "__main__": # pragma: no cover
if __name__ == "__main__": # pragma: no cover - script entry point; tests call build_server() directly
_server, _port, _container = build_server("[::]:50051")
print(f"Serving on port {_port}; Ctrl+C to stop") # noqa: T201
try:
Expand Down
7 changes: 2 additions & 5 deletions tests/test_aio.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,11 +142,8 @@ async def test_aio_app_finalizer_runs_on_root_close() -> None:
stub = greeter_pb2_grpc.GreeterStub(channel)
await stub.SayHello(HelloRequest(name="x"))
await server.stop(0)
# coveragepy #2124: on Python 3.11 the tracer can lose 1-2 lines right after awaiting
# a coroutine that cancels tasks internally, which server.stop does to its own
# background tasks. Excluded (not skipped) so the flake can't fail --cov-fail-under.
await container.close_async() # pragma: no cover
assert app_teardowns == ["app-closed"] # pragma: no cover
await container.close_async() # pragma: no cover - coveragepy #2124: 3.11 can drop lines after server.stop
assert app_teardowns == ["app-closed"] # pragma: no cover - coveragepy #2124: 3.11 can drop lines after server.stop


async def test_wrap_unary_aio_resets_context_var_when_close_raises() -> None:
Expand Down
2 changes: 1 addition & 1 deletion tests/test_inject.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ def test_inject_raises_without_interceptor() -> None:
def method(
_self: object, _request: str, _context: object, _app_res: typing.Annotated[AppResource, FromDI(AppResource)]
) -> None:
pass # pragma: no cover
pass # pragma: no cover - never runs; inject raises before the call without DIInterceptor

def _call() -> None:
with pytest.raises(RuntimeError, match="DIInterceptor"):
Expand Down
Loading