kmock package

class kmock.AiohttpInterceptor(host, port=None, extra=None, own_resolver=None)[source]

Bases: AsyncContextManager[AiohttpInterceptor, bool | None]

Intercept hostname resolution for selected hostnames to specified IP:port.

The hostname filter can be specified either by explicit string or a regexp, or by a list/set of strings/regexpsts. The regexp patterns must match fully. String hostnames are case-insensitive, while patterns are case-sensitive (add re.I to make it case insensitive explicitly).

For extra precision, the filter can be also specified as tuple(s) with the hostname (as defined above) and a specific port (integer).

Examples:

  • 'kopf.dev'

  • re.compile(r'kopf..*')

  • ['kopf.dev', re.compile(r'kopf..*')]

  • ('kopf.dev', 80)

  • (re.compile(r'kopf..*'), 443)

  • {('kopf.dev', 80), (re.compile(r'kopf..*'), 443), 'kopf.cloud'}

If the target port is not specified, it resolves to the requested port. Otherwise, it is also intercepted and forced to the specified port.

The filter can be modified at runtime if needed.

The target host & port are always intercepted regardless of the filter.

The resolver affects the hostname resolution only since the moment it is entered and only until it is exited. Multiple resolvers must be stacked as LIFO (last-in-first-out): i.e. exited in the reverse sequence as entered.

Parameters:
host: str[source]
port: int | None[source]
extra: Pattern[str] | str | tuple[Pattern[str] | str, int | None] | Collection[Pattern[str] | str | tuple[Pattern[str] | str, int | None]] | None[source]
class kmock.Selectable(*args, **kwargs)[source]

Bases: Protocol

A minimally sufficient resource to be recognized for selection/filtering. Anything less that that is not even considered. More fields CAN be present (as per other protocols), but not required.

group: str | None[source]
version: str | None[source]
plural: str | None[source]
class kmock.Request(*, id=0, method=None, url=URL(''), params=NOTHING, headers=NOTHING, cookies=NOTHING, body=b'', text='', data=None, action=None, resource=None, namespace=None, name=None, subresource=None, raw_request=None)[source]

Bases: object

An incoming request with pre-parsed HTTP- & Kubernetes-specific intentions.

The request can be compared/asserted against a wide range of simpler types if they are understood by kmock.Criteria and descendants:

assert kmock.gets[0] == b'{}'  # bytes are request bodies
assert kmock.gets[0] == {'status': {}}  # dicts do partial nested json matching
assert kmock.gets[0] == '/api/v1'  # strings starting with slash are full paths
assert kmock.gets[0] == re.compile(r'/api/v1.*')  # regexps are paths
assert kmock.gets[0] == 'get'  # known HTTP methods are directly supported
assert kmock.gets[0] == 'list'  # known Kubernetes actions are directly supported
assert kmock.gets[0] == kmock.method.POST  # so as enums
assert kmock.gets[0] == kmock.action.WATCH  # so as enums
assert kmock.gets[0] == kmock.resource('', 'v1', 'pods')  # Kubernetes specific resources
assert kmock.gets[0] == kmock.resource(group='kopf.dev')  # Kubernetes partial resources
assert kmock.gets[0] == 'ns'  # BEWARE: strings are namespaces, not the requested content

If implicit comparision is not sufficient, specific fields can be used. It is more verbose and wordy but very precise.

Note

int is not supported for HTTP statuses: requests have no statuses, those are in responses. We do not assert on responses since the user typically defines the responses manually.

Parameters:
id: int[source]
method: method[source]
url: URL[source]
params: Mapping[str, str][source]
headers: Mapping[str, str][source]
cookies: Mapping[str, str][source]
body: bytes[source]
text: str[source]
data: Any[source]
action: action | None[source]
resource: resource | None[source]
namespace: str | None[source]
name: str | None[source]
subresource: str | None[source]
class kmock.Criteria[source]

Bases: object

A standalone container for specialized named criteria.

static guess(arg, /)[source]
Parameters:

arg (Criteria | EllipsisType | None | str | bool | bytes | SupportsBool | Pattern[str] | Pattern[bytes] | action | method | resource | Selectable | Mapping[str, EllipsisType | None | str | bool | bytes | SupportsBool | Pattern[str] | Pattern[bytes] | action | method | resource | Selectable | Mapping[str, Criterion] | set[Criterion] | frozenset[Criterion] | tuple[Criterion, ...] | Event | Event | Future[Criterion] | Future[Criterion] | Callable[[], bool] | Callable[[], SupportsBool] | Callable[[Request], bool] | Callable[[Request], SupportsBool]] | set[EllipsisType | None | str | bool | bytes | SupportsBool | Pattern[str] | Pattern[bytes] | action | method | resource | Selectable | Mapping[str, Criterion] | set[Criterion] | frozenset[Criterion] | tuple[Criterion, ...] | Event | Event | Future[Criterion] | Future[Criterion] | Callable[[], bool] | Callable[[], SupportsBool] | Callable[[Request], bool] | Callable[[Request], SupportsBool]] | frozenset[EllipsisType | None | str | bool | bytes | SupportsBool | Pattern[str] | Pattern[bytes] | action | method | resource | Selectable | Mapping[str, Criterion] | set[Criterion] | frozenset[Criterion] | tuple[Criterion, ...] | Event | Event | Future[Criterion] | Future[Criterion] | Callable[[], bool] | Callable[[], SupportsBool] | Callable[[Request], bool] | Callable[[Request], SupportsBool]] | tuple[EllipsisType | None | str | bool | bytes | SupportsBool | Pattern[str] | Pattern[bytes] | action | method | resource | Selectable | Mapping[str, Criterion] | set[Criterion] | frozenset[Criterion] | tuple[Criterion, ...] | Event | Event | Future[Criterion] | Future[Criterion] | Callable[[], bool] | Callable[[], SupportsBool] | Callable[[Request], bool] | Callable[[Request], SupportsBool], ...] | Event | Event | Future[EllipsisType | None | str | bool | bytes | SupportsBool | Pattern[str] | Pattern[bytes] | action | method | resource | Selectable | Mapping[str, Criterion] | set[Criterion] | frozenset[Criterion] | tuple[Criterion, ...] | Event | Event | Future[Criterion] | Future[Criterion] | Callable[[], bool] | Callable[[], SupportsBool] | Callable[[Request], bool] | Callable[[Request], SupportsBool]] | Future[EllipsisType | None | str | bool | bytes | SupportsBool | Pattern[str] | Pattern[bytes] | action | method | resource | Selectable | Mapping[str, Criterion] | set[Criterion] | frozenset[Criterion] | tuple[Criterion, ...] | Event | Event | Future[Criterion] | Future[Criterion] | Callable[[], bool] | Callable[[], SupportsBool] | Callable[[Request], bool] | Callable[[Request], SupportsBool]] | Callable[[], bool] | Callable[[], SupportsBool] | Callable[[Request], bool] | Callable[[Request], SupportsBool] | data | text[Pattern[str]] | body[Pattern[bytes]] | params[Pattern[str]] | headers[Pattern[str]] | cookies[Pattern[str]])

Return type:

Criteria | None

class kmock.RawHandler(stream_queue=NOTHING, stream_task=None, *, limit=None, strict=False, entered=0)[source]

Bases: Root

A mock handler to be injected into WSGi-like servers.

The criteria & responses can be injected with the <</>> (see Requests matching and Prepared responses).

For a locally running server with a TCP port, see Server.

Implementation details:

Due to the nature of async/await routines, it is impossible to notify the stream consumers about the addition of new items from synchronous methods like << / >> — it is only possible from asynchronous ones.

The only known primitive supporting synchronous “nowait” is asyncio.Queue: even if the item is put via asyncio.Queue.put_nowait, all async consumers get it immediately.

To work around this limitation, the handler/server MUST be “entered” as a context manager (async with), which implicitly starts a background task and uses a queue to deliver the items in the “nowait” mode.

This comes with trade-offs:

  • First, there can be a minor delay after stream feeding and before the items are delivered to live streams. This is actually utilised to pack several fed items into one batch, as in: kmock[...] << b'hello' << b'world' << ....

  • Second, if the code has no await somewhere after feeding the content, the streams can be blocked with no actual delivery happening until the code gives control back to the event loop via the next await.

Parameters:

  • stream_queue (Queue[tuple[Iterable[Bus[StreamBatch]], StreamBatch]])

  • stream_task (Task[None] | None)

  • limit (int | None)

  • strict (bool)

  • entered (int)

limit: int | None[source]
strict: bool[source]
servers: list[Server][source]
property clients: list[ClientSession][source]
property active: bool[source]
property errors: list[Exception][source]

Errors of the current level of context managers (for assertions).

property url: URL[source]
request(method, url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

get(url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

put(url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

post(url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

patch(url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

options(url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

head(url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

delete(url, **kwargs)[source]
Parameters:
Return type:

_RequestContextManager

class kmock.Server(handler, host='127.0.0.1', port=None, server_cls=<class 'aiohttp.test_utils.RawTestServer'>, client_fct=<class 'aiohttp.client.ClientSession'>, hostnames=None, *, user_agent='kmock/0.7.dev7+g7e5aaad4d aiohttp/3.13.5')[source]

Bases: object

All-in-one bundle needed to run the mock app on an HTTP port.

The handlers are simple WSGI-like apps that have no networking capabilities. The servers implement networking, specifically listening on a TCP/HTTP port and marshalling the requests & responses to/from the handler.

Kmock uses aiohttp by default. Using other libraries is possible as long as the incoming reqeusts & responses support the core signatures of aiohttp. This is NOT an officially supported functionality and might break any time.

There can be several severs pointing to the same handler — e.g. with dfferent hosts:ports, therefore different URLs. The traffic is balanced across such entry points when the client methods of the handler are used. When the client methods of a specific server are used, the traffic goes through that server’s endpoint only.

A few examples with 2 servers on different random ports pointing to the same handler, which defines the reactions & accumulates the requests.

Automatically balanced traffic:

async with RawHandler() as kmock, Server(kmock), Server(kmock):
    for _ in range(10):
        await kmock.get('/')

Routing the traffic manually by directly using the server:

async with RawHandler() as kmock, Server(kmock), Server(kmock) as srv2:
    for _ in range(10):
        await srv2.client.get('/')

The same as above, but via the handler (e.g. via a fixture):

async with RawHandler() as kmock, Server(kmock), Server(kmock):
    for _ in range(10):
        await kmock.clients[1].get('/')
Parameters:
handler: RawHandler[source]
hostnames: Pattern[str] | str | tuple[Pattern[str] | str, int | None] | Collection[Pattern[str] | str | tuple[Pattern[str] | str, int | None]] | None[source]
user_agent: str[source]
property url: URL[source]
property host: str[source]

The real host (IP address) assigned to the server (when started).

If listening on a catch-all host (e.g. 0.0.0.0), this will report any IP address of the current machine, preferably localhost/127.0.0.1. I.e., it points to the host where the server can be accessed, not only where it is listening.

property port: int[source]

The real port assigned to the server (when started).

property client: ClientSession[source]
property server: BaseTestServer[source]
class kmock.KubernetesScaffold(stream_queue=NOTHING, stream_task=None, *, limit=None, strict=False, entered=0)[source]

Bases: RawHandler

A bare structure of the Kubernetes API: errors, API & resource discovery.

It is stateless! It keeps nothing in memory except for what was fed into it. For the stateful API, see KubernetesEmulator.

Parameters:

  • stream_queue (Queue[tuple[Iterable[Bus[StreamBatch]], StreamBatch]])

  • stream_task (Task[None] | None)

  • limit (int | None)

  • strict (bool)

  • entered (int)

property resources: ResourcesArray[source]
class kmock.KubernetesEmulator(stream_queue=NOTHING, stream_task=None, *, limit=None, strict=False, entered=0, buses=NOTHING)[source]

Bases: KubernetesScaffold

A server that mimics Kubernetes and tracks the state of objects in memory.

Object creation, patching, and deletion are tracked. The operations emulate the behaviour of a realistic Kubernetes server, but very simplistically. The server then serves these objects on listings, watch-streams, fetching. It is essentially an in-memory-database-over-http.

However, unlike the real Kubernetes server, this emulator only modifies the objects as simple JSON structures: merge the dicts, overwrite the keys. There is no special treatment of “special” fields, e.g. lists where the new values are added/merged instead of overwriting the whole dict key. If you need “special” field treament, inherit and implement for your cases.

All objects are available for assertions via the kmock.objects field (in addition to the inherited kmock.requests and others).

Note: the object tracking happens even if there are reactions that catch the creation/update/deletion operations. However, the objects are shown only in the default handlers, i.e. when the fetching/listing/watching requests are not intercepted (because those responses will return their content instead of the stored states of the objects).

See also

Simulator vs. emulator: https://stackoverflow.com/a/1584701/857383

Parameters:
property objects: ObjectsArray[source]
class kmock.SinkBox(arg, /)[source]

Bases: object

Parameters:

arg (None | Path | RawIOBase | TextIOBase | BufferedIOBase | MutableSequence[Request] | MutableSet[Request] | Future[Request] | Future[Request] | Queue[Request] | Queue[Request] | Event | Condition | Event | Condition | Bus[Request] | Generator[Sink | Any, Request | None, Sink | None] | AsyncGenerator[Sink | Any, Request | None] | Awaitable[Sink | Any] | Callable[[], Sink | Any] | Callable[[Request], Sink | Any] | SinkBox)

class kmock.action(*values)[source]

Bases: str, Enum

Which K8s action is requested, based on the HTTP method & URL structure.

Note: It follows the same convention of using verbs as in HTTP method, except for the deletion: it would conflict with the well-known HTTP verb. In most cases, there is no big difference between the HTTP method DELETE and the Kubernetes-level action of deletion, so the "delete" can be used. If it matters, the action criterion can be enforced either by passing an enum value action.DELETE explicitly (not a string), or as the keyword argument action='delete' (both name & value are accepted in this case).

LIST = 'LIST'[source]
WATCH = 'WATCH'[source]
FETCH = 'FETCH'[source]
CREATE = 'CREATE'[source]
UPDATE = 'UPDATE'[source]
DELETE = 'DELETE'[source]
classmethod guess(value: None) → None[source]
classmethod guess(value: Self) → Self
classmethod guess(value: str) → Self
Parameters:

value (str | None | Self)

Return type:

Self | None

class kmock.method(*values)[source]

Bases: str, Enum

Which HTTP method is requested (of those we recognize).

GET = 'GET'[source]
PUT = 'PUT'[source]
POST = 'POST'[source]
HEAD = 'HEAD'[source]
PATCH = 'PATCH'[source]
DELETE = 'DELETE'[source]
OPTIONS = 'OPTIONS'[source]
classmethod guess(value: None) → None[source]
classmethod guess(value: Self) → Self
classmethod guess(value: str) → Self
Parameters:

value (str | None | Self)

Return type:

Self | None

class kmock.data(arg: Any, /)[source]
class kmock.data(*args: Mapping[Any, Any], **kwargs: Any)

Bases: object

Parameters:
data: Any[source]
class kmock.text(arg: None | P, /)[source]
class kmock.text(*args: str | bytes)

Bases: Generic[P]

Parameters:

args (str | bytes | None | P)

text: str | P[source]
class kmock.body(arg: None | P, /)[source]
class kmock.body(*args: str | bytes)

Bases: Generic[P]

Parameters:

args (str | bytes | None | P)

body: bytes | P[source]
class kmock.params(*args, **kwargs)[source]

Bases: Generic[P], patterndict[P]

Parameters:
class kmock.headers(*args, **kwargs)[source]

Bases: Generic[P], patterndict[P]

Parameters:
class kmock.cookies(*args, **kwargs)[source]

Bases: Generic[P], patterndict[P]

Parameters:
class kmock.resource(arg1=None, arg2=None, arg3=None, /, *, group=None, version=None, plural=None)[source]

Bases: Selectable

A resource specification that can match several resource kinds.

The resource specifications are not usable in K8s API calls, as the API has no endpoints with masks or placeholders for unknown or catch-all resource identifying parts (e.g. any API group, any API version, any name).

They are used only locally in the operator to match against the actual resources with specific names (kmock.resource). The handlers are defined with resource specifications, but are invoked with specific resource kinds. Even if those specifications look very concrete and allow no variations, they still remain specifications.

Parameters:
group: str | None[source]
version: str | None[source]
plural: str | None[source]
check(resource)[source]
Parameters:

resource (Selectable)

Return type:

bool

kmock.subresource(arg)[source]
Parameters:

arg (Pattern[str] | str | None)

Return type:

Criteria

kmock.name(arg)[source]
Parameters:

arg (Pattern[str] | str | None)

Return type:

Criteria

kmock.namespace(arg)[source]
Parameters:

arg (Pattern[str] | str | None)

Return type:

Criteria

class kmock.View[source]

Bases: ABC

A base view into collections of requests with all the filtering DSL in it.

property override: Priority[source]
property fallback: Priority[source]
class kmock.Root(stream_queue=NOTHING, stream_task=None)[source]

Bases: View

A view for the root handlers/servers where the requests first arrive.

It is needed as the terminal point of walking the tree of views for registering the payloads on << & >> operators. Also as a shared sync-to-async live feeder (at least one root must be “entered”).

Parameters:

  • stream_queue (Queue[tuple[Iterable[Bus[StreamBatch]], StreamBatch]])

  • stream_task (Task[None] | None)

class kmock.Group(sources)[source]

Bases: View

Base class for & & | logic groups of views.

Parameters:

sources (Sequence[View])

class kmock.OrGroup(sources)[source]

Bases: Group

The OR-group to check that a request belongs to ANY of the grouped filters.

Examples: (kmock['get'] | kmock['/'] | kmock['?q=query']) << b'hello', except that not all filters can be put in a set (e.g. dicts).

When it comes to iterating or indexing the requests, they are concatenated in the order of grouping, i.e. a|b|c are yielded in this order: a, b, c. As such, (a|b|c)[0] or (a|b|c)[-1] can belong to either group depending on what is seen (recorded & filtered) in the other groups.

Parameters:

sources (Sequence[View])

class kmock.AndGroup(sources)[source]

Bases: Group

The AND-group to check that a request belongs to ALL of the grouper filters.

Examples: kmock['get'] & kmock['/']. The same as kmock['get', '/'].

Grouping views to different roots (handlers/servers) makes no sense, as it will always be empty by definition (a request can come only from one root).

Parameters:

sources (Sequence[View])

class kmock.Chained(source)[source]

Bases: View

A base view for all other views with a single source, forming a _chain_.

Parameters:

source (View)

class kmock.Exclusion(source, exclusions)[source]

Bases: Chained

A NOT-filter for requests in the source EXCEPT requests in excluded sources.

Examples: kmock['get'] - kmock['/'] - kmock['?q=query'].

Parameters:
class kmock.Slicer(source, s)[source]

Bases: Chained

A slice of requests, as in list: with the [start:stop:step] notation.

Examples: kmock['get'][1:], kmock['post'][:3], kmock['/'][::2].

Parameters:
s: slice[source]
class kmock.Filter(source, criteria)[source]

Bases: Chained

A criteria-based filter or filter by arbitrary bool-evaluable callback.

Examples:

  • kmock['get']['/'][{'watch': 'true'}]

  • kmock[{'Content-Type': 'application/json'}]

  • kmock[lambda req: req.params.get('idx') == '5']

  • kmock['post', kmock.body(re.compile(r'.*hello.*'))]

A sequence of filters is collapsed into one filter preserving the logic. If there are apparently conflicting criteria, which lead to no result, an errors is raised: e.g. kmock['get']['post'].

Parameters:
criteria: Criteria[source]
class kmock.Priority(source, priority)[source]

Bases: Chained

An internal hint to pass the info through the chain.

It is consumed when ordering the payloads to be served on each request.

The chain can stretch from the first root node till the final payload node, e.g.: kmock.fallback['get']['/'][:1] << b''; or directly precede the payload: kmock['get']['/'][:1].fallback << b''.

An alternative would be to store the priority attribute on every node, even on those that have no use or need for prioritizing (bad abstractions).

A hint: use (… ** a) ** b for sub-priority A for the same values of B. But beware the Python math: a**b**c is a**(b**c), not (a**b)**c. E.g. (kmock ** -100) ** -math.inf << 404 for the sub-fallback.

Parameters:
priority: float[source]
class kmock.Reaction(source, response=NOTHING)[source]

Bases: Chained

An actionable response reaction for a filter with payloads and side effects.

Examples:

  • kmock['get /'] << b'hello'

  • kmock['get /'] >> callback_fn >> (reqs:=[])

  • kmock['get /'] << 301 << 'Location: https://kopf.dev' << b'Use Kopf!'

  • kmock.fallback << 404

Worth noting: creating a new reaction from the old reaction (by using chained << & >> operators) does not simply add the new reaction in addition to the old one, but replaces it to avoid duplicates.

As such, in this code, r1 will never serve or record any requests, only r2 will (both payload items will be served by r2 as a stream). Essentially, r1 becomes abandoned & deactivated at creation of r2:

r1 = kmock['get /'] << b'hello, '
r2 = r1 << b'world!\n'

This is an intentional design decision. Otherwise, i.e. by keeping r1 active, it might lead to iconsistencies: either r1 will record requests that it did not actually serve, or r1 will serve regular (non-streamed) payloads followed by streaming r2, which is a conflicting behaviour.

For extra code safety, assertions on deactivated reactions are prohibited. If there is a scenaro where this is needed, please report it as an issue.

Parameters:

  • source (View)

  • response (Response)

response: Response[source]
property priorities: tuple[float, ...][source]
class kmock.Stream(source, batch=NOTHING)[source]

Bases: Chained

A pseudo-view into “live” (aka “lazy”) responses or streams.

Examples:

  • kmock['/'] << ...; loop.call_later(9, lambda: kmock['get /'][...] << b'hello')

  • kmock['/'] << (...,); loop.call_later(9, lambda: kmock[...] << b'all streams see' << ...)

Parameters:

  • source (View)

  • batch (StreamBatch)

class kmock.ObjectVersion(value=None, **kwargs)[source]

Bases: MutableMapping[str, Any]

A single-dict wrapper with extra syntax features for simpler/shorter tests.

In the core, it is a read-only mapping with a view into the wrapped dict, typically with the keys like metadata, spec, and status, but this is neither required nor guaranteed.

Unlike the regular dict, the object has set-like recursive partial matching. However, both the keys presence and their values must match on comparison.

The actual “greater” object may contain more keys than the “smaller” pattern expects (typically, the metadata, names/namespaces, or resource versions), but those in the pattern are required and must store the expected values.

The inverse comparison —the “smaller” raw dict & the “greater” dict view— is NOT a boolean inverse of the straighforward operator, but a role switch: the “smaller” unwrapped raw dict becomes a pattern, the “greater” wrapped dict becomes an object to be checked with extra keys.

There are, by design, no > or < operators, only >= and <=. The strict inclusion-except-equality of dicts makes no sense semantically. Also to avoid confusion between the role switching and inverse comparison.

All in all, the actual pattern is always on the “smaller” side (always!).

Example with the object containing the pattern (and maybe more):

assert kmock.objects[r, 'ns', 'n'] >= {'spec': {'key': 'must be this'}}
assert {'spec': {'key': 'must be this'}} <= kmock.objects[r, 'ns', 'n']

Example with the object NOT containing the pattern (can contain more keys):

assert not kmock.objects[r, 'ns', 'n'] >= {'spec': {'key': 'not this'}}
assert not {'spec': {'key': 'not this'}} <= kmock.objects[r, 'ns', 'n']

The ... aka Ellipsis means “any value but the key must be present” (or “must be absent” with the negation/inversion):

assert kmock.objects[r, 'ns', 'n'] >= {'spec': {'key': ...}}  # present
assert not kmock.objects[r, 'ns', 'n'] >= {'spec': {'key': ...}}  # absent

Similar to regular dicts, the equality compares the dict as a whole, i.e. no extra keys are allowed in the object (... is supported though):

assert kmock.objects[r, 'ns', 'n'] == {
    'metadata': {'name': 'n1', 'namespace': ...},
    'spec': 'must be this',
}

The inequality means that the values mismatch or there are the extra keys:

assert kmock.objects[r, 'ns', 'n'] != {'spec': {'key': ...}}

For pattern matching of two raw dicts, e.g. from an API, wrap one & compare to a plain-dict pattern, or wrap the plain-dict pattern itself, just to bring the magic to play (but mind the direction of the comparison):

assert kmock.PatternDict(resp['items'][0]) >= {'spec': {'key': ...}}
assert resp['items'][0] >= kmock.PatternDict({'spec': {'key': ...}})
assert {'spec': {'key': ...}} <= kmock.PatternDict(resp['items'][0])
assert kmock.PatternDict({'spec': {'key': ...}}) <= resp['items'][0]

Nested dicts (and only dicts/mappings) are enhanced with magic on access:

assert kmock.objects[r, 'ns', 'n']['spec'] >= {'key': ...}

But the nested non-dicts, i.e. regular values, lists, tuples, are not:

assert kmock.objects[r, 'ns', 'n']['spec']['key'] == 'val'  # no "..."!

To access a raw dict —nested or top-level— convert it to dict(). Then Python does the comparison, i.e. with no recursive partial matching:

assert dict(kmock.objects[r, 'ns', 'n'])['spec'] == {'key': 'only this'}
assert dict(kmock.objects[r, 'ns', 'n']['spec'])['key'] == 'only this'

Note: it is challenging to reliably wrap the nested non-dict values and expose their specialized fields/methods, behaviour, math, reprs, etc. Especially in identity checks like obj['key'] is None. The ... magic is not supported on the extracted scalars or lists.

Parameters:
property raw: dict[str, Any][source]

Return the raw dict ready to be used in JSON, no magic views.

class kmock.ObjectHistory(items=(), /)[source]

Bases: MutableSequence[ObjectVersion | None]

Addressable and adjustable history of dict versions.

None is used as a soft-deletion marker.

Parameters:

items (Iterable[Mapping[str, Any] | None])

clear() → None -- remove all items from S[source]
Return type:

None

insert(index, value)[source]

S.insert(index, value) – insert value before index

Parameters:
Return type:

None

append(value)[source]

S.append(value) – append value to the end of the sequence

Parameters:

value (Mapping[str, Any] | None)

Return type:

None

extend(value)[source]

S.extend(iterable) – extend sequence by appending elements from the iterable

Parameters:

value (Iterable[Mapping[str, Any] | None])

Return type:

None

pop([index]) → item -- remove and return item at index (default last).[source]

Raise IndexError if list is empty or index is out of range.

Parameters:

index (int)

Return type:

ObjectVersion | None

remove(value, /)[source]

S.remove(value) – remove first occurrence of value. Raise ValueError if the value is not present.

Parameters:

value (Mapping[str, Any] | None)

Return type:

None

property last: ObjectVersion[source]

The last seen state before the soft-deletion(s) or after the patch(es).

property raw: list[ObjectVersion | None][source]
class kmock.Object(history=(), /)[source]

Bases: MutableMapping[str, Any]

A view of a single object: kmock.objects[resource, namespace, name].

  • == & != recursively compare for precise equality.

  • <= & >= recursively compare against a partial pattern.

Accessing/modifying the versioned dict delegates to the latest version of the dict. The previous versions are unaffected unless modified directly:

kmock.objects['v1/pods', 'ns', 'n'].history[0]['spec'] = {'key': 'val'}

The modification of individual keys or keys en mass, such as .update(…), changes the latest version and does not grow the history — in order to make it easier to modify objects in tests without the API-level side effects. Only the dedicated methods for creating/patching/deleting grow the history.

Another critical difference of .patch() vs. .update(): .patch() merges the sub-dicts as it happens in the Kubernetes API; .update() does not merge the sub-dicts, instead it overwrites them as if simply assigned by keys — to mimic the behavior of Python’s dicts. This method is is a part of the MutableMapping, not an API-like method:

obj = VersionedDict({'spec': {'interval': 10, 'timeout': 60}})

obj.patch({'spec': {'interval': 20}})
assert obj == {'spec': {'interval': 20, 'timeout': 60}}

obj.update({'spec': {'interval': 30}})
assert obj == {'spec': {'interval': 30}}

# Synonymous to:
obj['spec'] = {'interval': 30}

If there is no latest version —whether because it was never populated or because it was soft-deleted— then a new empty dict is auto-populated.

Parameters:

history (Mapping[str, Any] | None | Iterable[Mapping[str, Any] | None])

clear()[source]

Clear the data from the latest visible version (not the history!).

Return type:

None

delete()[source]

Soft-delete the object by marking it as such but retaining the history.

It is possible to soft-delete the versioned dict a few times in a row.

In contrast, del kmock.objects[…] physically deletes it from memory, while kmock.objects[…].delete() keeps the object’s history. The API emulator uses the soft-deletion internally.

Return type:

None

create(changes=None, /, **kwargs)[source]

Populate the dict with the new data.

Can be called only if the object is either empty or soft-deleted.

Parameters:
Return type:

None

patch(changes=None, /, **kwargs)[source]

Recursively patch the object by creating a new version.

For scalar values, it is the same as native Python update (replacing). For dicts, it is a recursive merge (patching). This is so to prevent a dict value in the patch to replace the whole pre-existing dict in the base object instead of several keys in it:

obj = Object({'spec': {'timeout': 60, 'interval': 10}})
obj.patch({'spec': {'interval': 15}})
assert obj == {'spec': {'timeout': 60, 'interval': 15}}

Note the preservance of timeout=60 in this example, which would be lost with the native Python update.

None removes the key (even if nested).

Parameters:
Return type:

None

property empty: bool[source]

Whether the object was never created/populated (yet).

property deleted: bool[source]

Whether the object was never created or was recently soft-deleted.

property last: ObjectVersion[source]

The last seen state before the soft-deletion(s) or after the patch(es).

property raw: dict[str, Any][source]

Return the raw dict ready to be used in JSON, no magic views.

property history: ObjectHistory[source]

The full history of the dict, all its versions & soft-deletion markers.

class kmock.ObjectsArray[source]

Bases: object

An associative array of K8s objects and their past versions.

The following notations are supported with a 3-item key:

kmock.objects['v1/pods', 'ns', 'name']          # namespaced objects
kmock.objects['v1/pods', None, 'name']          # cluster-wide objects

The 4-item key is a shortcut for the .history field of the object view:

kmock.objects['v1/pods', 'ns', 'name', 0]       # the initial version
kmock.objects['v1/pods', 'ns', 'name', -2]      # the pre-last version
kmock.objects['v1/pods', 'ns', 'name', -1]      # the current version
kmock.objects['v1/pods', 'ns', 'name', 1:3]     # a slice of versions
kmock.objects['v1/pods', 'ns', 'name', :]       # a list of all versions

In both cases, the first item of the key is either a pre-created instance of kmock.resource, or a string or an object that can be passed to and parsed by kmock.resource: e.g., 'kopf.dev/v1/kopfexamples', 'kopfexamples.v1.kopf.dev', 'v1/pods', 'pods.v1', and some other notations; this includes all objects with string properties .group, .version, .plural.

There are 2 ways to delete the object:

  • del kmock.objects['v1/pods', 'ns', 'n'] hard-deletes it from memory.

  • kmock.objects['v1/pods', 'ns', 'n'].delete() soft-deletes the object.

When the object is hard-deleted, a new object with the same name begins a new history starting with the version zero. The previous history is lost.

When the object is soft-deleted, a placeholder None is put as the latest version. It does NOT compare to any dict anymore (always raises an error). However, it still has .history to access the past states of the object, or .last to access the last seen non-deleted state.

The array does not do any introspection or interpretation of an object’s stored contents, such as name-, namespace-, or resource-guessing. These elements must be provided as the object’s address (key) explicitly. The API handler does the introspection from the URL and the request payload.

A new object with the same name continues the versioning with the new state. The following state of the object’s history is possible (note: the <= & >= mean set-like “includes but may contain more”):

POST /…/namespaces/ns/… ← {'spec': '1st', 'metadata': {'name': 'n1'}}
PATCH /…/namespaces/ns/…/n1 ← {'spec': '1st modified'}
DELETE /…/namespaces/ns/…/n1
POST /…/namespaces/ns/… ← {'spec': '2nd', 'metadata': {'name': 'n1'}}

assert len(kmock.objects[r, 'ns', 'n1', 0].history) == 4
assert kmock.objects[r, 'ns', 'n1', 0] >= {'spec': '1st'}
assert kmock.objects[r, 'ns', 'n1', 1] >= {'spec': '1st modified'}
assert kmock.objects[r, 'ns', 'n1', 2] is None
assert kmock.objects[r, 'ns', 'n1', 3] >= {'spec': '2nd'}

Objects tend to be sorted chronologically in the order of first appearance with the same code flow — the same as it happens for usual Python dicts. The hard-deletion followed by re-creation moves the object to the end. The soft-deletion followed by re-creation keeps it in its original position.

keys()[source]
Return type:

Iterator[tuple[resource, str | None, str]]

values()[source]
Return type:

Iterator[Object]

items()[source]
Return type:

Iterator[tuple[tuple[resource, str | None, str], Object]]

clear()[source]

Remove all past & current objects, reset to the initial state.

Return type:

None

class kmock.ResourceDict[source]

Bases: TypedDict

namespaced: bool[source]
kind: str[source]
singular: str[source]
verbs: Iterable[str][source]
shortnames: Iterable[str][source]
categories: Iterable[str][source]
subresources: Iterable[str][source]
class kmock.ResourceInfo(*, namespaced=None, kind=None, singular=None, verbs=NOTHING, shortnames=NOTHING, categories=NOTHING, subresources=NOTHING)[source]

Bases: object

The extended discovery information about a resource.

The identifying fields —group, version, plural— are not part of the class, as they are used as the key of ResourceArray or kmock.resources that leads to the extended resource information.

Parameters:
namespaced: bool | None[source]
kind: str | None[source]
singular: str | None[source]
verbs: set[str][source]
shortnames: set[str][source]
categories: set[str][source]
subresources: set[str][source]
class kmock.ResourcesArray(resources=None, /)[source]

Bases: MutableMapping[str | Selectable | resource, ResourceInfo | ResourceDict]

An associative array of extended information about cluster references.

Exposed via kmock.resources.

The key is either a pre-created instance of kmock.resource, or a string that can be passed to and parsed by kmock.resource: e.g., 'kopf.dev/v1/kopfexamples', 'kopfexamples.v1.kopf.dev', 'v1/pods', 'pods.v1', and some other notations; this includes all objects with string properties .group, .version, .plural. It is stored pre-parsed, so iterating over the array yields the parsed instances of kmock.resource.

The values are of type ResourceInfo. They are auto-created on access, so the properties can be assigned without preparations:

kmock.resources['kopf.dev/v1/kopfexamples'].kind = 'KopfExample'
kmock.resources['v1/pods'].shortnames = {'pod', 'po'}

The alternative is storing a new instance of ResourceInfo:

kmock.resources['v1/pods'] = ResourceInfo(shortnames={'pod', 'po'})

It is used in the API discovery only and is not used for payload validation. As such, all the information is fully optional. If the whole record or any fields are absent, they are returned either empty or guessed from the plural name of the resource (not grammatically correct, of course, but sufficient).

Parameters:

resources (Mapping[str | Selectable | resource, ResourceInfo | ResourceDict] | None)

clear() → None.  Remove all items from D.[source]
Return type:

None

load_data(data, /)[source]

Load the resource definitions from a data blob.

The data structure is opaque, i.e. you should not interpret it. Store it and use it exactly as generated by kmock fetch resources. The data format may change in the future without a warning (with backwards compatibility for the old formats).

Parameters:

data (str | bytes)

Return type:

None

load_path(path, /)[source]

Load the resource definitions from a file.

See ResourcesArray.load_data().

Parameters:

path (str | Path)

Return type:

None

load_bundled()[source]

Load the prepared bundled resource definitions from the package itself.

Beware: it might be outdated and incomplete. The file is maintained on the best effort basis. Usually, it is constructed from a recent K3s cluster in the default setup, with CRDs stripped, only with builtins.

For anything different from this quick and “good enough” option, build your own resource file with kmock fetch resources --help.

Return type:

None

exception kmock.KubernetesError(status=500, reason=None, message=None, details=None)[source]

Bases: Exception

An error with its own K8s-specific payload & status code.

Parameters:

  • status (int)

  • reason (str | None)

  • message (str | None)

  • details (Any | None)

Return type:

None

status: int[source]
reason: str | None[source]
message: str | None[source]
details: Any | None[source]
exception kmock.KubernetesNotFoundError(details=None, status=404, reason='Not Found', message='The URL or resource was not found')[source]

Bases: KubernetesError

An error for the HTTP 404 errors for lacking URLs in Kubernetes.

Parameters:

  • details (Any | None)

  • status (int)

  • reason (str)

  • message (str)

Return type:

None

status: int[source]
reason: str[source]
message: str[source]
exception kmock.KubernetesEndpointNotFoundError(details=None, status=404, reason='Endpoint Not Found', message="The URL was not found and does not match any known pattern. Declare it as: kmock['/endpoint'] << b'some payload'")[source]

Bases: KubernetesNotFoundError

An error for the HTTP 404 errors for unknown endpoint in Kubernetes.

Parameters:

  • details (Any | None)

  • status (int)

  • reason (str)

  • message (str)

Return type:

None

reason: str[source]
message: str[source]
exception kmock.KubernetesResourceNotFoundError(details=None, status=404, reason='Resource Not Found', message="The resource is not declared in the Kubernetes server. Declare it either way, e.g.: kmock.resources['group/v1/plural'] = {}")[source]

Bases: KubernetesNotFoundError

An error for the HTTP 404 errors for unknown resources in Kubernetes.

Parameters:

  • details (Any | None)

  • status (int)

  • reason (str)

  • message (str)

Return type:

None

reason: str[source]
message: str[source]
exception kmock.KubernetesObjectNotFoundError(details=None, status=404, reason='Object Not Found', message="The object is not found in the Kubernetes server for a known resource. Add it as: kmock.objects[res, 'ns', 'name'] = {'spec': ...}")[source]

Bases: KubernetesNotFoundError

An error for the HTTP 404 errors for unknown objects in Kubernetes.

Parameters:

  • details (Any | None)

  • status (int)

  • reason (str)

  • message (str)

Return type:

None

reason: str[source]
message: str[source]