Test upload client

Tests and fakes for UploadClient.

FakeLambdaClient

FakeLambdaClient(response: str | bytes = b'success', *, invoke_exception: BaseException | None = None)

Minimal mock of the Lambda client used by UploadClient.

The fake records every invoke() call and returns a dict containing a readable Payload object, matching the only part of boto3's response shape that UploadClient.initiate_transfer() currently consumes.

last_call `property`

last_call: LambdaInvokeCall

Return the most recent invoke call.

invoke

invoke(*, FunctionName: str, Payload: bytes, **kwargs: Any) -> dict[str, BytesIO]

Record one invocation and return the configured response.

FakeSSMClient

Fake SSM client that only returns fixed network configuration parameters.

FakeSession

FakeSession(lambda_client: FakeLambdaClient | None = None, ssm_client: FakeSSMClient | None = None)

Minimal mock session for upload-client tests.

client

client(service_name: str, *, config: Any = None, **kwargs: Any) -> FakeLambdaClient | FakeSSMClient

Return the fake client for service_name == 'lambda' or 'ssm'.

LambdaInvokeCall `dataclass`

LambdaInvokeCall(function_name: str, payload: bytes, kwargs: dict[str, Any])

Recorded call to FakeLambdaClient.invoke().

json_payload `property`

json_payload: dict[str, Any]

Return the invoke payload parsed as JSON.

text_payload `property`

text_payload: str

Return the invoke payload decoded as UTF-8 text.

SessionClientCall `dataclass`

SessionClientCall(service_name: str, config: Any = None, kwargs: dict[str, Any] | None = None)

Recorded call to FakeSession.client().

UploadClientRig `dataclass`

UploadClientRig(client: UploadClient, registry: FakeBucketRegistry, fake_session: FakeSession, source: Path, monkeypatch: MonkeyPatch, bucket_factory: Callable[..., Any])

Small test harness for an UploadClient connected to fake AWS objects.

bucket_factory_for

bucket_factory_for(registry: FakeBucketRegistry) -> Callable[..., Any]

Return the Bucket replacement used while UploadClient connects.

install_upload_client_fakes

install_upload_client_fakes(monkeypatch: MonkeyPatch) -> None

Patch non-Bucket collaborators to deterministic fakes.

make_source_tree

make_source_tree(tmp_path: Path, files: Mapping[str, str]) -> Path

Create a local upload source tree from relative paths to text.

make_upload_client_rig

make_upload_client_rig(monkeypatch: MonkeyPatch, tmp_path: Path, files: Mapping[str, str], *, registry: FakeBucketRegistry | None = None, fake_session: FakeSession | None = None, transfer_bucket_files: Mapping[str, str | bytes] | None = None, n_threads: int = 1, max_retries: int = 2) -> UploadClientRig

Build a local-source UploadClient wired to in-memory AWS fakes.

test_concurrent_transfer_scheduling_uploads_each_file_once

test_concurrent_transfer_scheduling_uploads_each_file_once(monkeypatch: MonkeyPatch, tmp_path: Path) -> None

Concurrent scheduling does not skip or duplicate manifest files.

test_one_file_exhausts_retries_while_other_file_validates

test_one_file_exhausts_retries_while_other_file_validates(monkeypatch: MonkeyPatch, tmp_path: Path) -> None

A failed upload can become complete by exhausting retries.