|
| 1 | +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. |
| 2 | +# SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +"""Generated async stubs retain native gRPC transport behavior.""" |
| 5 | + |
| 6 | +import asyncio |
| 7 | +from collections.abc import AsyncGenerator |
| 8 | +from contextlib import asynccontextmanager |
| 9 | + |
| 10 | +import grpc |
| 11 | +import grpc.aio |
| 12 | +import pytest |
| 13 | + |
| 14 | +from openshell._proto import datamodel_pb2 as dm |
| 15 | +from openshell._proto import openshell_pb2 as pb |
| 16 | +from openshell._proto import openshell_pb2_grpc as rpc |
| 17 | + |
| 18 | + |
| 19 | +class Gateway(rpc.OpenShellServicer): |
| 20 | + def __init__(self) -> None: |
| 21 | + self.started = asyncio.Event() |
| 22 | + self.cancelled = asyncio.Event() |
| 23 | + |
| 24 | + async def GetProvider( |
| 25 | + self, request: pb.GetProviderRequest, context: grpc.aio.ServicerContext |
| 26 | + ) -> pb.ProviderResponse: |
| 27 | + assert ("authorization", "Bearer test-token") in ( |
| 28 | + context.invocation_metadata() or () |
| 29 | + ) |
| 30 | + if request.name == "missing": |
| 31 | + await context.abort(grpc.StatusCode.NOT_FOUND, "No provider") |
| 32 | + if request.name == "blocked": |
| 33 | + self.started.set() |
| 34 | + try: |
| 35 | + await asyncio.Future() |
| 36 | + finally: |
| 37 | + self.cancelled.set() |
| 38 | + return pb.ProviderResponse( |
| 39 | + provider=dm.Provider(metadata=dm.ObjectMeta(name=request.name)) |
| 40 | + ) |
| 41 | + |
| 42 | + |
| 43 | +@asynccontextmanager |
| 44 | +async def gateway() -> AsyncGenerator[tuple[Gateway, grpc.aio.Channel]]: |
| 45 | + service = Gateway() |
| 46 | + server = grpc.aio.server() |
| 47 | + rpc.add_OpenShellServicer_to_server(service, server) |
| 48 | + port = server.add_insecure_port("127.0.0.1:0") |
| 49 | + await server.start() |
| 50 | + try: |
| 51 | + async with grpc.aio.insecure_channel(f"127.0.0.1:{port}") as channel: |
| 52 | + yield service, channel |
| 53 | + finally: |
| 54 | + await server.stop(0) |
| 55 | + |
| 56 | + |
| 57 | +_METADATA = (("authorization", "Bearer test-token"),) |
| 58 | + |
| 59 | + |
| 60 | +async def test_concurrent_calls_and_task_cancellation() -> None: |
| 61 | + async with gateway() as (service, channel): |
| 62 | + stub = rpc.OpenShellStub(channel) |
| 63 | + |
| 64 | + async def blocked_call() -> pb.ProviderResponse: |
| 65 | + return await stub.GetProvider( |
| 66 | + pb.GetProviderRequest(name="blocked"), timeout=5, metadata=_METADATA |
| 67 | + ) |
| 68 | + |
| 69 | + pending = asyncio.create_task(blocked_call()) |
| 70 | + try: |
| 71 | + await asyncio.wait_for(service.started.wait(), timeout=5) |
| 72 | + response = await stub.GetProvider( |
| 73 | + pb.GetProviderRequest(name="ready"), timeout=5, metadata=_METADATA |
| 74 | + ) |
| 75 | + assert response.provider.metadata.name == "ready" |
| 76 | + finally: |
| 77 | + pending.cancel() |
| 78 | + with pytest.raises(asyncio.CancelledError): |
| 79 | + await pending |
| 80 | + await asyncio.wait_for(service.cancelled.wait(), timeout=5) |
| 81 | + response = await stub.GetProvider( |
| 82 | + pb.GetProviderRequest(name="still-ready"), timeout=5, metadata=_METADATA |
| 83 | + ) |
| 84 | + assert response.provider.metadata.name == "still-ready" |
| 85 | + |
| 86 | + |
| 87 | +@pytest.mark.parametrize( |
| 88 | + ("name", "expected"), |
| 89 | + [ |
| 90 | + ("missing", grpc.StatusCode.NOT_FOUND), |
| 91 | + ("blocked", grpc.StatusCode.DEADLINE_EXCEEDED), |
| 92 | + ], |
| 93 | +) |
| 94 | +async def test_status_and_deadline(name: str, expected: grpc.StatusCode) -> None: |
| 95 | + async with gateway() as (_, channel): |
| 96 | + stub = rpc.OpenShellStub(channel) |
| 97 | + with pytest.raises(grpc.aio.AioRpcError) as error: |
| 98 | + await stub.GetProvider( |
| 99 | + pb.GetProviderRequest(name=name), timeout=0.2, metadata=_METADATA |
| 100 | + ) |
| 101 | + assert error.value.code() == expected |
0 commit comments