|
| 1 | +# Copyright 2026 Google LLC All Rights Reserved. |
| 2 | +# |
| 3 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +# you may not use this file except in compliance with the License. |
| 5 | +# You may obtain a copy of the License at |
| 6 | +# |
| 7 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +# |
| 9 | +# Unless required by applicable law or agreed to in writing, software |
| 10 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +# See the License for the specific language governing permissions and |
| 13 | +# limitations under the License. |
| 14 | + |
| 15 | +import importlib |
| 16 | +import sys |
| 17 | +import uuid |
| 18 | +from unittest import mock |
| 19 | + |
| 20 | +import pytest |
| 21 | +from google.protobuf import descriptor_pb2, descriptor_pool |
| 22 | + |
| 23 | + |
| 24 | +def test_operations_descriptor_name(): |
| 25 | + """Validates the canonical proto name registered in the descriptor pool.""" |
| 26 | + from google.longrunning import operations_pb2 |
| 27 | + |
| 28 | + # The descriptor name MUST be the canonical path 'google/longrunning/operations.proto'. |
| 29 | + # If named 'google/longrunning/operations_proto.proto', any downstream proto that |
| 30 | + # imports 'google/longrunning/operations.proto' will fail to resolve. |
| 31 | + assert operations_pb2.DESCRIPTOR.name == "google/longrunning/operations.proto" |
| 32 | + |
| 33 | + # Verify lookup by canonical path succeeds in the global descriptor pool |
| 34 | + pool = descriptor_pool.Default() |
| 35 | + file_desc = pool.FindFileByName("google/longrunning/operations.proto") |
| 36 | + assert file_desc is not None |
| 37 | + assert file_desc.name == "google/longrunning/operations.proto" |
| 38 | + |
| 39 | + |
| 40 | +def test_operations_message_instantiation(): |
| 41 | + """Validates message instantiation from both public module entry points.""" |
| 42 | + from google.longrunning import operations_pb2, operations_proto_pb2 |
| 43 | + |
| 44 | + op1 = operations_pb2.Operation(name="operations/123", done=True) |
| 45 | + assert op1.name == "operations/123" |
| 46 | + assert op1.done is True |
| 47 | + |
| 48 | + op2 = operations_proto_pb2.Operation(name="operations/456", done=False) |
| 49 | + assert op2.name == "operations/456" |
| 50 | + assert op2.done is False |
| 51 | + |
| 52 | + |
| 53 | +def test_downstream_dependency_resolution(): |
| 54 | + """Regression test for b/566222096. |
| 55 | +
|
| 56 | + Downstream clients compile protos containing: |
| 57 | + import "google/longrunning/operations.proto"; |
| 58 | +
|
| 59 | + When their generated *_pb2 modules load, protobuf calls: |
| 60 | +
|
| 61 | + ``` |
| 62 | + _descriptor_pool.Default().AddSerializedFile(...) |
| 63 | + ``` |
| 64 | +
|
| 65 | + which checks that 'google/longrunning/operations.proto' exists in the pool. |
| 66 | + """ |
| 67 | + # Ensure operations_pb2 is imported and registered |
| 68 | + from google.longrunning import operations_pb2 # noqa: F401 |
| 69 | + |
| 70 | + pool = descriptor_pool.Default() |
| 71 | + |
| 72 | + # Create a uniquely named proto to avoid collision during repeated test runs |
| 73 | + unique_name = f"google/test/downstream_{uuid.uuid4().hex}.proto" |
| 74 | + |
| 75 | + downstream_file = descriptor_pb2.FileDescriptorProto() |
| 76 | + downstream_file.name = unique_name |
| 77 | + downstream_file.package = "google.test" |
| 78 | + downstream_file.dependency.append("google/longrunning/operations.proto") |
| 79 | + |
| 80 | + # 1. This step will throw the TypeError if operations.proto is missing (under C++ runtime). |
| 81 | + pool.Add(downstream_file) |
| 82 | + |
| 83 | + # 2. Retrieve the descriptor from the pool (required because pool.Add() returns None). |
| 84 | + file_descriptor = pool.FindFileByName(unique_name) |
| 85 | + assert file_descriptor is not None |
| 86 | + assert file_descriptor.name == unique_name |
| 87 | + assert "google/longrunning/operations.proto" in [ |
| 88 | + dep.name for dep in file_descriptor.dependencies |
| 89 | + ] |
| 90 | + |
| 91 | + |
| 92 | +def test_operations_servicer_interface(): |
| 93 | + """Verifies OperationsServicer default method implementations raise NotImplementedError.""" |
| 94 | + from google.longrunning import operations_pb2_grpc |
| 95 | + |
| 96 | + servicer = operations_pb2_grpc.OperationsServicer() |
| 97 | + context = mock.Mock() |
| 98 | + request = mock.Mock() |
| 99 | + |
| 100 | + # Servicer interface must define all RPCs and default to NotImplementedError |
| 101 | + for method in [ |
| 102 | + servicer.ListOperations, |
| 103 | + servicer.GetOperation, |
| 104 | + servicer.DeleteOperation, |
| 105 | + servicer.CancelOperation, |
| 106 | + servicer.WaitOperation, |
| 107 | + ]: |
| 108 | + with pytest.raises(NotImplementedError): |
| 109 | + method(request, context) |
| 110 | + |
| 111 | + |
| 112 | +def test_add_operations_servicer_to_server(): |
| 113 | + """Verifies server handler registration for OperationsServicer.""" |
| 114 | + from google.longrunning import operations_pb2_grpc |
| 115 | + |
| 116 | + servicer = operations_pb2_grpc.OperationsServicer() |
| 117 | + mock_server = mock.Mock() |
| 118 | + |
| 119 | + operations_pb2_grpc.add_OperationsServicer_to_server(servicer, mock_server) |
| 120 | + |
| 121 | + # Assert that generic RPC handlers were registered with the server |
| 122 | + assert mock_server.add_generic_rpc_handlers.called |
| 123 | + assert mock_server.add_generic_rpc_handlers.call_count == 1 |
| 124 | + |
| 125 | + # Verify the handler structure passed to the server |
| 126 | + handlers = mock_server.add_generic_rpc_handlers.call_args[0][0] |
| 127 | + assert len(handlers) == 1 |
| 128 | + assert hasattr(handlers[0], "service_name") |
| 129 | + assert handlers[0].service_name() == "google.longrunning.Operations" |
| 130 | + |
| 131 | + |
| 132 | +def test_operations_client_stubs_and_static_methods(): |
| 133 | + """Verifies OperationsStub client calls and static method dispatch.""" |
| 134 | + from google.longrunning import operations_pb2_grpc |
| 135 | + |
| 136 | + # 1. Test OperationsStub channel binding and RPC dispatch |
| 137 | + mock_callable = mock.Mock(return_value=mock.Mock()) |
| 138 | + mock_channel = mock.Mock() |
| 139 | + mock_channel.unary_unary = mock.Mock(return_value=mock_callable) |
| 140 | + mock_channel.unary_stream = mock.Mock(return_value=mock_callable) |
| 141 | + |
| 142 | + stub = operations_pb2_grpc.OperationsStub(mock_channel) |
| 143 | + assert mock_channel.unary_unary.call_count == 5 |
| 144 | + |
| 145 | + request = mock.Mock() |
| 146 | + for rpc in [ |
| 147 | + stub.ListOperations, |
| 148 | + stub.GetOperation, |
| 149 | + stub.DeleteOperation, |
| 150 | + stub.CancelOperation, |
| 151 | + stub.WaitOperation, |
| 152 | + ]: |
| 153 | + response = rpc(request) |
| 154 | + assert response is not None |
| 155 | + |
| 156 | + assert mock_callable.call_count == 5 |
| 157 | + |
| 158 | + # 2. Test Operations static invocation helpers |
| 159 | + with mock.patch( |
| 160 | + "grpc.experimental.unary_unary", return_value=mock.Mock() |
| 161 | + ) as mock_unary: |
| 162 | + for static_rpc in [ |
| 163 | + operations_pb2_grpc.Operations.ListOperations, |
| 164 | + operations_pb2_grpc.Operations.GetOperation, |
| 165 | + operations_pb2_grpc.Operations.DeleteOperation, |
| 166 | + operations_pb2_grpc.Operations.CancelOperation, |
| 167 | + operations_pb2_grpc.Operations.WaitOperation, |
| 168 | + ]: |
| 169 | + result = static_rpc(request, "localhost:50051") |
| 170 | + assert result is not None |
| 171 | + |
| 172 | + assert mock_unary.call_count == 5 |
| 173 | + |
| 174 | + |
| 175 | +def test_operations_pb2_non_grpc_fallback_environment(): |
| 176 | + """Verifies module loading and class contract in non-gRPC environments. |
| 177 | +
|
| 178 | + In environments where gRPC is unsupported, the module must fall back to pure |
| 179 | + protobuf-only imports without throwing exceptions, and all message classes |
| 180 | + must remain instantiable. |
| 181 | + """ |
| 182 | + # 1. Simulate an environment where the gRPC-dependent pb2 module is unavailable |
| 183 | + with mock.patch.dict(sys.modules, {"google.longrunning.operations_grpc_pb2": None}): |
| 184 | + # Reload to force execution of the fallback import path |
| 185 | + if "google.longrunning.operations_pb2" in sys.modules: |
| 186 | + fallback_module = importlib.reload( |
| 187 | + sys.modules["google.longrunning.operations_pb2"] |
| 188 | + ) |
| 189 | + else: |
| 190 | + import google.longrunning.operations_pb2 as fallback_module # pragma: NO COVER |
| 191 | + |
| 192 | + # 2. Assert that canonical messages can still be resolved and instantiated |
| 193 | + assert hasattr(fallback_module, "Operation") |
| 194 | + assert hasattr(fallback_module, "GetOperationRequest") |
| 195 | + |
| 196 | + op = fallback_module.Operation(name="operations/789", done=True) |
| 197 | + assert op.name == "operations/789" |
| 198 | + assert op.done is True |
| 199 | + |
| 200 | + # 3. Restore the environment back to its gRPC-enabled state |
| 201 | + importlib.reload(sys.modules["google.longrunning.operations_pb2"]) |
0 commit comments