roboto.testing.stub_roboto_client#

Module Contents#

roboto.testing.stub_roboto_client.RequestBodyMatcher#
roboto.testing.stub_roboto_client.StubResponse#
class roboto.testing.stub_roboto_client.StubRobotoClient#

Bases: roboto.http.RobotoClient

A stub RobotoClient for unit testing that returns pre-configured responses for expected requests.

Example

>>> from roboto_test_utils import StubRobotoClient
>>> from roboto.domain.datasets import Dataset, DatasetRecord
>>> from roboto.time import utcnow
>>>
>>> client = StubRobotoClient()
>>> client.expect_get(
...     "/v1/datasets/ds_123",
...     response=DatasetRecord(
...         dataset_id="ds_123",
...         created=utcnow(),
...         created_by="test@example.com",
...         modified=utcnow(),
...         modified_by="test@example.com",
...         org_id="og_test",
...     ),
... )
>>>
>>> dataset = Dataset.from_id("ds_123", roboto_client=client)
>>> assert dataset.dataset_id == "ds_123"
delete(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#
Parameters:
  • path (collections.abc.Sequence[str] | str)

  • caller_org_id (Optional[str])

  • data (Any)

  • headers (Optional[dict[str, str]])

  • idempotent (bool)

  • owner_org_id (Optional[str])

  • query (Optional[dict[str, Any]])

  • retry_wait_fn (Optional[Any])

  • timeout (Any)

Return type:

roboto.http.response.HttpResponse

expect_delete(path, response, query=None, request_body_matcher=None)#
Parameters:
  • path (str)

  • response (StubResponse)

  • query (Optional[dict[str, Any]])

  • request_body_matcher (Optional[RequestBodyMatcher])

Return type:

StubRobotoClient

expect_get(path, response, query=None)#
Parameters:
  • path (str)

  • response (StubResponse)

  • query (Optional[dict[str, Any]])

Return type:

StubRobotoClient

expect_patch(path, response, query=None, request_body_matcher=None)#
Parameters:
  • path (str)

  • response (StubResponse)

  • query (Optional[dict[str, Any]])

  • request_body_matcher (Optional[RequestBodyMatcher])

Return type:

StubRobotoClient

expect_post(path, response, query=None, request_body_matcher=None)#
Parameters:
  • path (str)

  • response (StubResponse)

  • query (Optional[dict[str, Any]])

  • request_body_matcher (Optional[RequestBodyMatcher])

Return type:

StubRobotoClient

expect_put(path, response, query=None, request_body_matcher=None)#
Parameters:
  • path (str)

  • response (StubResponse)

  • query (Optional[dict[str, Any]])

  • request_body_matcher (Optional[RequestBodyMatcher])

Return type:

StubRobotoClient

get(path, caller_org_id=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#
Parameters:
  • path (collections.abc.Sequence[str] | str)

  • caller_org_id (Optional[str])

  • headers (Optional[dict[str, str]])

  • idempotent (bool)

  • owner_org_id (Optional[str])

  • query (Optional[dict[str, Any]])

  • retry_wait_fn (Optional[Any])

  • timeout (Any)

Return type:

roboto.http.response.HttpResponse

patch(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#
Parameters:
  • path (collections.abc.Sequence[str] | str)

  • caller_org_id (Optional[str])

  • data (Any)

  • headers (Optional[dict[str, str]])

  • idempotent (bool)

  • owner_org_id (Optional[str])

  • query (Optional[dict[str, Any]])

  • retry_wait_fn (Optional[Any])

  • timeout (Any)

Return type:

roboto.http.response.HttpResponse

post(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#
Parameters:
  • path (collections.abc.Sequence[str] | str)

  • caller_org_id (Optional[str])

  • data (Any)

  • headers (Optional[dict[str, str]])

  • idempotent (bool)

  • owner_org_id (Optional[str])

  • query (Optional[dict[str, Any]])

  • retry_wait_fn (Optional[Any])

  • timeout (Any)

Return type:

roboto.http.response.HttpResponse

put(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#
Parameters:
  • path (collections.abc.Sequence[str] | str)

  • caller_org_id (Optional[str])

  • data (Any)

  • headers (Optional[dict[str, str]])

  • idempotent (bool)

  • owner_org_id (Optional[str])

  • query (Optional[dict[str, Any]])

  • retry_wait_fn (Optional[Any])

  • timeout (Any)

Return type:

roboto.http.response.HttpResponse

verify_no_pending_expectations()#

Verify that all registered expectations have been consumed.

Raises:

AssertionError – If there are any pending expectations that were not consumed.

Return type:

None