sift_py.grpc.transport_test ¶

CLASS	DESCRIPTION
`AuthInterceptor`
`DataService`
`ForceFailInterceptor`	Force RPC to fail a few times before letting it pass.

AuthInterceptor ¶

Bases: ServerInterceptor

METHOD	DESCRIPTION
`intercept`
`intercept_service`

ATTRIBUTE	DESCRIPTION
`AUTH_REGEX`

AUTH_REGEX = compile('^Bearer (.+)$')

intercept(
    method: Callable,
    request_or_iterator: Any,
    context: ServicerContext,
    method_name: str,
) -> Any

intercept_service(continuation, handler_call_details)

Bases: DataServiceServicer

METHOD	DESCRIPTION
`GetData`

GetData(request: GetDataRequest, context: ServicerContext)

ForceFailInterceptor(
    expected_num_fails: int,
    failure_code: StatusCode = UNKNOWN,
)

Bases: ServerInterceptor

Force RPC to fail a few times before letting it pass.

failed_attempts: Count of how many times failed expected_num_fails: How many times you want call to fail

METHOD	DESCRIPTION
`intercept`
`intercept_service`

expected_num_fails: int = expected_num_fails

failed_attempts: int = 0

failure_code: StatusCode = failure_code

intercept(
    method: Callable,
    request_or_iterator: Any,
    context: ServicerContext,
    method_name: str,
) -> Any

intercept_service(continuation, handler_call_details)

server_spy(
    mocker: MockFixture, *interceptors: ServerInterceptor
) -> Iterator[MockType]

test_internal_error_retry(mocker: MockFixture)

test_sift_channel(mocker: MockFixture)