grpc.framework.face.testing package¶

Submodules¶

grpc.framework.face.testing.base_util module¶

Utilities for creating Base-layer objects for use in Face-layer tests.

class grpc.framework.face.testing.base_util.LinkedPair[source]¶

Bases: object

A Front and Back that are linked to one another.

front¶

An interfaces.Front.

back¶

An interfaces.Back.

shut_down()[source]¶

Shuts down this object and releases its resources.

grpc.framework.face.testing.base_util.linked_pair(servicer, default_timeout)[source]¶

Creates a Server and Stub linked together for use.

grpc.framework.face.testing.blocking_invocation_inline_service_test_case module¶

A test to verify an implementation of the Face layer of RPC Framework.

class grpc.framework.face.testing.blocking_invocation_inline_service_test_case.BlockingInvocationInlineServiceTestCase[source]¶

Bases: grpc.framework.face.testing.test_case.FaceTestCase, grpc.framework.face.testing.coverage.BlockingCoverage

A test of the Face layer of RPC Framework.

Concrete subclasses must also extend unittest.TestCase.

setUp()[source]¶

See unittest.TestCase.setUp for full specification.

Overriding implementations must call this implementation.

tearDown()[source]¶

See unittest.TestCase.tearDown for full specification.

Overriding implementations must call this implementation.

testExpiredStreamRequestStreamResponse()[source]¶

testExpiredStreamRequestUnaryResponse()[source]¶

testExpiredUnaryRequestStreamResponse()[source]¶

testExpiredUnaryRequestUnaryResponse()[source]¶

testFailedStreamRequestStreamResponse()[source]¶

testFailedStreamRequestUnaryResponse()[source]¶

testFailedUnaryRequestStreamResponse()[source]¶

testFailedUnaryRequestUnaryResponse()[source]¶

testSequentialInvocations()[source]¶

testSuccessfulStreamRequestStreamResponse()[source]¶

testSuccessfulStreamRequestUnaryResponse()[source]¶

testSuccessfulUnaryRequestStreamResponse()[source]¶

testSuccessfulUnaryRequestUnaryResponse()[source]¶

grpc.framework.face.testing.callback module¶

A utility useful in tests of asynchronous, event-driven interfaces.

class grpc.framework.face.testing.callback.Callback[source]¶

Bases: grpc.framework.foundation.stream.Consumer

A utility object useful in tests of asynchronous code.

abort(abortion)[source]¶

abortion()[source]¶

block_until_terminated()[source]¶

complete(unary_response)[source]¶

consume(streamed_response)[source]¶

consume_and_terminate(streamed_response)[source]¶

response()[source]¶

responses()[source]¶

terminate()[source]¶

grpc.framework.face.testing.control module¶

Code for instructing systems under test to block or fail.

class grpc.framework.face.testing.control.Control[source]¶

Bases: object

An object that accepts program control from a system under test.

Systems under test passed a Control should call its control() method frequently during execution. The control() method may block, raise an exception, or do nothing, all according to the enclosing test’s desire for the system under test to simulate hanging, failing, or functioning.

control()[source]¶

Potentially does anything.

class grpc.framework.face.testing.control.PauseFailControl[source]¶

Bases: grpc.framework.face.testing.control.Control

A Control that can be used to pause or fail code under control.

control()[source]¶

fail(*args, **kwds)[source]¶

Fails code under control while controlling code is in context.

pause(*args, **kwds)[source]¶

Pauses code under control while controlling code is in context.

grpc.framework.face.testing.coverage module¶

Governs coverage for the tests of the Face layer of RPC Framework.

class grpc.framework.face.testing.coverage.BlockingCoverage[source]¶

Bases: object

Specification of test coverage for blocking behaviors.

testExpiredStreamRequestStreamResponse()[source]¶

testExpiredStreamRequestUnaryResponse()[source]¶

testExpiredUnaryRequestStreamResponse()[source]¶

testExpiredUnaryRequestUnaryResponse()[source]¶

testFailedStreamRequestStreamResponse()[source]¶

testFailedStreamRequestUnaryResponse()[source]¶

testFailedUnaryRequestStreamResponse()[source]¶

testFailedUnaryRequestUnaryResponse()[source]¶

testSequentialInvocations()[source]¶

testSuccessfulStreamRequestStreamResponse()[source]¶

testSuccessfulStreamRequestUnaryResponse()[source]¶

testSuccessfulUnaryRequestStreamResponse()[source]¶

testSuccessfulUnaryRequestUnaryResponse()[source]¶

class grpc.framework.face.testing.coverage.FullCoverage[source]¶

Bases: grpc.framework.face.testing.coverage.BlockingCoverage

Specification of test coverage for non-blocking behaviors.

testCancelledStreamRequestStreamResponse()[source]¶

testCancelledStreamRequestUnaryResponse()[source]¶

testCancelledUnaryRequestStreamResponse()[source]¶

testCancelledUnaryRequestUnaryResponse()[source]¶

testParallelInvocations()[source]¶

testWaitingForSomeButNotAllParallelInvocations()[source]¶

grpc.framework.face.testing.digest module¶

Code for making a service.TestService more amenable to use in tests.

class grpc.framework.face.testing.digest.TestServiceDigest[source]¶

Bases: grpc.framework.face.testing.digest.TestServiceDigest

A transformation of a service.TestService.

name¶

The RPC service name to be used in the test.

methods¶

A sequence of interfaces.Method objects describing the RPC methods that will be called during the test.

inline_method_implementations¶

A dict from RPC method name to face_interfaces.MethodImplementation object to be used in tests of in-line calls to behaviors under test.

event_method_implementations¶

A dict from RPC method name to face_interfaces.MethodImplementation object to be used in tests of event-driven calls to behaviors under test.

multi_method_implementation¶

A face_interfaces.MultiMethodImplementation to be used in tests of generic calls to behaviors under test.

unary_unary_messages_sequences¶

A dict from method name to sequence of service.UnaryUnaryTestMessages objects to be used to test the method with the given name.

unary_stream_messages_sequences¶

A dict from method name to sequence of service.UnaryStreamTestMessages objects to be used to test the method with the given name.

stream_unary_messages_sequences¶

A dict from method name to sequence of service.StreamUnaryTestMessages objects to be used to test the method with the given name.

stream_stream_messages_sequences¶

A dict from method name to sequence of service.StreamStreamTestMessages objects to be used to test the method with the given name.

serialization¶

A serial.Serialization object describing serialization behaviors for all the RPC methods.

grpc.framework.face.testing.digest.digest(service, control, pool)[source]¶

Creates a TestServiceDigest from a TestService.

Parameters:	service – A testing_service.TestService. control – A testing_control.Control. pool – If RPC methods should be serviced in a separate thread, a thread pool. None if RPC methods should be serviced in the thread belonging to the run-time that calls for their service.
Returns:	A TestServiceDigest synthesized from the given service.TestService.

grpc.framework.face.testing.event_invocation_synchronous_event_service_test_case module¶

A test to verify an implementation of the Face layer of RPC Framework.

class grpc.framework.face.testing.event_invocation_synchronous_event_service_test_case.EventInvocationSynchronousEventServiceTestCase[source]¶

Bases: grpc.framework.face.testing.test_case.FaceTestCase, grpc.framework.face.testing.coverage.FullCoverage

A test of the Face layer of RPC Framework.

Concrete subclasses must also extend unittest.TestCase.

setUp()[source]¶

See unittest.TestCase.setUp for full specification.

Overriding implementations must call this implementation.

tearDown()[source]¶

See unittest.TestCase.tearDown for full specification.

Overriding implementations must call this implementation.

testCancelledStreamRequestStreamResponse()[source]¶

testCancelledStreamRequestUnaryResponse()[source]¶

testCancelledUnaryRequestStreamResponse()[source]¶

testCancelledUnaryRequestUnaryResponse()[source]¶

testExpiredStreamRequestStreamResponse()[source]¶

testExpiredStreamRequestUnaryResponse()[source]¶

testExpiredUnaryRequestStreamResponse()[source]¶

testExpiredUnaryRequestUnaryResponse()[source]¶

testFailedStreamRequestStreamResponse()[source]¶

testFailedStreamRequestUnaryResponse()[source]¶

testFailedUnaryRequestStreamResponse()[source]¶

testFailedUnaryRequestUnaryResponse()[source]¶

testParallelInvocations()[source]¶

testSequentialInvocations()[source]¶

testSuccessfulStreamRequestStreamResponse()[source]¶

testSuccessfulStreamRequestUnaryResponse()[source]¶

testSuccessfulUnaryRequestStreamResponse()[source]¶

testSuccessfulUnaryRequestUnaryResponse()[source]¶

testWaitingForSomeButNotAllParallelInvocations(*args, **kwargs)[source]¶

grpc.framework.face.testing.future_invocation_asynchronous_event_service_test_case module¶

A test to verify an implementation of the Face layer of RPC Framework.

class grpc.framework.face.testing.future_invocation_asynchronous_event_service_test_case.FutureInvocationAsynchronousEventServiceTestCase[source]¶

Bases: grpc.framework.face.testing.test_case.FaceTestCase, grpc.framework.face.testing.coverage.FullCoverage

A test of the Face layer of RPC Framework.

Concrete subclasses must also extend unittest.TestCase.

setUp()[source]¶

See unittest.TestCase.setUp for full specification.

Overriding implementations must call this implementation.

tearDown()[source]¶

See unittest.TestCase.tearDown for full specification.

Overriding implementations must call this implementation.

testCancelledStreamRequestStreamResponse()[source]¶

testCancelledStreamRequestUnaryResponse()[source]¶

testCancelledUnaryRequestStreamResponse()[source]¶

testCancelledUnaryRequestUnaryResponse()[source]¶

testExpiredStreamRequestStreamResponse()[source]¶

testExpiredStreamRequestUnaryResponse()[source]¶

testExpiredUnaryRequestStreamResponse()[source]¶

testExpiredUnaryRequestUnaryResponse()[source]¶

testFailedStreamRequestStreamResponse()[source]¶

testFailedStreamRequestUnaryResponse()[source]¶

testFailedUnaryRequestStreamResponse()[source]¶

testFailedUnaryRequestUnaryResponse()[source]¶

testParallelInvocations()[source]¶

testSequentialInvocations()[source]¶

testSuccessfulStreamRequestStreamResponse()[source]¶

testSuccessfulStreamRequestUnaryResponse()[source]¶

testSuccessfulUnaryRequestStreamResponse()[source]¶

testSuccessfulUnaryRequestUnaryResponse()[source]¶

testWaitingForSomeButNotAllParallelInvocations(*args, **kwargs)[source]¶

grpc.framework.face.testing.interfaces module¶

Interfaces implemented by data sets used in Face-layer tests.

class grpc.framework.face.testing.interfaces.Method[source]¶

Bases: object

An RPC method to be used in tests of RPC implementations.

cardinality()[source]¶

Identify the cardinality of the method.

Returns:	A cardinality.Cardinality value describing the streaming semantics of the method.

deserialize_request(serialized_request)[source]¶

Synthesize a request object from a given bytestring.

Parameters:	serialized_request – A bytestring deserializable into a request object appropriate for this method.

deserialize_response(serialized_response)[source]¶

Synthesize a response object from a given bytestring.

Parameters:	serialized_response – A bytestring deserializable into a response object appropriate for this method.

name()[source]¶

Identify the name of the method.

Returns:	The name of the method.

request_class()[source]¶

Identify the class used for the method’s request objects.

Returns:	The class object of the class to which the method’s request objects belong.

response_class()[source]¶

Identify the class used for the method’s response objects.

Returns:	The class object of the class to which the method’s response objects belong.

serialize_request(request)[source]¶

Serialize the given request object.

Parameters:	request – A request object appropriate for this method.

serialize_response(response)[source]¶

Serialize the given response object.

Parameters:	response – A response object appropriate for this method.

grpc.framework.face.testing.serial module¶

Utility for serialization in the context of test RPC services.

class grpc.framework.face.testing.serial.Serialization[source]¶

Bases: grpc.framework.face.testing.serial._Serialization

An aggregation of serialization behaviors for an RPC service.

request_serializers¶

A dict from method name to request object serializer behavior.

request_deserializers¶

A dict from method name to request object deserializer behavior.

response_serializers¶

A dict from method name to response object serializer behavior.

response_deserializers¶

A dict from method name to response object deserializer behavior.

grpc.framework.face.testing.serial.serialization(methods)[source]¶

Creates a Serialization from a sequences of interfaces.Method objects.

grpc.framework.face.testing.service module¶

Private interfaces implemented by data sets used in Face-layer tests.

class grpc.framework.face.testing.service.StreamStreamTestMessages[source]¶

Bases: object

A type for stream-request-stream-response message pairings.

requests()[source]¶

Affords a sequence of request messages.

Implementations of this method should return a different sequences with each call so that multiple test executions of the test method may be made with different inputs.

Returns:	A sequence of request messages.

verify(requests, responses, test_case)[source]¶

Verifies that the computed response matches the given requests.

Parameters:	requests – A sequence of request messages. responses – A sequence of response messages. test_case – A unittest.TestCase object affording useful assertion methods.
Raises:	AssertionError – If the requests and responses do not match, indicating that there was some problem executing the RPC under test.

class grpc.framework.face.testing.service.StreamStreamTestMethodImplementation[source]¶

Bases: grpc.framework.face.testing.interfaces.Method

A controllable implementation of a stream-stream RPC method.

service(response_consumer, context, control)[source]¶

Services an RPC that accepts and produces streams of messages.

Parameters:	response_consumer – A stream.Consumer to be called to accept the response messages of the RPC. context – A face_interfaces.RpcContext object. control – A test_control.Control to control execution of this method.
Returns:	A stream.Consumer with which to accept the request messages of the RPC. The consumer returned from this method may or may not be invoked to completion: in the case of RPC abortion, RPC Framework will simply stop passing messages to this object. Implementations must not assume that this object will be called to completion of the request stream or even called at all.
Raises:	abandonment.Abandoned – May or may not be raised when the RPC has been aborted.

class grpc.framework.face.testing.service.StreamUnaryTestMessages[source]¶

Bases: object

A type for stream-request-unary-response message pairings.

requests()[source]¶

Affords a sequence of request messages.

Implementations of this method should return a different sequences with each call so that multiple test executions of the test method may be made with different inputs.

Returns:	A sequence of request messages.

verify(requests, response, test_case)[source]¶

Verifies that the computed response matches the given requests.

Parameters:	requests – A sequence of request messages. response – A response message. test_case – A unittest.TestCase object affording useful assertion methods.
Raises:	AssertionError – If the requests and response do not match, indicating that there was some problem executing the RPC under test.

class grpc.framework.face.testing.service.StreamUnaryTestMethodImplementation[source]¶

Bases: grpc.framework.face.testing.interfaces.Method

A controllable implementation of a stream-unary RPC method.

service(response_callback, context, control)[source]¶

Services an RPC that takes a stream of messages and produces one message.

Parameters:	response_callback – A callback to be called to accept the response message of the RPC. context – A face_interfaces.RpcContext object. control – A test_control.Control to control execution of this method.
Returns:	A stream.Consumer with which to accept the request messages of the RPC. The consumer returned from this method may or may not be invoked to completion: in the case of RPC abortion, RPC Framework will simply stop passing messages to this object. Implementations must not assume that this object will be called to completion of the request stream or even called at all.
Raises:	abandonment.Abandoned – May or may not be raised when the RPC has been aborted.

class grpc.framework.face.testing.service.TestService[source]¶

Bases: object

A specification of implemented RPC methods to use in tests.

name()[source]¶

Identifies the RPC service name used during the test.

Returns:	The RPC service name to be used for the test.

stream_stream_scenarios()[source]¶

Affords stream-request-stream-response test methods and their messages.

Returns:	A dict from method name to pair. The first element of the pair is a StreamStreamTestMethodImplementation object and the second element is a sequence of StreamStreamTestMethodMessages objects.

stream_unary_scenarios()[source]¶

Affords stream-request-unary-response test methods and their messages.

Returns:	A dict from method name to pair. The first element of the pair is a StreamUnaryTestMethodImplementation object and the second element is a sequence of StreamUnaryTestMethodMessages objects.

unary_stream_scenarios()[source]¶

Affords unary-request-stream-response test methods and their messages.

Returns:	A dict from method name to pair. The first element of the pair is a UnaryStreamTestMethodImplementation object and the second element is a sequence of UnaryStreamTestMethodMessages objects.

unary_unary_scenarios()[source]¶

Affords unary-request-unary-response test methods and their messages.

Returns:	A dict from method name to pair. The first element of the pair is a UnaryUnaryTestMethodImplementation object and the second element is a sequence of UnaryUnaryTestMethodMessages objects.

class grpc.framework.face.testing.service.UnaryStreamTestMessages[source]¶

Bases: object

A type for unary-request-stream-response message pairings.

request()[source]¶

Affords a request message.

Implementations of this method should return a different message with each call so that multiple test executions of the test method may be made with different inputs.

Returns:	A request message.

verify(request, responses, test_case)[source]¶

Verifies that the computed responses match the given request.

Parameters:	request – A request message. responses – A sequence of response messages. test_case – A unittest.TestCase object affording useful assertion methods.
Raises:	AssertionError – If the request and responses do not match, indicating that there was some problem executing the RPC under test.

class grpc.framework.face.testing.service.UnaryStreamTestMethodImplementation[source]¶

Bases: grpc.framework.face.testing.interfaces.Method

A controllable implementation of a unary-stream RPC method.

service(request, response_consumer, context, control)[source]¶

Services an RPC that takes one message and produces a stream of messages.

Parameters:	request – The single request message for the RPC. response_consumer – A stream.Consumer to be called to accept the response messages of the RPC. context – A face_interfaces.RpcContext object. control – A test_control.Control to control execution of this method.
Raises:	abandonment.Abandoned – May or may not be raised when the RPC has been aborted.

class grpc.framework.face.testing.service.UnaryUnaryTestMessages[source]¶

Bases: object

A type for unary-request-unary-response message pairings.

request()[source]¶

Affords a request message.

Implementations of this method should return a different message with each call so that multiple test executions of the test method may be made with different inputs.

Returns:	A request message.

verify(request, response, test_case)[source]¶

Verifies that the computed response matches the given request.

Parameters:	request – A request message. response – A response message. test_case – A unittest.TestCase object affording useful assertion methods.
Raises:	AssertionError – If the request and response do not match, indicating that there was some problem executing the RPC under test.

class grpc.framework.face.testing.service.UnaryUnaryTestMethodImplementation[source]¶

Bases: grpc.framework.face.testing.interfaces.Method

A controllable implementation of a unary-unary RPC method.

service(request, response_callback, context, control)[source]¶

Services an RPC that accepts one message and produces one message.

Parameters:	request – The single request message for the RPC. response_callback – A callback to be called to accept the response message of the RPC. context – An face_interfaces.RpcContext object. control – A test_control.Control to control execution of this method.
Raises:	abandonment.Abandoned – May or may not be raised when the RPC has been aborted.

grpc.framework.face.testing.stock_service module¶

Examples of Python implementations of the stock.proto Stock service.

class grpc.framework.face.testing.stock_service.GetHighestTradePrice[source]¶

Bases: grpc.framework.face.testing.service.StreamUnaryTestMethodImplementation

GetHighestTradePrice for use in tests.

cardinality()[source]¶

deserialize_request(serialized_request)[source]¶

deserialize_response(serialized_response)[source]¶

name()[source]¶

request_class()[source]¶

response_class()[source]¶

serialize_request(request)[source]¶

serialize_response(response)[source]¶

service(response_callback, context, control)[source]¶

class grpc.framework.face.testing.stock_service.GetHighestTradePriceMessages[source]¶

Bases: grpc.framework.face.testing.service.StreamUnaryTestMessages

requests()[source]¶

verify(requests, response, test_case)[source]¶

class grpc.framework.face.testing.stock_service.GetLastTradePrice[source]¶

Bases: grpc.framework.face.testing.service.UnaryUnaryTestMethodImplementation

GetLastTradePrice for use in tests.

cardinality()[source]¶

deserialize_request(serialized_request)[source]¶

deserialize_response(serialized_response)[source]¶

name()[source]¶

request_class()[source]¶

response_class()[source]¶

serialize_request(request)[source]¶

serialize_response(response)[source]¶

service(request, response_callback, context, control)[source]¶

class grpc.framework.face.testing.stock_service.GetLastTradePriceMessages[source]¶

Bases: grpc.framework.face.testing.service.UnaryUnaryTestMessages

request()[source]¶

verify(request, response, test_case)[source]¶

class grpc.framework.face.testing.stock_service.GetLastTradePriceMultiple[source]¶

Bases: grpc.framework.face.testing.service.StreamStreamTestMethodImplementation

GetLastTradePriceMultiple for use in tests.

cardinality()[source]¶

deserialize_request(serialized_request)[source]¶

deserialize_response(serialized_response)[source]¶

name()[source]¶

request_class()[source]¶

response_class()[source]¶

serialize_request(request)[source]¶

serialize_response(response)[source]¶

service(response_consumer, context, control)[source]¶

class grpc.framework.face.testing.stock_service.GetLastTradePriceMultipleMessages[source]¶

Bases: grpc.framework.face.testing.service.StreamStreamTestMessages

Pairs of message streams for use with GetLastTradePriceMultiple.

requests()[source]¶

verify(requests, responses, test_case)[source]¶

class grpc.framework.face.testing.stock_service.StockTestService[source]¶

Bases: grpc.framework.face.testing.service.TestService

A corpus of test data with one method of each RPC cardinality.

name()[source]¶

stream_stream_scenarios()[source]¶

stream_unary_scenarios()[source]¶

unary_stream_scenarios()[source]¶

unary_unary_scenarios()[source]¶

class grpc.framework.face.testing.stock_service.WatchFutureTrades[source]¶

Bases: grpc.framework.face.testing.service.UnaryStreamTestMethodImplementation

WatchFutureTrades for use in tests.

cardinality()[source]¶

deserialize_request(serialized_request)[source]¶

deserialize_response(serialized_response)[source]¶

name()[source]¶

request_class()[source]¶

response_class()[source]¶

serialize_request(request)[source]¶

serialize_response(response)[source]¶

service(request, response_consumer, context, control)[source]¶

class grpc.framework.face.testing.stock_service.WatchFutureTradesMessages[source]¶

Bases: grpc.framework.face.testing.service.UnaryStreamTestMessages

Pairs of a single request message and a sequence of response messages.

request()[source]¶

verify(request, responses, test_case)[source]¶

grpc.framework.face.testing.test_case module¶

Tools for creating tests of implementations of the Face layer.

class grpc.framework.face.testing.test_case.FaceTestCase[source]¶

Bases: object

Describes a test of the Face Layer of RPC Framework.

Concrete subclasses must also inherit from unittest.TestCase and from at least one class that defines test methods.

set_up_implementation(name, methods, method_implementations, multi_method_implementation)[source]¶

Instantiates the Face Layer implementation under test.

Parameters:

name – The service name to be used in the test. methods – A sequence of interfaces.Method objects describing the RPC methods that will be called during the test. method_implementations – A dictionary from string RPC method name to face_interfaces.MethodImplementation object specifying implementation of an RPC method. multi_method_implementation – An face_interfaces.MultiMethodImplementation or None.

Returns:

A sequence of length two the first element of which is a face_interfaces.GenericStub (backed by the given method implementations), and the second element of which is an arbitrary memo object to be kept and passed to tearDownImplementation at the conclusion of the test.

tear_down_implementation(memo)[source]¶

Destroys the Face layer implementation under test.

Parameters:	memo – The object from the second position of the return value of set_up_implementation.

Module contents¶