synology_apm_repo.sdk.storage.recording module

RecordingStore/ReplayStore capture real ObjectStore traffic into a small, committable fixture, then replay it later without touching any real sample data.

The problem this solves: integration tests need real, production-shaped bytes to be meaningful, but the real sample repositories are far too large to live in CI. Wrapping ObjectStore’s four narrow byte-oriented methods to record every call/result pair is cheap, and the resulting fixture for one realistic scenario is typically a few hundred KB to a couple MB uncompressed — small enough to commit alongside the test it supports: record once against a real sample with RecordingStore, dump it to JSON, then replay it forever after in CI via ReplayStore, with zero real I/O. Fixtures are committed gzip-compressed (tests/fixtures/*.json.gz) via write_fixture_text; load_fixture_text/ ReplayStore.from_path decompress transparently on read.

TracingStore below reuses the same mechanism for the CLI’s --trace flag — observing every ObjectStore.read() call is the same capability whether the goal is a test fixture or showing a user why a command was slow.

synology_apm_repo.sdk.storage.recording.load_fixture_text(path)

Read a fixture, transparently gzip-decompressing.

Parameters:

path (Path) – Fixture file to read. Read as plain text unless it ends in .gz — this project’s on-disk convention for a committed tests/fixtures/*.json.gz cassette.

synology_apm_repo.sdk.storage.recording.write_fixture_text(path, text)

Write fixture text, gzip-compressing when path ends in .gz.

class synology_apm_repo.sdk.storage.recording.RecordingStore(backing)

Bases: object

Wraps a real ObjectStore and records every call/result pair for later dump.

async read(path, offset=0, length=None)
async size(path)
async exists(path)
async listdir(path)
async aclose()

Delegates to _InstrumentedStore.aclose() so a real S3Store/AzureStore recorded against (--record-against=profile:<name>) still gets its aiohttp connector closed once recording is done.

dump()

Serialize everything recorded so far to a JSON string suitable for committing as a test fixture (binary read results are base64-encoded).

class synology_apm_repo.sdk.storage.recording.ReplayStore(fixture_json)

Bases: object

Answers ObjectStore calls purely from a RecordingStore.dump fixture — no real backing store, no I/O.

Raises:

NotFoundError – For any call that was not recorded, so a fixture’s coverage gaps surface immediately as a test failure rather than silently returning wrong data (this is why exists() also raises rather than defaulting to False for an unrecorded path — a “does not exist” result must have been recorded).

classmethod from_path(path)

Construct from a fixture file, transparently gzip-decompressing a .json.gz path (see load_fixture_text).

async read(path, offset=0, length=None)
async size(path)
async exists(path)
async listdir(path)
class synology_apm_repo.sdk.storage.recording.TraceEvent(method, path, offset=0, length=None, result_length=None, elapsed=0.0)

Bases: object

One ObjectStore call, for the --trace CLI flag.

Unlike RecordingStore, nothing is buffered — a real --trace run against a real repository can touch millions of chunks, so TracingStore streams one event per call straight to a callback instead.

Variables:

  • method (str) – "read", "size", "exists", or "listdir".

  • path (str) – Store-relative path the call targeted.

  • offset (int) – read’s byte offset; 0 for every other method.

  • length (int | None) – read’s requested length; None for every other method.

  • result_length (int | None) – Bytes actually returned (read) or entries listed (listdir); None for size/exists, which report a plain value rather than a count.

  • elapsed (float) – Wall-clock seconds the call took.

method: str
path: str
offset: int = 0
length: int | None = None
result_length: int | None = None
elapsed: float = 0.0
class synology_apm_repo.sdk.storage.recording.TracingStore(backing, on_event)

Bases: object

Wraps a real ObjectStore, calling on_event once per call with a TraceEvent — the same “wrap the four narrow methods” mechanism RecordingStore above uses. Every call is forwarded to backing and its result returned/raised unchanged.

property backing: ObjectStore

The real store this instance forwards to – lets a caller tracking store identity (Session.close_repo()) recognize two separate TracingStore wraps of the same backing connector as one shared resource, not two.

async read(path, offset=0, length=None)
async size(path)
async exists(path)
async listdir(path)
async aclose()

Delegates to _InstrumentedStore.aclose() so a real S3Store/AzureStore wrapped for --trace (or by this project’s own smoke tooling, which always traces) still gets its aiohttp connector closed by Session.close().