ds_provider_mock_py_lib.linked_service.mock

File: mock.py Region: ds_provider_mock_py_lib/linked_service/mock

Mock linked service for e2e testing.

Every failure mode is driven by settings so tests can tune behaviour without touching code. Follows LINKED_SERVICE_CONTRACT.md.

Field classification:

exposed : id, name, description, version, settings.* internal : _connection, _connect_attempts

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()
>>> linked_service.connection.call_count
0
>>> linked_service.close()

Attributes

logger

Classes

MockLinkedServiceSettings	Settings required to connect to the in-memory mock backend.
MockBackend	Object returned by .connection. Stands in for a real SDK client.
MockLinkedService	In-memory linked service with tunable connect, health, and request failures.

Module Contents

ds_provider_mock_py_lib.linked_service.mock.logger

class ds_provider_mock_py_lib.linked_service.mock.MockLinkedServiceSettings

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.

class ds_provider_mock_py_lib.linked_service.mock.MockBackend(settings: MockLinkedServiceSettings)

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

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

Mark the backend handle as closed.

class ds_provider_mock_py_lib.linked_service.mock.MockLinkedService

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

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]

Verify mock backend health without raising on failure.

Returns:

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

Return type:

tuple[bool, str]

close() → None

Release the mock backend handle. Safe to call repeatedly.