Skip to content
Roboto
Esc
↑↓navigate↵open⌘Jpreview
On this page

roboto.testing

Public testing utilities.

Submodules

Package Contents

FakeHttpResponseFactory

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

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

Usage

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]]

StubRobotoClient

class roboto.testing.StubRobotoClient#View Source

Bases: roboto.http.RobotoClient

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

Usage

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"

StubRobotoClient.delete()

delete(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#View Source

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

StubRobotoClient.expect_delete()

expect_delete(path, response, query=None, request_body_matcher=None)#View Source

Parameters

path str
response StubResponse
query Optional[dict[str, Any]]
request_body_matcher Optional[RequestBodyMatcher]

Return type

StubRobotoClient

StubRobotoClient.expect_get()

expect_get(path, response, query=None, owner_org_id=None)#View Source

Parameters

path str
response StubResponse
query Optional[dict[str, Any]]
owner_org_id Optional[str]

Return type

StubRobotoClient

StubRobotoClient.expect_patch()

expect_patch(path, response, query=None, request_body_matcher=None)#View Source

Parameters

path str
response StubResponse
query Optional[dict[str, Any]]
request_body_matcher Optional[RequestBodyMatcher]

Return type

StubRobotoClient

StubRobotoClient.expect_post()

expect_post(path, response, query=None, request_body_matcher=None)#View Source

Parameters

path str
response StubResponse
query Optional[dict[str, Any]]
request_body_matcher Optional[RequestBodyMatcher]

Return type

StubRobotoClient

StubRobotoClient.expect_put()

expect_put(path, response, query=None, request_body_matcher=None)#View Source

Parameters

path str
response StubResponse
query Optional[dict[str, Any]]
request_body_matcher Optional[RequestBodyMatcher]

Return type

StubRobotoClient

StubRobotoClient.get()

get(path, caller_org_id=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#View Source

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

StubRobotoClient.patch()

patch(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#View Source

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

StubRobotoClient.post()

post(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#View Source

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

StubRobotoClient.put()

put(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=None)#View Source

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

StubRobotoClient.verify_no_pending_expectations()

verify_no_pending_expectations()#View Source

Verify that all registered expectations have been consumed.

Raises

AssertionError

If there are any pending expectations that were not consumed.

Return type

None

Was this page helpful?