Testing

Public test helpers. FakeSession / FakeResponse are the recommended seam for downstream integrators driving the Polaris client without real HTTP.

iceberg_ops.testing

Public test helpers for tools that integrate with iceberg-ops.

Today the module exposes FakeSession and FakeResponse - a lightweight stand-in for a requests.Session and requests.Response pair. The experimental iceberg-ops Polaris client (PolarisClient) takes an injectable session= kwarg so that downstream consumers - and our own test suite - can drive it against this in-process fake instead of a live HTTP service. Its endpoint paths do not match the Polaris 1.x policy API yet, so the fake is for exercising the client, not for modelling a real Polaris server.

Usage

from iceberg_ops.testing import FakeResponse, FakeSession

sess = FakeSession()
sess.register("GET", "/policies", FakeResponse(json_body={"policies": []}))

client = PolarisClient(
    "https://polaris.example.com",
    catalog="mycat",
    token="t",
    session=sess,
)
client.list_policies("ns")

# Inspect what the client actually sent.
assert sess.calls[0]["method"] == "GET"
assert "/policies" in sess.calls[0]["url"]

Behaviour

• FakeSession records every request() / post() call in FakeSession.calls.

• Responses are matched by (method, path_substring) and consumed on first match (subsequent calls to the same endpoint require another FakeSession.register()).

• An unmatched call raises AssertionError so tests fail loudly rather than silently producing surprising output.

Stability

Both classes are part of the documented public API. Their attributes / methods follow semantic versioning: additions are non-breaking; renames or removals are breaking and bump the major version. Code outside this repository may rely on:

• FakeSession.__init__() signature

• FakeSession.register(), FakeSession.request(), FakeSession.post(), FakeSession.calls

• FakeResponse.__init__ keyword-only signature, plus the attributes status_code and text and the methods json and raise_for_status.

class iceberg_ops.testing.FakeResponse(*, status_code=200, json_body=None, text=None)

Bases: object

Minimal stand-in for requests.Response.

Exposes the four attributes a typical REST client touches: status_code, text, json(), and raise_for_status() (the last is a no-op here, kept for forward-compatibility if a client adopts it).

Parameters:

• status_code (int, default: 200) – HTTP status code. Default 200.

• json_body (Any | None, default: None) – Body returned by json(). None means the response carries no JSON body and json() raises.

• text (str | None, default: None) – Raw response text. Defaults to a JSON-serialised json_body when json_body is given, otherwise the empty string.

json()

Return the response body decoded as JSON.

Raises:

ValueError – If no json_body was supplied at construction.

Return type:

Any

raise_for_status()

No-op stand-in for requests.Response.raise_for_status().

Return type:

None

class iceberg_ops.testing.FakeSession

Bases: object

Records every request and returns canned responses.

Each call inspects the (method, path_substring) pair and returns the matching response. Tests register response sequences via register(). Unmatched calls raise AssertionError - keeps tests honest about which calls they expect.

Variables:

calls (list[dict]) – Every call this session has serviced, in chronological order. Each dict carries the keys method, url, plus the per-call kwargs (params, json, headers, timeout for request(); data and timeout for post()).

register(method, path_substring, response)

Queue response for the next method/path_substring match.

Each registration is consumed on first match. To service the same endpoint twice, call register() twice.

Parameters:

• method (str) – HTTP method (case-insensitive).

• path_substring (str) – Substring matched against the request URL. Useful when a client builds its own full URL from a base + path.

• response (FakeResponse) – The response to return when the next call matches.

Return type:

None

request(method, url, *, params=None, json=None, headers=None, timeout=None)

Mock for requests.Session.request().

Records the call and returns the next registered response whose (method, path_substring) matches.

Raises:

AssertionError – When no registered response matches.

Parameters:

• method (str)

• url (str)

• params (dict[str, Any] | None, default: None)

• json (Any, default: None)

• headers (dict[str, str] | None, default: None)

• timeout (float | None, default: None)

Return type:

FakeResponse

post(url, *, data=None, timeout=None)

Mock for requests.Session.post().

Some clients call session.post(url, data=...) directly for OAuth2 token requests instead of going through request(). Both paths land here in the right place.

Parameters:

• url (str)

• data (Any, default: None)

• timeout (float | None, default: None)

Return type:

FakeResponse