roboto.testing#

Public testing utilities.

Submodules#

Package Contents#

class roboto.testing.FakeHttpResponseFactory(url='https://test.roboto.ai', response_data='{}', status_code=200, headers=None)#

A factory for creating fake HTTP responses, for use with the roboto.http.HttpClient.

Example

>>> import contextlib
>>> import unittest.mock
>>> from roboto.http import HttpClient, FakeHttpResponseFactory
>>> mock_http_client = unittest.mock.create_autospec(HttpClient, instance=True)
>>> with contextlib.ExitStack() as stack:
...     http_get_mock = stack.enter_context(unittest.mock.patch.object(mock_http_client, "get"))
...     http_get_mock.side_effect = FakeHttpResponseFactory(
...         "https://example.com",
...         {"foo": "bar"},
...         status_code=200,
...         headers={"Content-Type": "application/json"},
...     )
Parameters:

  • url (str)

  • response_data (Any)

  • status_code (int)

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

class roboto.testing.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

On this page