ds_provider_mock_py_lib.linked_service

File: __init__.py Region: ds_provider_mock_py_lib/linked_service

Mock Linked Service

This module implements an in-memory linked service for e2e testing.

Example

>>> from uuid import uuid4
>>> from ds_provider_mock_py_lib.linked_service import MockLinkedService, MockLinkedServiceSettings
>>> linked_service = MockLinkedService(
...     id=uuid4(),
...     name="mock-connection",
...     version="1.0.0",
...     settings=MockLinkedServiceSettings(),
... )
>>> linked_service.connect()

Submodules

Classes

MockBackend

Object returned by .connection. Stands in for a real SDK client.

MockLinkedService

In-memory linked service with tunable connect, health, and request failures.

MockLinkedServiceSettings

Settings required to connect to the in-memory mock backend.

Package Contents

class ds_provider_mock_py_lib.linked_service.MockBackend(settings: MockLinkedServiceSettings)[source]

Object returned by .connection. Stands in for a real SDK client.

Tracks call counts so tests can assert round-trips, and applies linked-service level failure rules independent of any dataset.

settings
call_count = 0
closed = False
request(page: int) None[source]

Simulate one round-trip for page.

Parameters:

page – 1-based page number requested by the dataset.

Raises:

MockBackendError – When the connection is closed or a failure rule matches.

close() None[source]

Mark the backend handle as closed.

class ds_provider_mock_py_lib.linked_service.MockLinkedService[source]

Bases: ds_resource_plugin_py_lib.common.resource.linked_service.LinkedService[MockLinkedServiceSettings]

In-memory linked service with tunable connect, health, and request failures.

settings: MockLinkedServiceSettings
_connection: MockBackend | None = None
_connect_attempts: int = 0
property type: ds_provider_mock_py_lib.enums.ResourceType

Get the type of the linked service.

Returns:

ResourceType

property connection: MockBackend

Return the mock backend established by connect().

Returns:

The connected backend handle.

Return type:

MockBackend

Raises:

ConnectionError – If connect() has not been called.

connect() None[source]

Establish a connection to the mock backend.

Raises:
  • ConnectionError – When connect is configured to fail or is still in the transient-failure window.

  • AuthenticationError – When connect_behaviour is authentication_error.

  • AuthorizationError – When connect_behaviour is authorization_error.

test_connection() tuple[bool, str][source]

Verify mock backend health without raising on failure.

Returns:

(True, "") on success, otherwise (False, reason).

Return type:

tuple[bool, str]

close() None[source]

Release the mock backend handle. Safe to call repeatedly.

class ds_provider_mock_py_lib.linked_service.MockLinkedServiceSettings[source]

Bases: ds_resource_plugin_py_lib.common.resource.linked_service.LinkedServiceSettings

Settings required to connect to the in-memory mock backend.

connect_behaviour: ds_provider_mock_py_lib.enums.ConnectBehaviour

What connect() does after optional transient failures.

connect_delay_ms: int = 0

Artificial delay applied during connect().

connect_fail_first_n: int = 0

First N connect() calls raise ConnectionError, then succeed.

connect_error: ds_provider_mock_py_lib.models.MockError

Error spec used when connect() is configured to fail.

test_connection_ok: bool = True

When False, test_connection() returns a failed health check.

test_connection_message: str = ''

Reason returned when test_connection_ok is False.

latency_ms: int = 0

Artificial delay applied to every connection.request() call.

fail_on_call: int | None = None

Raise MockBackendError on the Nth request() call. None disables.

request_error: ds_provider_mock_py_lib.models.MockError

Error spec used when fail_on_call matches.

drop_after_calls: int | None = None

Every request() after N calls raises. None disables.

drop_error: ds_provider_mock_py_lib.models.MockError

Error spec used when the connection is dropped after N calls.