def test_invalid_argument_returns_correct_code():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = OrdersStub(ch)
        with pytest.raises(grpc.RpcError) as exc:
            stub.CreateOrder(OrderRequest(item_count=-1))
        assert exc.value.code() == grpc.StatusCode.INVALID_ARGUMENT

def test_request_metadata_round_trip():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = OrdersStub(ch)
        metadata = (("x-trace-id", "abc123"),)
        resp, call = stub.CreateOrder.with_call(OrderRequest(), metadata=metadata)

        # Server reflects request-id in response trailing metadata
        trailing = call.trailing_metadata()
        assert ("x-trace-id-echo", "abc123") in trailing

ghz \
  --insecure \
  --proto orders.proto \
  --call orders.Orders/CreateOrder \
  -d '{"item_count":1}' \
  -c 50 \
  -n 10000 \
  localhost:50051

{
  "methodConfig": [{
    "name": [{"service": "example.v1.UserService"}],
    "retryPolicy": {
      "maxAttempts": 4,
      "initialBackoff": "0.1s",
      "maxBackoff": "1s",
      "backoffMultiplier": 2,
      "retryableStatusCodes": ["UNAVAILABLE"]
    }
  }]
}

import grpc
import pytest

def test_get_user_not_found_returns_not_found(stub):
    with pytest.raises(grpc.RpcError) as exc_info:
        stub.GetUser(GetUserRequest(id="nonexistent"))
    assert exc_info.value.code() == grpc.StatusCode.NOT_FOUND

def test_create_user_with_duplicate_email_returns_already_exists(stub, existing_user):
    with pytest.raises(grpc.RpcError) as exc_info:
        stub.CreateUser(CreateUserRequest(email=existing_user.email))
    assert exc_info.value.code() == grpc.StatusCode.ALREADY_EXISTS

def test_unauthenticated_call_returns_unauthenticated(stub_no_auth):
    with pytest.raises(grpc.RpcError) as exc_info:
        stub_no_auth.GetUser(GetUserRequest(id="any"))
    assert exc_info.value.code() == grpc.StatusCode.UNAUTHENTICATED

from google.rpc import error_details_pb2, status_pb2
from grpc_status import rpc_status

def get_user_with_details(stub, user_id):
    try:
        return stub.GetUser(GetUserRequest(id=user_id))
    except grpc.RpcError as e:
        status = rpc_status.from_call(e)
        for detail in status.details:
            if detail.Is(error_details_pb2.BadRequest.DESCRIPTOR):
                bad_req = error_details_pb2.BadRequest()
                detail.Unpack(bad_req)
                for v in bad_req.field_violations:
                    print(f"{v.field}: {v.description}")

func TestUpload_Success(t *testing.T) {
    fake := &fakeUploader{accept: 3}
    client := setupClient(t, fake)

    stream, err := client.Upload(context.Background())
    if err != nil { t.Fatal(err) }

    chunks := []*pb.Chunk{{Data: []byte("a")}, {Data: []byte("b")}, {Data: []byte("c")}}
    for _, c := range chunks {
        if err := stream.Send(c); err != nil { t.Fatal(err) }
    }
    result, err := stream.CloseAndRecv()
    if err != nil { t.Fatal(err) }
    if result.Bytes != 3 { t.Fatalf("got %d, want 3", result.Bytes) }
}

func TestUpload_ServerCompletesEarly(t *testing.T) {
    fake := &fakeUploader{completeAfter: 1}
    client := setupClient(t, fake)

    stream, _ := client.Upload(context.Background())
    stream.Send(&pb.Chunk{Data: []byte("a")})

    // Sending more after server completes should yield io.EOF
    err := stream.Send(&pb.Chunk{Data: []byte("b")})
    if err != io.EOF {
        t.Fatalf("got %v, want io.EOF", err)
    }
    result, _ := stream.CloseAndRecv()
    if result.Bytes != 1 { t.Fatalf("got %d, want 1", result.Bytes) }
}

func TestUpload_EmptyStream(t *testing.T) {
    client := setupClient(t, &fakeUploader{})
    stream, _ := client.Upload(context.Background())
    result, err := stream.CloseAndRecv()
    if err != nil { t.Fatal(err) }
    if result.Bytes != 0 { t.Fatalf("got %d, want 0", result.Bytes) }
}

func TestConversation_Ordering(t *testing.T) {
    fake := &fakeChatter{
        clientMsgs: []*pb.Message{},
        replies: []*pb.Reply{{Seq: 1}, {Seq: 2}, {Seq: 3}},
    }
    client := setupClient(t, fake)
    stream, _ := client.Conversation(context.Background())

    // Client sends 3 messages
    for i := 0; i < 3; i++ {
        stream.Send(&pb.Message{Seq: int32(i)})
    }
    stream.CloseSend()

    // Server sends back 3 replies in order
    var got []int32
    for {
        r, err := stream.Recv()
        if err == io.EOF { break }
        if err != nil { t.Fatal(err) }
        got = append(got, r.Seq)
    }
    want := []int32{1, 2, 3}
    if !reflect.DeepEqual(got, want) {
        t.Fatalf("got %v, want %v", got, want)
    }
}

func TestConversation_ClientHalfClose(t *testing.T) {
    // Server keeps sending after client CloseSend()
    fake := &fakeChatter{repliesAfterCloseSend: []*pb.Reply{{Seq: 99}}}
    client := setupClient(t, fake)
    stream, _ := client.Conversation(context.Background())
    stream.Send(&pb.Message{Seq: 0})
    stream.CloseSend()  // half-close: no more sends, still receiving

    r, err := stream.Recv()
    if err != nil { t.Fatal(err) }
    if r.Seq != 99 { t.Fatalf("got %d, want 99", r.Seq) }
}

func TestConversation_ServerSideCancel(t *testing.T) {
    fake := &fakeChatter{cancelAfterMsg: 1}
    client := setupClient(t, fake)
    stream, _ := client.Conversation(context.Background())
    stream.Send(&pb.Message{Seq: 0})

    _, err := stream.Recv()
    st, _ := status.FromError(err)
    if st.Code() != codes.Cancelled {
        t.Fatalf("got %v, want Cancelled", st.Code())
    }
}

import grpc
from orders_pb2 import OrderRequest
from orders_pb2_grpc import OrdersStub

def test_unary_create_order():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = OrdersStub(ch)
        resp = stub.CreateOrder(OrderRequest(item_count=2), timeout=5.0)
        assert resp.order_id != ""

def test_server_streaming_price_ticker():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = PricesStub(ch)
        stream = stub.SubscribePrices(SubscribeRequest(symbol="AAPL"), timeout=10.0)
        ticks = []
        for tick in stream:
            ticks.append(tick)
            if len(ticks) >= 5:
                stream.cancel()
                break

        assert len(ticks) == 5
        assert all(t.symbol == "AAPL" for t in ticks)

def test_client_streaming_upload():
    def chunks():
        for i in range(10):
            yield UploadChunk(seq=i, data=b"x" * 1024)

    with grpc.insecure_channel("localhost:50051") as ch:
        stub = UploadsStub(ch)
        resp = stub.Upload(chunks(), timeout=10.0)

        assert resp.total_chunks == 10
        assert resp.total_bytes == 10 * 1024

import asyncio

async def test_bidi_chat():
    async def client_messages():
        for msg in ["hello", "how are you", "bye"]:
            yield ChatMessage(text=msg)
            await asyncio.sleep(0.1)

    async with grpc.aio.insecure_channel("localhost:50051") as ch:
        stub = ChatStub(ch)
        responses = []
        async for resp in stub.Chat(client_messages()):
            responses.append(resp)

        assert len(responses) >= 3

def test_deadline_returns_correct_status():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = SlowStub(ch)
        with pytest.raises(grpc.RpcError) as exc_info:
            stub.SlowOperation(SlowRequest(), timeout=0.5)

        assert exc_info.value.code() == grpc.StatusCode.DEADLINE_EXCEEDED

def test_server_observes_deadline_propagation():
    # Service should respect deadline and cancel its own downstream calls
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = OrchestratorStub(ch)
        with pytest.raises(grpc.RpcError):
            stub.Compose(ComposeRequest(), timeout=0.1)

    # Verify downstream call observed the cancellation
    downstream_state = fetch_downstream_state()
    assert downstream_state.cancelled_count >= 1

def test_cancellation_is_observed_server_side():
    with grpc.insecure_channel("localhost:50051") as ch:
        stub = LongRunningStub(ch)
        future = stub.LongOperation.future(LongRequest())
        time.sleep(0.5)
        future.cancel()

        # Server should record cancellation
        time.sleep(0.5)
        state = fetch_server_metrics()
        assert state.cancelled_count >= 1