### HTTP Mock Server Example

Source: https://github.com/nolar/kmock/blob/main/docs/intro.rst

Demonstrates how to set up a simple HTTP mock server using KMock. It pre-populates the server with a specific response for a GET request to the root endpoint and then uses the embedded client to retrieve and assert the response.

```python
import kmock


async def test_http_access(kmock: kmock.RawHandler) -> None:

    # Pre-populate the HTTP server with request criteria and associated responses.
    kmock['get /'] << 418 << {'hello': 'world'}

    # Perform a sample request using the embedded client and assert the response.
    resp = await kmock.get('/')
    data = await resp.json()
    assert resp.status == 418
    assert data == {'hello': 'world'}
```

--------------------------------

### Kubernetes Watch Stream Example

Source: https://github.com/nolar/kmock/blob/main/docs/intro.rst

Demonstrates how to mock Kubernetes watch streams using KMock. It shows setting up a watch stream before simulating resource creation and deletion, then parsing the resulting ADDED and DELETED events from the stream.

```python
import json
import kmock
import asyncio


async def test_k8s_watch(kmock: kmock.KubernetesEmulator) -> None:
    # Announce the existence of the resource to the server.
    kmock.resources['kopf.dev/v1/kopfexamples'] = {}

    # The stream must start BEFORE the activity happens.
    async with kmock.get('/apis/kopf.dev/v1/kopfexamples?watch=true') as resp:

        # Simulate the activity (ignore the responses).
        body = {'metadata': {'namespace': 'ns1', 'name': 'name3'}, 'spec': 789}
        await kmock.post('/apis/kopf.dev/v1/kopfexamples', json=body)
        await kmock.delete('/apis/kopf.dev/v1/namespaces/ns1/kopfexamples/name3')
        await asyncio.sleep(0.1)  # the loopback network stack takes some time

        # Read the accumulated stream and parse into individual events on each line.
        lines: list[bytes] = resp.content.read_nowait().splitlines()
        events = [json.loads(line.decode()) for line in lines]

    # Close the connection, and assert the results.
    
```

--------------------------------

### Kubernetes Mock Server List Example

Source: https://github.com/nolar/kmock/blob/main/docs/intro.rst

Illustrates mocking Kubernetes API interactions with KMock's KubernetesEmulator. It shows how to pre-populate the emulator with objects and then list objects from a specific namespace, asserting the partial content of the response.

```python
import kmock


async def test_k8s_list(kmock: kmock.KubernetesEmulator) -> None:

    # Pre-populate the Kubernetes emulator with objects.
    kmock.objects['kopf.dev/v1/kopfexamples', 'ns1', 'name1'] = {'spec': 123}
    kmock.objects['kopf.dev/v1/kopfexamples', 'ns2', 'name2'] = {'spec': 456}

    # Perform a sample request using the embedded client and assert the response.
    resp = await kmock.get('/apis/kopf.dev/v1/namespaces/ns1/kopfexamples')
    data = await resp.json()
    assert kmock.Object(data) >= {'items': [{'spec': 123}]}
```

--------------------------------

### Mocking HTTP GET Requests with KMock

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Demonstrates how to use kmock to mock HTTP GET requests and assert the responses. It shows how to define expected responses for specific URL paths and read the response body.

```python
resp = await kmock.get('/greetings?name=John')
text = await resp.read()
assert text == b'Hello, John!"

resp = await kmock.get('/greetings')
text = await resp.read()
assert text == b'Hello, user!"
```

--------------------------------

### Mock Kubernetes Actions with kmock

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Demonstrates how to mock Kubernetes actions like 'list', 'watch', 'fetch', and 'create' using the `kmock.action` instance. It shows that common actions can be used directly as strings, while 'delete' requires explicit wrapping due to potential conflicts with HTTP methods. The example also illustrates how to make a GET request and assert the response status.

```python
import kmock

async def test_kubernetes_actions(kmock: kmock.RawHandler) -> None:
    # All these filters are identical. Use the shortest one:
    kmock['list'] << 200
    kmock[kmock.action('list')] << 200

    # Try listing all resources globally in the cluster.
    # There will be no response data, since we gave no useful payload above.
    # But the request will be counted.
    resp = await kmock.get('/apis/kopf.dev/v1/kopfexamples')
    assert resp.status == 200
```

--------------------------------

### Match Kubernetes Sub-resources with kmock

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Illustrates how to match Kubernetes sub-resources using the `kmock.subresource` function, which also accepts regular expressions. The example demonstrates matching the 'scale' sub-resource and any sub-resource starting with 'scale', along with a GET request to a sub-resource endpoint.

```python
import kmock
import re

async def test_kubernetes_subresource_filters(kmock: kmock.RawHandler) -> None:
    kmock['v1/replicasets', kmock.subresource('scale')] << 200
    kmock['v1/replicasets', kmock.subresource(re.compile('scale.*'))] << 200

    resp = await kmock.get('/api/v1/replicasets/example1/scale')
    assert resp.status == 200
```

--------------------------------

### Mocking Kubernetes Stateful APIs

Source: https://github.com/nolar/kmock/blob/main/README.md

Illustrates how to simulate a stateful Kubernetes API using kmock. It demonstrates resource tracking and verifying state changes across multiple POST and GET requests.

```python
import aiohttp
import kmock
import pytest

@pytest.fixture
def k8surl() -> str:
  return 'http://localhost'

async def test_k8s_out_of_the_box(kmock: kmock.RawHandler, k8surl: str) -> None:
    async with aiohttp.ClientSession(base_url=k8surl) as session:
        pod1 = {'metadata': {'name': 'pod1'}, 'spec': {'key': 'val'}}
        pod2 = {'metadata': {'name': 'pod1'}, 'spec': {'key': 'val'}}
        await session.post('/api/v1/namespace/default/pods', json=pod1)
        await session.post('/api/v1/namespace/default/pods', json=pod2)
        resp = await session.get('/api/v1/namespace/default/pods')
        data = await resp.json()
        assert data['items'] == [pod1, pod2]

    assert len(kmock[kmock.LIST]) == 1
    assert len(kmock[kmock.resource['pods']]) == 3
    assert kmock[kmock.resource('v1/pods')][-1].method == 'GET'
```

--------------------------------

### Verify Mocked API Endpoints via Shell

Source: https://github.com/nolar/kmock/blob/main/docs/intro.rst

Demonstrates how to interact with the KMock-powered server using curl to verify mocked responses and streaming behavior from the command line.

```shell
$ curl -i -X POST http://localhost:12345/apis/kopf.dev/v1/namespaces/ns1/kopfexamples
$ curl -i http://localhost:12345/apis/kopf.dev/v1/namespaces/ns1/kopfexamples
$ curl -i http://localhost:12345/apis/kopf.dev/v1/namespaces/ns1/kopfexamples?watch=true | xargs -L 1 echo $(date +'[%Y-%m-%d %H:%M:%S]')
```

--------------------------------

### Match Kubernetes Actions and Resources with kmock

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Shows how to match both Kubernetes actions and resources simultaneously by providing a single space-separated string. The action comes first, followed by the resource name (e.g., 'list pods.v1'). Examples include matching 'list pods.v1' and 'watch kopfexamples.v1.kopf.dev', along with corresponding GET requests.

```python
import kmock

async def test_kubernetes_action_and_resource(kmock: kmock.RawHandler) -> None:
    kmock['list pods.v1'] << 200
    kmock['watch kopfexamples.v1.kopf.dev'] << 200

    resp = await kmock.get('/api/v1/pods')
    assert resp.status == 200

    resp = await kmock.get('/apis/kopf.dev/v1/kopfexamples?watch=true')
    assert resp.status == 200
```

--------------------------------

### Simulate Kubernetes API Events with KMock

Source: https://github.com/nolar/kmock/blob/main/docs/intro.rst

Uses the KMock DSL to define asynchronous streaming responses for Kubernetes watch and list operations. The example demonstrates injecting delays, conditional events, and error codes into the mock server.

```python
import asyncio
import datetime
import json
import pytest
import kmock

@pytest.mark.kmock(port=12345, cls=kmock.RawHandler)
async def test_bizzarily_complex_k8s_simulation(kmock):
    deletion_event = asyncio.Event()
    asyncio.get_running_loop().call_later(6, deletion_event.set)

    gets = kmock['get']
    lists = kmock['list']
    watches = kmock['watch']

    kmock['list kopf.dev/v1/kopfexamples', kmock.namespace('ns1')] << {'items': [], 'metadata': {'resourceVersion': 'v1'}}
    kmock['watch kopf.dev/v1/kopfexamples', kmock.namespace('ns1')] << [
        {'type': 'ADDED', 'object': {'spec': {}}},
        lambda: asyncio.sleep(3),
        lambda: {'type': 'MODIFIED', 'object': {'spec': {'time': datetime.datetime.now(tz=datetime.UTC).isoformat()}}},
        deletion_event.wait(),
        [
            {'type': 'DELETED', 'object': {'metadata': {'name': f'demo{i}'}}}
            for i in range(3)
        ],
        410,
    ]

    kmock << 404 << b'{"error": "not served"}' << {'X-MyServer-Info': 'error'}

    await function_under_test()
```

--------------------------------

### Filter Kubernetes Namespaces with kmock

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Explains how to filter Kubernetes requests based on namespaces using the `kmock.namespace` function. It supports exact namespace matching and regular expressions for broader matching. The example demonstrates matching 'ns1' and any namespace starting with 'ns', along with a GET request to a namespaced resource.

```python
import kmock
import re

async def test_kubernetes_namespace_filtering(kmock: kmock.RawHandler) -> None:
    kmock[kmock.namespace('ns1')] << 200
    kmock[kmock.namespace(re.compile('ns.*'))] << 200

    resp = await kmock.get('/apis/kopf.dev/v1/namespaces/ns1/kopfexamples')
    assert resp.status == 200
```

--------------------------------

### KMock Streaming Responses Syntax (Python)

Source: https://github.com/nolar/kmock/blob/main/docs/streams.rst

Demonstrates how to define streaming responses using KMock. It shows examples of replayable streams, streams with continuations, and how to inject new data into a stream.

```python
import kmock

async def test_streams(kmock: kmock.RawHandler) -> None:
    # Equivalent replayable streams:
    kmock['/'] << 200 << b'hello\n' << b'world\n'
    kmock['/'] << 200 << (b'hello\n', b'world\n')

    # A stream with a continuation, which freezes and wait for new data:
    kmock['/'] << 200 << (b'hello\n', b'world\n', ...)

    # Inject the new data into the stream:
    kmock['/'][...] << b'again\n'
```

--------------------------------

### GET /apis/{group}/{version}/namespaces/{namespace}/{plural}/{name}

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/persistence.rst

Retrieves a specific namespaced resource from the emulator.

```APIDOC
## GET /apis/{group}/{version}/namespaces/{namespace}/{plural}/{name}

### Description
Retrieves the latest version of a namespaced object from the Kubernetes emulator.

### Method
GET

### Endpoint
/apis/{group}/{version}/namespaces/{namespace}/{plural}/{name}

### Parameters
#### Path Parameters
- **group** (string) - Required - The API group of the resource
- **version** (string) - Required - The API version
- **namespace** (string) - Required - The target namespace
- **plural** (string) - Required - The plural name of the resource
- **name** (string) - Required - The name of the object

### Response
#### Success Response (200)
- **object** (dict) - The current state of the object

#### Response Example
{
  "spec": 123
}
```

--------------------------------

### Pytest Integration with KMock KubernetesEmulator

Source: https://github.com/nolar/kmock/blob/main/docs/usage.rst

This example shows how to use KMock within a pytest test function. It utilizes the provided 'kmock' fixture, which is preconfigured with a KubernetesEmulator handler and a running server. The example demonstrates adding an object to the emulator and then querying it via the API.

```python
from kmock import KubernetesEmulator

def test_me(kmock: KubernetesEmulator) -> None:
    kmock.objects['kopf.dev/v1/kopfexamples', 'ns', 'name'] = {'spec': 123}
    resp = await kmock.get('/apis/kopf.dev/v1/kopfexamples')  # list cluster-wide
    data = await resp.json()
    assert kmock.Object(data) >= {'items': [{'spec': 123}]}
```

--------------------------------

### Match HTTP Query Parameters and Headers

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Illustrates how to filter requests based on query parameters and headers using dictionaries or standardized string formats. Includes examples of both explicit wrappers and implicit dictionary matching.

```python
import kmock

async def test_params_and_headers(kmock: kmock.RawHandler) -> None:
    # Query parameters
    kmock[kmock.params('name=john&mode=formal')] << b'Greetings, John!'
    kmock[{'name': 'john', 'mode': 'formal'}] << b'Greetings, John!'
    
    # Headers
    kmock[kmock.headers({'X-API-Token': '123'})] << b'hello'
    kmock[{'X-API-Token': '123'}] << b'hello'
    kmock[kmock.headers('X-API-Token: 123')] << b'hello'
```

--------------------------------

### Run Standalone KMock Server with RawHandler

Source: https://github.com/nolar/kmock/blob/main/docs/usage.rst

This example demonstrates setting up a standalone KMock server using RawHandler. It listens on port 12345, responds to the root endpoint with a hardcoded message, and makes an initial request to google.com using the embedded client. The server runs indefinitely, allowing external access via curl.

```python
import asyncio
from kmock import Server, RawHandler

async def main() -> None:
    async with RawHandler() as kmock, Server(kmock, port=12345, hostnames=['google.com']):
        kmock['/'] << b'Hello, there!' << lambda req: print(f"{req}")

        resp = await kmock.get('http://google.com/')
        text = await resp.read()
        print(f"Response: {text}")

        print(f"Now do: curl -i {kmock.url!s}")
        await asyncio.Event().wait()  # sleep forever


if __name__ == '__main__':
    asyncio.run(main())
```

--------------------------------

### Suppressing Linter Warnings for KMock DSL

Source: https://github.com/nolar/kmock/blob/main/docs/caveats.rst

Demonstrates how to use the KMock DSL operator overloading and provides a workaround for IDE 'statement has no effect' warnings by assigning results to an underscore variable.

```python
import kmock

async def test_me(kmock: kmock.RawHandler) -> None:
    # Standard usage triggering linter warning
    kmock['get /'] << b'hello'
    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello'

    # Workaround for linter warning
    _ = kmock['get /'] << b'hello'
```

--------------------------------

### Configuring Streaming Response Headers

Source: https://github.com/nolar/kmock/blob/main/docs/caveats.rst

Shows the difference between delayed and instant header transmission in KMock streams. Delayed headers wait for the first non-None payload, while including an empty byte string forces immediate header transmission.

```python
import kmock
import asyncio

async def test_delayed_headers_in_stream(kmock: kmock.RawHandler) -> None:
    kmock['get /'] << (
        lambda: asyncio.sleep(5),
        b'hello'
    )
    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello'

async def test_instant_headers_in_stream(kmock: kmock.RawHandler) -> None:
    kmock['get /'] << (
        b'',  # Forces immediate header transmission
        lambda: asyncio.sleep(5),
        b'hello'
    )
    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello'
```

--------------------------------

### Match Kubernetes Object Names with kmock

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Details how to match individual Kubernetes resource object names using the `kmock.name` function. This function also supports regular expressions for flexible name matching. The example shows matching 'example1' and any name starting with 'example', followed by a DELETE request to a specific object.

```python
import kmock
import re

async def test_kubernetes_object_name_filters(kmock: kmock.RawHandler) -> None:
    kmock[kmock.name('example1')] << 200
    kmock[kmock.name(re.compile('example.*'))] << 200

    resp = await kmock.delete('/apis/kopf.dev/v1/kopfexamples/example1')
    assert resp.status == 200
```

--------------------------------

### Mocking HTTP Servers with kmock

Source: https://github.com/nolar/kmock/blob/main/README.md

Demonstrates how to use kmock to define mock HTTP endpoints and verify client-side interactions. It covers setting up GET/POST handlers and asserting call counts and request data.

```python
import aiohttp
import kmock

async def function_under_test(base_url: str) -> None:
    async with aiohttp.ClientSession(base_url=base_url) as session:
        resp = await session.get('/')
        text = await resp.read()
        resp = await session.post('/hello', json={'name': text.decode()})
        data = await resp.json()
        return data

async def test_simple_http_server(kmock: kmock.RawHandler) -> None:
    # Setup the server side.
    kmock['get /'] << b'john'
    kmock['post /hello'] << (lambda req: {'you-are': req.params.get('name', 'anonymous')})
    never_called = kmock['/'] << b''

    # Work in the client side.
    data = await function_under_test(str(kmock.url))
    assert data == {'you-are': 'john'}

    # Check the server side.
    assert len(kmock) == 2
    assert len(kmock['get']) == 1
    assert len(kmock['post']) == 1
    assert kmock['post'][0].data == {'name': 'john'}
```

--------------------------------

### Stream responses from files and paths

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Shows how to serve content from local files. Using open() depletes the file content, whereas using pathlib.Path allows the file to be re-opened and served on every request.

```python
import kmock

async def test_responses_from_open_files(kmock: kmock.RawHandler, tmp_path) -> None:
    path = tmp_path / "file.txt"
    path.write_bytes(b'hello')
    kmock['/'] << open(str(path))

async def test_responses_from_paths(kmock: kmock.RawHandler, tmp_path) -> None:
    path = tmp_path / "file.txt"
    path.write_bytes(b'hello')
    kmock['/'] << path
```

--------------------------------

### Pre-populate Kubernetes objects in kmock

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/persistence.rst

Demonstrates how to initialize objects in the emulator by assigning them to the kmock.objects associative array using a resource, namespace, and name triplet.

```python
import kmock

async def test_object_prepopulation(kmock: kmock.KubernetesEmulator) -> None:
    kmock.objects['kopf.dev/v1/kopfexamples', 'ns1', 'name1'] = {'spec': 123}

    # Make sure it is accessible via the API:
    resp = await kmock.get('/apis/kopf.dev/v1/namespaces/ns1/kopfexamples/name1')
    data = await resp.json()
    assert rest.status == 200
    assert data == {'spec': 123}
```

--------------------------------

### Simulating Live Streaming Responses

Source: https://github.com/nolar/kmock/blob/main/README.md

Shows how to handle live streaming data and asynchronous updates in kmock. It demonstrates using lambdas and asyncio tasks to broadcast data to streaming requests.

```python
import datetime
import asyncio
import aiohttp
import freezegun
import kmock

@freezegun.freeze_time("2020-01-01T00:00:00")
async def test_k8s_out_of_the_box(kmock: kmock.RawHandler) -> None:
    kmock['/'] << (
        b'hello', lambda: asyncio.sleep(1), b', world!\n',
        {'key': 'val'},
        lambda: [(f"{i}…\n".encode(), asyncio.sleep(1)) for i in range(3)],
        ...  # live continuation
    )

    async def pulse():
        while True:
            # Broadcast to every streaming request (any method, any URL).
            kmock[...] << (lambda: datetime.datetime.now(tz=datetime.UTC).isoformat(), ...)
            await asyncio.sleep(1)

    asyncio.create_task(pulse())

    async with aiohttp.ClientSession(base_url='http://localhost', read_timeout=5) as session:
        resp = await session.get('/')
        text = await resp.read()

    assert text == b'hello, world!\n{"key": "val"}\n3…\n2…\n1…\n2020-01-01T00:00:05'
```

--------------------------------

### Serve responses from IO buffers

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Illustrates using io.StringIO or BytesIO to serve dynamic content. Note that buffers are consumed as they are read, requiring manual management if multiple requests are expected.

```python
import io
import kmock

async def test_responses_from_io(kmock: kmock.RawHandler) -> None:
    buffer = io.StringIO('prepared buffer')
    kmock['/'] << buffer
```

--------------------------------

### Pre-populate Object History in Kmock

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/persistence.rst

Shows how to initialize the emulator with a specific history for a resource by assigning a list of states to the kmock.objects key. This is useful for testing scenarios involving soft-deleted objects or specific state transitions.

```python
import kmock

async def test_history_prepopulation(kmock: kmock.KubernetesEmulator) -> None:
    kmock.objects['kopf.dev/v1/kopfexamples', 'ns1', 'name1'] = [{'spec': 123}, None]

    resp = await kmock.get('/apis/kopf.dev/v1/namespaces/ns1/kopfexamples/name1')
    assert rest.status == 404
```

--------------------------------

### Configure Resource Meta-information

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/discovery.rst

Demonstrates how to manually define resource metadata such as kind, shortnames, and verbs for a Kubernetes resource within the kmock scaffold. This approach automatically initializes ResourceInfo instances if they do not exist.

```python
import kmock

async def test_resource_information_field_by_field(kmock: kmock.KubernetesScaffold) -> None:
    kmock.resources['v1/pods'].kind = 'Pod'
    kmock.resources['v1/pods'].singular = 'pod'
    kmock.resources['v1/pods'].shortnames = {'po'}
    kmock.resources['v1/pods'].categories = {'category1', 'category2'}
    kmock.resources['v1/pods'].verbs = {'get', 'post', 'patch', 'delete'}
    kmock.resources['v1/pods'].subresources = {'status'}
    kmock.resources['v1/pods'].namespaced = True
```

--------------------------------

### Respond with Cookies in Python

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Demonstrates how to send cookies from the server to the client by wrapping a dictionary of cookie data with the `kmock.cookies` function. This is a simplified way to set cookies.

```python
import kmock

async def test_cookies_responses(kmock: kmock.RawHandler) -> None:
    kmock['/'] << kmock.cookies({'SessionId': '123abc'})

    resp = await kmock.get('/')
    assert resp.cookies['Session'].value == '123abc'
```

--------------------------------

### Create custom KMock fixtures

Source: https://github.com/nolar/kmock/blob/main/docs/usage.rst

Demonstrates how to define a custom pytest_asyncio fixture using RawHandler and Server classes to run mock servers in parallel.

```python
import pytest_asyncio
from kmock import RawHandler, Server

@pytest_asyncio.fixture
async def myk8s() -> AsyncIterator[RawHandler]:
    async with RawHandler() as handler, Server(handler, host='127.0.0.1', port=12345) as server:
        print(f"Listening on {server.url}")
        yield handler
```

--------------------------------

### Asserting Unexpected Errors with Kmock (Python)

Source: https://github.com/nolar/kmock/blob/main/docs/assertions.rst

Shows how to assert unexpected errors that occur within the mock server callbacks using kmock. The `kmock.RawHandler.errors` attribute, a list of exceptions, can be checked to ensure no errors occurred. This example demonstrates simulating an error and then asserting its presence and type in the errors list.

```python
import kmock

async def test_errors_assertions(kmock: kmock.RawHandler) -> None:
    kmock['/'] << ZeroDivisionError('boo!')

    resp = await kmock.get('/')
    assert resp.status == 500

    assert len(kmock.errors) == 1
    assert str(kmock.errors[0]) == 'boo!'
    assert isinstance(kmock.errors[0], ZeroDivisionError)
```

--------------------------------

### Assert Precise Object History with Soft Deletion (Python)

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/assertions.rst

This code example shows how to precisely assert the entire history of a Kubernetes object managed by kopf.Kmock. It covers creating, patching, and soft-deleting an object, then verifying that its history array exactly matches a predefined sequence of object states, including the `None` marker for deletion.

```python
import kmock

async def test_history_precisely(kmock: kmock.KubernetesEmulator) -> None:
    # Declare the resource as supported.
    await kmock.resources['kopf.dev/v1/kopfexamples'] = {}

    # Create and modify the resource object several times, then soft-delete it.
    await kmock.post('/apis/kopf.dev/v1/kopfexamples', json={'spec': 123, 'metadata': {'name': 'n1'}})
    await kmock.patch('/apis/kopf.dev/v1/kopfexamples/n1', json={'spec': 456})
    await kmock.patch('/apis/kopf.dev/v1/kopfexamples/n1', json={'spec': 789})
    await kmock.delete('/apis/kopf.dev/v1/kopfexamples/n1')

    # Check that the object's history contains at least these two versions.
    assert kmock.objects['kopf.dev/v1/kopfexamples', None, 'n1'].history == [
        {'spec': 123, 'metadata': {'name': 'n1'}},
        {'spec': 456, 'metadata': {'name': 'n1'}},
        {'spec': 789, 'metadata': {'name': 'n1'}},
        None,
    ]
```

--------------------------------

### Respond with Inferred HTTP Headers in Python

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Shows how to send response headers using a dictionary where keys are header names (prefixed with 'X-' or are well-known headers) and values are header content. This method infers that the dictionary represents headers.

```python
import kmock

async def test_inferred_headers_responses(kmock: kmock.RawHandler) -> None:
    kmock['/'] << {'X-My-Header': 'hello'}

    resp = await kmock.get('/')
    assert resp.headers['X-My-Header'] == 'hello'
```

--------------------------------

### Create dynamic responses with callables

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Demonstrates the use of callables (lambdas, functions) to generate responses dynamically based on the incoming request. The callable can optionally accept a kmock.Request object to tailor the output.

```python
import kmock

async def test_callable_responses(kmock: kmock.RawHandler) -> None:
    kmock['/greetings'] << (lambda req: f"Hello, {req.params.get('name', 'user')}!".encode())
```

--------------------------------

### Strict Mode Error Handling in Kmock (Python)

Source: https://github.com/nolar/kmock/blob/main/docs/assertions.rst

Explains and demonstrates kmock's strict mode for error handling. When `kmock.RawHandler.strict` is set to `True`, assertions on errors are performed during the test's teardown phase. This example shows a test that would pass during execution but fail during teardown due to a simulated `ZeroDivisionError`, highlighting how strict mode catches unhandled errors.

```python
import kmock
import pytest

@pytest.mark.kmock(strict=True)
async def test_errors_assertions(kmock: kmock.RawHandler) -> None:
    kmock['/'] << ZeroDivisionError('boo!')

    resp = await kmock.get('/')
    assert resp.status == 500
```

--------------------------------

### Define resource meta-information

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/discovery.rst

Illustrates how to provide extended metadata for Kubernetes resources, such as kind, shortnames, and verbs, using either ResourceInfo objects or plain dictionaries.

```python
import kmock

async def test_resource_information_as_one_object(kmock: kmock.KubernetesScaffold) -> None:
    kmock.resources['v1/pods'] = kmock.ResourceInfo(
        kind='Pod',
        singular='pod',
        shortnames={'po'},
        categories={'category1', 'category2'},
        verbs={'get', 'post', 'patch', 'delete'},
        subresources={'status'},
        namespaced=True,
    )

async def test_resource_information_as_one_dict(kmock: kmock.KubernetesScaffold) -> None:
    kmock.resources['v1/pods'] = {
        'kind': 'Pod',
        'singular': 'pod',
        'shortnames': {'po'},
        'categories': {'category1', 'category2'},
        'verbs': {'get', 'post', 'patch', 'delete'},
        'subresources': {'status'},
        'namespaced': True,
    }
```

--------------------------------

### Using Awaitable Primitives as KMock Responses

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Illustrates how KMock can use awaitable primitives like asyncio.Queue as responses. The test sets up a queue with delayed put operations and verifies that kmock retrieves messages in the order they are put.

```python
import kmock
import asyncio
from typing import Any

@pytest.mark.looptime
async def test_awaitable_response(kmock: kmock.RawHandler) -> None:
    queue: asyncio.Queue[Any] = asyncio.Queue()
    kmock['/'] << queue

    loop = asyncio.get_running_loop()
    loop.call_later(1, queue.put_nowait, b'hello')
    loop.call_later(3, queue.put_nowait, b'world')

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello'
    assert loop.time() == 1

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'world'
    assert loop.time() == 3
```

--------------------------------

### Configure KMock with TOML File

Source: https://github.com/nolar/kmock/blob/main/docs/configuration.rst

Demonstrates global KMock configuration using the `pyproject.toml` file. Options such as `kmock_cls`, `kmock_limit`, and `kmock_strict` can be specified under the `[tool.pytest]` section.

```toml
[tool.pytest]
kmock_cls = "kmock.RawHandler"
kmock_limit = 999
kmock_strict = true
```

--------------------------------

### Construct replayable streams

Source: https://github.com/nolar/kmock/blob/main/docs/streams.rst

Shows how to use tuples to create replayable streams that return the same data for every incoming request.

```python
import kmock

async def test_replayable_stream(kmock: kmock.RawHandler) -> None:
    kmock['/'] << (b'Served always!',)
    kmock['/'] << (b'Never happens!',)

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'Served always!'

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'Served always!'
```

--------------------------------

### Match Kubernetes Resources

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Explains how to use the kmock.resource wrapper or direct string specifiers to match Kubernetes resources by group, version, and plural name. Supports various notation styles for Core and custom APIs.

```python
import kmock

async def test_k8s_resource_specifiers(kmock: kmock.RawHandler) -> None:
    kmock[kmock.resource(group='kopf.dev', version='v1', plural='kopfexamples')] << 200
    kmock[kmock.resource('kopf.dev', 'v1', 'kopfexamples')] << 200
    kmock[kmock.resource('kopf.dev/v1/kopfexamples')] << 200
    kmock[kmock.resource('kopfexamples.v1.kopf.dev')] << 200
    kmock['kopf.dev/v1/kopfexamples'] << 200
    kmock['kopfexamples.v1.kopf.dev'] << 200

    resp = await kmock.get('/apis/kopf.dev/v1/kopfexamples')
    assert resp.status == 200
```

--------------------------------

### Define Request Matching Criteria in kmock

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Demonstrates how to match requests using HTTP methods and paths. Shows shorthand syntax for combining criteria and explicit wrappers for non-standard verbs or regex paths.

```python
import kmock
import re

async def test_matching_examples(kmock: kmock.RawHandler) -> None:
    # Method and path shorthand
    kmock['get /'] << b'hello'
    
    # HTTP methods
    kmock[kmock.method.GET] << b'hello'
    kmock[kmock.method('store')] << b'hello'
    
    # HTTP paths
    kmock['/greetings'] << b'hello'
    kmock[kmock.path(re.compile('/greetings/.*'))] << b'hello'
    
    # Combined method and path
    kmock['get /greetings'] << b'hello'
```

--------------------------------

### Verify Resource Discovery Endpoints

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/discovery.rst

Shows how to configure custom resource metadata and verify the output of the discovery endpoint. This ensures that the scaffold correctly reports resource information to external Kubernetes clients.

```python
import kmock

async def test_resource_information_discovery(kmock: kmock.KubernetesScaffold) -> None:
    kmock.resources['kopf.dev/v1/kopfexamples'].kind = 'KopfExample'
    kmock.resources['kopf.dev/v1/kopfexamples'].singular = 'kopfexample'
    kmock.resources['kopf.dev/v1/kopfexamples'].shortnames = {'kex'}
    kmock.resources['kopf.dev/v1/kopfexamples'].categories = {'category1', 'category2'}
    kmock.resources['kopf.dev/v1/kopfexamples'].verbs = {'get', 'post', 'patch', 'delete'}
    kmock.resources['kopf.dev/v1/kopfexamples'].subresources = {'status'}
    kmock.resources['kopf.dev/v1/kopfexamples'].namespaced = True

    resp = await kmock.get('/apis/kopf.dev/v1')
    data = await resp.read()
    assert data == {
        'apiVersion': 'v1',
        'kind': 'APIResourceList',
        'groupVersion': f'kopf.dev/v1',
        'resources': [
            {
                'name': f'kopfexamples',
                'kind': 'KopfExample',
                'singularName': 'kopfexample',
                'shortNames': ['kex'],
                'categories': ['category1', 'category2'],
                'verbs': ['get', 'post', 'patch', 'delete'],
                'namespaced': True,
            },
            {
                'name': f'kopfexamples/status',
                'kind': 'KopfExample',
                'singularName': 'kopfexample',
                'shortNames': ['kex'],
                'categories': ['category1', 'category2'],
                'verbs': ['get', 'post', 'patch', 'delete'],
                'namespaced': True,
            },
        ],
    }
```

--------------------------------

### Add resources via criteria

Source: https://github.com/nolar/kmock/blob/main/docs/kubernetes/discovery.rst

Shows how to add resources to the kmock scaffold using direct payload injection with criteria strings.

```python
import kmock

async def test_resource_adding_via_criteria(kmock: kmock.KubernetesScaffold) -> None:
    kmock['list kopf.dev/v1/kopfexamples'] << {'items': []}
```

--------------------------------

### Send binary and text responses with KMock

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Demonstrates sending raw binary blobs using bytes and UTF-8 encoded text using the kmock.text wrapper. It highlights the difference between raw bytes and JSON-encoded strings.

```python
import kmock

async def test_responses_from_open_files(kmock: kmock.RawHandler) -> None:
    kmock['/'] << b'hello'

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello'

async def test_raw_string_responses(kmock: kmock.RawHandler) -> None:
    kmock['/as-json'] << 'hello'
    kmock['/as-text'] << kmock.text('hello')
    kmock['/as-body'] << kmock.body(b'hello')
```

--------------------------------

### Return raw aiohttp responses in KMock

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Demonstrates how to bypass default KMock behavior by returning a low-level aiohttp.web.Response object. This is useful for fallback scenarios where fine-grained control over headers and status codes is required.

```python
import aiohttp.web
import kmock

async def test_explicit_responses(kmock: kmock.RawHandler) -> None:
    kmock['/'] << aiohttp.web.Response(
        status=404, reason='Not Found',
        headers={'My-Header': 'hello'},
        text='hello')

    resp = await kmock.get('/')
    text = await resp.read()
    assert resp.status == 404
    assert resp.reason == 'Not Found'
    assert resp.headers['My-Header'] == 'hello'
    assert text == b'hello'
```

--------------------------------

### Stream from local files using open()

Source: https://github.com/nolar/kmock/blob/main/docs/streams.rst

Demonstrates streaming content from a local file using the built-in open function. Note that this method is depletable, meaning subsequent requests will receive an empty stream once the file pointer reaches the end.

```python
import kmock

async def test_responses_from_open_files(kmock: kmock.RawHandler, tmp_path) -> None:
    path = tmp_path / "file.txt"
    path.write_bytes(b'hello')
    kmock['/'] << (open(str(path)), b'//end')

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello//end'

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'//end'  # the file has been depleted, but the stream is not
```

--------------------------------

### Configure Global KMock Defaults with Fixture (Python)

Source: https://github.com/nolar/kmock/blob/main/docs/configuration.rst

Shows how to set global default configurations for KMock across an entire test suite using a pytest fixture in `conftest.py`. The fixture must return a dictionary of options.

```python
import kmock
import pytest

@pytest.fixture
def kmock_options() -> dict[str, Any]:
    return dict(port=12345, cls=kmock.RawHandler)
```

--------------------------------

### Configure KMock with Pytest Marks (Python)

Source: https://github.com/nolar/kmock/blob/main/docs/configuration.rst

Demonstrates how to apply KMock configuration using pytest marks, both at the module level and for individual tests. This allows for flexible and granular control over KMock's behavior during test execution.

```python
import kmock

# Module-level config applies to all tests in the module.
pytestmark = [pytest.mark.kmock(port=12345, cls=kmock.RawHandler)]

# Individual tests can override the module-level or default configs.
@pytest.mark.kmock(port=23456, hostnames=['google.com'])
async def test_with_overridden_config(kmock: kmock.RawHandler) -> None:
    assert kmock.port == 23456
```

--------------------------------

### Construct explicit structured responses

Source: https://github.com/nolar/kmock/blob/main/docs/responses.rst

Demonstrates the use of the kmock.Response class to define custom status codes, reasons, and headers for fine-grained control over the HTTP response.

```python
import kmock

async def test_explicit_responses(kmock: kmock.RawHandler) -> None:
    kmock['/'] << kmock.Response(
        status=404, reason='Not Found',
        headers={'My-Header': 'hello'}
    )
```

--------------------------------

### Use Predefined Infinite Priorities in KMock (Python)

Source: https://github.com/nolar/kmock/blob/main/docs/requests.rst

Illustrates the use of `.fallback` (negative infinity priority) and `.override` (positive infinity priority) for convenient rule management. These allow for guaranteed lowest or highest priority rules, respectively.

```python
import kmock
import re

async def test_infinite_priorities(kmock: kmock.RawHandler) -> None:
    # Define the prioritised and non-priorities responses.
    kmock['/greetings'] << b'never served because there is an override below'
    kmock.fallback[re.compile(r'.*')] << 404
    kmock.override['/greetings'] << b'hello'

    # Try the catch-all rule for all URLs.
    resp = await kmock.get('/')
    assert resp.status == 404

    # Try the specifically defined overridden URL.
    resp = await kmock.get('/greetings')
    text = await resp.read()
    assert text == b'hello'
```

--------------------------------

### Implement Live Stream Feeding

Source: https://github.com/nolar/kmock/blob/main/docs/streams.rst

Shows how to use the Ellipsis (...) operator to create feeding points in a stream. This allows for asynchronous injection of data into a request stream while it is being consumed.

```python
import aiohttp
import kmock
import asyncio

async def consume_stream(url: str) -> None:
    async with aiohttp.ClientSession() as session:
        resp = session.get(url)
        resp.raise_for_status()
        for chunk in resp.content.iter_chunked():
            print(f"{chunk!r}")

async def test_stream_feeding(kmock: kmock.RawHandler) -> None:
    stream = kmock['get /'] << ('Hello!', ..., 'Good bye!')
    asyncio.create_task(consume_stream(str(kmock.url)))

    stream << b'Countdown:\n' << ...
    for i in range(3, 0, -1):
        stream << i << ...
        await asyncio.sleep(1)

    stream << b''
```