### Mocking GET Requests with Kmock in Python

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Demonstrates how to use the kmock library to mock GET requests and assert the response content. This is useful for testing API interactions without making actual network calls. It shows basic usage with query parameters and default responses.

```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!"
```

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

### Configure KMock via configuration files

Source: https://kmock.readthedocs.io/_sources/configuration.rst.txt

Provides examples of setting global KMock parameters using standard pytest configuration files in INI and TOML formats.

```ini
# pytest.ini
[pytest]
kmock_cls = kmock.RawHandler
kmock_limit = 999
kmock_strict = true
```

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

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

### Using asyncio.Queue as a lazy response in kmock

Source: https://kmock.readthedocs.io/responses

Demonstrates how to use an asyncio.Queue as a response provider in a kmock test. The example shows how items put into the queue at specific intervals are returned as responses to HTTP GET requests.

```python
import kmock
import asyncio
import pytest
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
```

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

### Simulate Kubernetes Streaming Events with Curl

Source: https://kmock.readthedocs.io/intro

This section provides command-line examples using `curl` to interact with a KMock server simulating Kubernetes API responses. It demonstrates making POST and GET requests, including a watch request that streams events over time. The output shows the HTTP status codes, headers, and the streamed JSON payloads, illustrating the simulated ADDED, MODIFIED, and DELETED events.

```bash
$ curl -i -X POST http://localhost:12345/apis/kopf.dev/v1/namespaces/ns1/kopfexamples
HTTP 404 Not Found

{"error": "not served"}

$ curl -i http://localhost:12345/apis/kopf.dev/v1/namespaces/ns1/kopfexamples
HTTP 200 OK

{'items': [], 'metadata': {'resourceVersion': 'v1'}}

$ 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]')
[2020-12-31 23:59:56] HTTP 200 OK
[2020-12-31 23:59:56]
[2020-12-31 23:59:56] {'type': 'ADDED', 'object': {'spec': {}}}
[2020-12-31 23:59:59] {'type': 'MODIFIED', 'object': {'spec': {'time': '2020-12-31T23:59:59.000Z'}}}
[2020-12-31 23:59:59] {'type': 'DELETED', 'object': {'metadata': {'name': f'demo0'}}}
[2020-12-31 23:59:59] {'type': 'DELETED', 'object': {'metadata': {'name': f'demo1'}}}
[2020-12-31 23:59:59] {'type': 'DELETED', 'object': {'metadata': {'name': f'demo2'}}}
[2020-12-31 23:59:59] {'type': 'ERROR', 'object': {'code': 410}}

```

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

### Respond with cookies in Python

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Demonstrates how to send a simplified cookie from the server to the client by wrapping it with the `kmock.cookies` function. The example shows setting a cookie and then asserting its value after making a request.

```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'
```

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

### Match Kubernetes Actions with kmock

Source: https://kmock.readthedocs.io/requests

Demonstrates how to match Kubernetes actions like 'list', 'watch', 'fetch', 'create', and 'update' using the kmock.action() instance. It highlights that 'delete' requires explicit wrapping due to potential conflicts with HTTP methods. The example shows setting up mock responses for these actions.

```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 Actions and Resources with kmock

Source: https://kmock.readthedocs.io/requests

Shows how to match both a Kubernetes action and its associated resource in a single string, separated by a space. The action should precede the resource. Examples include matching 'list pods.v1' and 'watch kopfexamples.v1.kopf.dev'.

```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

```

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

### Execute and Access KMock Server

Source: https://kmock.readthedocs.io/usage

Commands to start the Python KMock server and interact with it using curl.

```bash
$ python server.py

$ curl -i http://localhost:12345/
HTTP 200 OK

Hello, there!
```

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

### Standalone KMock Server Setup and Request Handling (Python)

Source: https://kmock.readthedocs.io/_sources/usage.rst.txt

This snippet demonstrates how to set up a standalone KMock server that listens on a specific port, defines a response for the root endpoint, and makes an initial request to an external service using its embedded client. It then waits indefinitely, allowing external clients to interact with the mock server.

```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())
```

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

### Access KMock Server and Client Parameters for URL Construction (Python)

Source: https://kmock.readthedocs.io/_sources/usage.rst.txt

This example shows how to access the underlying aiohttp server and client objects provided by the KMock fixture. It demonstrates retrieving the server's URL and port to construct dynamic request URLs within tests. This is useful for tests that need to interact with the mock server using generated or dynamic endpoints.

```python
import kmock

async def test_server_parameters(kmock: kmock.RawHandler) -> None:
    url = kmock.server.make_url('/path?q=params')
    assert str(url).startswith('http://127.0.0.1:')
    assert url.port >= 1024
    assert url.path == '/path'
```

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

### Using Asyncio Queue as a Kmock Response in Python

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Illustrates how to use an asyncio.Queue as a response for a mocked endpoint in Kmock. This allows simulating responses that are produced over time. The example uses `call_later` to add items to the queue and asserts the order and content of received responses.

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

@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
```

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

### HTTP Method Matching

Source: https://kmock.readthedocs.io/requests

Demonstrates how to match requests based on their HTTP method (e.g., GET, POST). It covers using enum values, string representations, and wrapping non-standard verbs.

```APIDOC
## HTTP Criteria - Matching HTTP Methods

A `kmock.method` instance (a string enum) is matched strictly against HTTP verbs of the request (case-insensitive). Strings `"get"`, `"post"`, `"patch"`, `"put"`, `"delete"`, `"options"`, `"head"` are automatically recognized as HTTP verbs and require no wrappers.

### Example
```python
import kmock

async def test_http_methods(kmock: kmock.RawHandler) -> None:
    # Enum values can be used directly.
    kmock[kmock.method.GET] << b'hello'

    # Standard verbs are recognized are simple string.
    kmock['get'] << b'hello'

    # Non-standard verbs MUST be wrapped.
    kmock[kmock.method('store')] << b'hello'
```
```

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

### Simulate Kubernetes Watch Events with curl (Shell)

Source: https://kmock.readthedocs.io/_sources/intro.rst.txt

This shell command sequence demonstrates how to interact with a kmock-simulated Kubernetes API server from the command line. It shows POST requests for creation, GET requests for listing, and WATCH requests that stream events like ADDED, MODIFIED, DELETED, and ERROR over time.

```shell
$ curl -i -X POST http://localhost:12345/apis/kopf.dev/v1/namespaces/ns1/kopfexamples
HTTP 404 Not Found

{"error": "not served"}

$ curl -i http://localhost:12345/apis/kopf.dev/v1/namespaces/ns1/kopfexamples
HTTP 200 OK

{'items': [], 'metadata': {'resourceVersion': 'v1'}}

$ 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]')
[2020-12-31 23:59:56] HTTP 200 OK
[2020-12-31 23:59:56]
[2020-12-31 23:59:56] {'type': 'ADDED', 'object': {'spec': {}}}
[2020-12-31 23:59:59] {'type': 'MODIFIED', 'object': {'spec': {'time': '2020-12-31T23:59:59.000Z'}}},
[2020-12-31 23:59:59] {'type': 'DELETED', 'object': {'metadata': {'name': f'demo0'}}}
[2020-12-31 23:59:59] {'type': 'DELETED', 'object': {'metadata': {'name': f'demo1'}}}
[2020-12-31 23:59:59] {'type': 'DELETED', 'object': {'metadata': {'name': f'demo2'}}}
[2020-12-31 23:59:59] {'type': 'ERROR', 'object': {'code': 410}}
```

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

### Python: Streaming Requests with kmock Context Managers

Source: https://kmock.readthedocs.io/_sources/usage.rst.txt

Demonstrates how to use kmock's embedded clients to handle streaming requests. The stream is initiated on entering a context manager and its content can be consumed as long as the context manager remains open. This example configures a timestamp generator and consumes the stream for a specified duration.

```python
import datetime
import kmock
from typing import AsyncIterator
import asyncio

async def test_streaming_requests(kmock: kmock.RawHandler) -> None:
    # The stream generator to stream the current time every second.
    # It exits by CancelledError from `await` when the connection is closed by the client side.
    async def timestamp_generator() -> AsyncIterator[kmock.Payload]:
        while True:
            now = datetime.datetime.now(tz=datetime.UTC)
            yield now.isoformat().encode() + b'\n'
            await asyncio.sleep(1)

    # Configure the stream at the root URL (all methods).
    kmock['/'] << timestamp_generator

    # Consume the stream for 3 seconds, close the connection on exiting the ctx manager:
    async with kmock.get('/') as resp:
        # Let the server send something for some time.
        await asyncio.sleep(3)

        # Consume all that was sent to the moment.
        buffer: bytes = resp.content.read_nowait()

    # Depending on your luck, it is 3±1 lines in the buffer.
    assert 2 <= len(buffer.splitlines()) <= 4
```

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

### Streaming Requests with kmock (Python)

Source: https://kmock.readthedocs.io/usage

Demonstrates how to perform streaming requests using kmock, similar to aiohttp's context manager pattern. The request is initiated upon entering the context manager, and the streamed content can be consumed while the manager is active. This example sets up a generator to stream timestamps and consumes the stream for a duration.

```python
import datetime
import kmock
from typing import AsyncIterator
import asyncio

async def test_streaming_requests(kmock: kmock.RawHandler) -> None:
    # The stream generator to stream the current time every second.
    # It exits by CancelledError from `await` when the connection is closed by the client side.
    async def timestamp_generator() -> AsyncIterator[kmock.Payload]:
        while True:
            now = datetime.datetime.now(tz=datetime.UTC)
            yield now.isoformat().encode() + b'\n'
            await asyncio.sleep(1)

    # Configure the stream at the root URL (all methods).
    kmock['/'] << timestamp_generator

    # Consume the stream for 3 seconds, close the connection on exiting the ctx manager:
    async with kmock.get('/') as resp:
        # Let the server send something for some time.
        await asyncio.sleep(3)

        # Consume all that was sent to the moment.
        buffer: bytes = resp.content.read_nowait()

    # Depending on your luck, it is 3±1 lines in the buffer.
    assert 2 <= len(buffer.splitlines()) <= 4

```

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

### Respond using files and paths in KMock

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Illustrates the difference between using open() file handles, which deplete after one request, and pathlib.Path objects, which re-open the file on every request.

```python
import kmock
from pathlib import 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

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

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

### Perform Simple HTTP Requests Using KMock Fixture Client (Python)

Source: https://kmock.readthedocs.io/_sources/usage.rst.txt

This snippet demonstrates how to use the KMock fixture's embedded client to perform simple HTTP requests. It shows how to mock a response for a specific path using the `<<` operator and then make a GET request to that path. The example verifies the response status code and parses the JSON body.

```python
import kmock

async def test_simple_requests(kmock: kmock.RawHandler) -> None:
    kmock['/'] << 444 << b'{"key": "val"}'

    resp = await kmock.get('/')
    data = await resp.json()
    assert resp.status == 444
    assert data == {'key': 'val'}
```

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

### Configure KMock Server Host and Hostnames via Pytest Mark (Python)

Source: https://kmock.readthedocs.io/_sources/usage.rst.txt

This snippet illustrates how to configure KMock server parameters like port and hostnames using pytest marks. The example shows a test function where the 'kmock' fixture is marked to listen on port 12345 and only respond to requests for 'google.com'. It also includes a basic example of mocking a response and making a request.

```python
import pytest

@pytest.mark.kmock(port=12345, hostnames=['google.com'])
def test_me(kmock):
    kmock << b'Hello! I am not Google.'
    resp = await kmock.get('http://google.com/?q=search')
    text = await resp.read()
    assert text == b'Hello! I am not Google.'
```

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

### Kmock Server Initialization and Usage with aiohttp

Source: https://kmock.readthedocs.io/packages/kmock

Demonstrates initializing and using the Kmock Server class with aiohttp. The Server class acts as a wrapper for handlers, providing networking capabilities. It can be used as a context manager and supports routing traffic through multiple server instances.

```python
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.5 aiohttp/3.13.3'_)[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_)
      * **host** (_str_)
      * **port** (_int_ _|_None_)
      * **server_cls** (_type_ _[__RawTestServer_ _]_)
      * **client_fct** (_Callable_ _[__[__URL_ _]__,__ClientSession_ _]_)
    
```

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

### GET / Kubernetes Discovery Endpoints

Source: https://kmock.readthedocs.io/kubernetes/discovery

Standard Kubernetes discovery endpoints exposed by the scaffold and emulator to provide cluster information and resource metadata.

```APIDOC
## GET /version

### Description
Returns the version information of the Kubernetes cluster.

### Method
GET

### Endpoint
/version

### Response
#### Success Response (200)
- **version** (object) - Kubernetes version details.

---

## GET /api

### Description
Returns the list of available API versions for the core Kubernetes API.

### Method
GET

### Endpoint
/api

### Response
#### Success Response (200)
- **versions** (array) - List of supported core API versions.

---

## GET /apis

### Description
Returns the list of available API groups and versions in the cluster.

### Method
GET

### Endpoint
/apis

### Response
#### Success Response (200)
- **groups** (array) - List of available API groups.

---

## GET /apis/{group}/{version}

### Description
Returns the list of resources available for a specific API group and version.

### Method
GET

### Endpoint
/apis/{group}/{version}

### Parameters
#### Path Parameters
- **group** (string) - Required - The API group name (e.g., kopf.dev).
- **version** (string) - Required - The API version (e.g., v1).

### Response
#### Success Response (200)
- **resources** (array) - List of resources available under this group/version.
```

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

### Chained Views in KMock

Source: https://kmock.readthedocs.io/packages/kmock

A base view for other views that operate on a single source, forming a chain. This is a fundamental building block for more complex filtering and reaction setups.

```python
class Chained(View):
    def __init__(self, source: View):
        pass
```

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

### Server Configuration

Source: https://kmock.readthedocs.io/packages/kmock

Configuring and initializing a Kmock server instance to handle incoming HTTP traffic.

```APIDOC
## Server Initialization

### Description
Initializes a server instance to listen on a TCP/HTTP port and marshal requests to the handler.

### Parameters
- **handler** (RawHandler) - Required - The handler instance to process requests.
- **host** (str) - Optional - Host address (default: '127.0.0.1').
- **port** (int) - Optional - Port number.
- **server_cls** (type) - Optional - Server class implementation.

### Usage Example
```python
async with RawHandler() as kmock, Server(kmock) as srv:
    await srv.client.get('/')
```
```

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

### Using Pytest Fixtures for Kubernetes Mocking

Source: https://kmock.readthedocs.io/usage

Demonstrates how to use the built-in 'kmock' fixture for Kubernetes object manipulation and how to define custom fixtures using 'RawHandler' and 'Server' for advanced testing scenarios.

```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')
    data = await resp.json()
    assert kmock.Object(data) >= {'items': [{'spec': 123}]}
```

```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
```

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

### Slicing Requests in KMock

Source: https://kmock.readthedocs.io/packages/kmock

The Slicer class allows for selecting a subset of requests using list-like slicing notation (start:stop:step). This is useful for accessing specific requests in a sequence.

```python
kmock['get'][1:]
```

```python
kmock['post'][:3]
```

```python
kmock['/'][::2]
```

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

### Configure KMock global defaults via fixture

Source: https://kmock.readthedocs.io/_sources/configuration.rst.txt

Shows how to set global KMock options by defining a kmock_options fixture in a conftest.py file.

```python
import kmock
import pytest

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

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

### Configure Kubernetes resource metadata field-by-field

Source: https://kmock.readthedocs.io/_sources/kubernetes/discovery.rst.txt

Demonstrates how to manually define resource attributes such as kind, singular name, shortnames, and verbs for a specific API version. This approach creates a ResourceInfo instance automatically if one does not already 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
```

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

### KMock Python: Delayed headers in streaming responses

Source: https://kmock.readthedocs.io/_sources/caveats.rst.txt

This code example illustrates how KMock handles delayed headers in streaming responses. The headers are sent only when the first non-None binary payload appears, after an initial delay.

```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'
```

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

### Respond from Local Files using open()

Source: https://kmock.readthedocs.io/responses

Send a response from a local file using the built-in `open()` function. This file is consumed globally for all requests. Be aware that open files can be depleted after being read once, unless content is appended.

```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))

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

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b''  # the file has been depleted
```

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

### Python: Lazy side effects with callables using kmock

Source: https://kmock.readthedocs.io/_sources/effects.rst.txt

Demonstrates using a lambda function with time.sleep as a lazy side effect in kmock. This example shows how to simulate server-side delays and capture the response.

```python
import kmock
import time

async def test_callable_side_effect(kmock: kmock.RawHandler, chronometer) -> None:
    # Simulate the slow behavior of the server side.
    kmock['/'] >> (lambda: time.sleep(1)) << b'hello'

    with chronometer:
        resp = await kmock.get('/')
        text = await resp.read()

    assert 0.8 <= chronometer.seconds <= 1.2  # 1s ± 0.2s uncertainty
    assert text == b'hello'
```

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

### Python: Lazy side effects with awaitables using kmock and asyncio.Queue

Source: https://kmock.readthedocs.io/_sources/effects.rst.txt

Illustrates using an asyncio.Queue as a lazy side effect with kmock. This example shows how to put data into a queue asynchronously when requests are made to a specific endpoint.

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

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

    await kmock.post('/', data=b'hello')
    await kmock.post('/', data=b'world')

    assert queue.qsize() == 2
    body1 = queue.get_nowait()
    body2 = queue.get_nowait()
    assert body1 == b'hello'
    assert body2 == b'world'
```

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

### KMock Python: Instant headers in streaming responses

Source: https://kmock.readthedocs.io/_sources/caveats.rst.txt

This code example demonstrates KMock's behavior with instant headers in streaming responses. The response headers are sent immediately with the first (even if empty) binary payload, followed by a delay before the actual content.

```python
import kmock
import asyncio

async def test_instant_headers_in_stream(kmock: kmock.RawHandler) -> None:
    kmock['get /'] << (
        b'',  # <-- this is the only difference
        lambda: asyncio.sleep(5),
        b'hello'
    )
    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello'
```

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

### Respond with explicit Response objects

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Shows how to manually construct a full HTTP response including status codes and reason phrases using the kmock.Response class.

```python
import kmock

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

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

### Accessing Standalone KMock Server (Shell)

Source: https://kmock.readthedocs.io/_sources/usage.rst.txt

This snippet shows how to access the standalone KMock server using curl from the command line. It demonstrates making a GET request to the root endpoint of the locally running mock server and displays the expected HTTP response.

```shell
$ curl -i http://localhost:12345/
HTTP 200 OK

Hello, there!
```

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

### Create Custom K8s Mock Server Fixture with RawHandler and Server (Python)

Source: https://kmock.readthedocs.io/_sources/usage.rst.txt

This snippet shows how to create a custom pytest fixture that uses KMock's RawHandler and Server to set up a mock Kubernetes server. It demonstrates how to manage the server lifecycle and yield the handler for use in tests. This is useful for advanced testing scenarios or when needing fine-grained control over the mock server.

```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 - Python

Source: https://kmock.readthedocs.io/_sources/assertions.rst.txt

Shows how to assert that no unexpected errors occurred during the execution of mock requests. The `kmock.RawHandler.errors` attribute provides a list of exceptions that were raised by the mock server's callbacks. This example demonstrates how to check if this list is empty and to inspect the details of any captured errors.

```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)
```

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

### Respond with inferred headers in Python

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Illustrates how to send response headers using a dictionary where keys are header names (prefixed with 'X-' or well-known headers) and values are header values. It shows how to make a request and assert the presence and value of a custom header.

```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'
```

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

### Wrap Arbitrary Data for Responses in Python

Source: https://kmock.readthedocs.io/responses

This example demonstrates using the `kmock.data()` wrapper to send arbitrary JSON-serializable values, preventing them from being misinterpreted (e.g., integers in the 100-999 range as status codes). It asserts that the response status is 200 and the JSON data is correctly received.

```python
import kmock

async def test_wrapped_int_responses(kmock: kmock.RawHandler) -> None:
    kmock['/'] << kmock.data(404)

    resp = await kmock.get('/')
    text = await resp.read()
    data = await resp.json()
    assert resp.status == 200
    assert text == b'404'
    assert data == 404
```

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

### Respond from Local Files using pathlib.Path

Source: https://kmock.readthedocs.io/responses

Send a response from a local file by using `pathlib.Path`. This method ensures the file is re-opened for every incoming request, preventing depletion issues seen with `open()`.

```python
import kmock

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

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

    resp = await kmock.get('/')
    text = await resp.read()
    assert text == b'hello'  # the file is re-opened again
```

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

### Match HTTP Query Parameters in Python with Kmock

Source: https://kmock.readthedocs.io/_sources/requests.rst.txt

Illustrates matching HTTP query parameters using Kmock. It shows how to use string syntax or dictionaries, and explains the behavior of Ellipsis for any value.

```python
import kmock

async def test_http_query_params(kmock: kmock.RawHandler) -> None:
    # Explicit wrapping checks against params regardless of keys' names.
    kmock[kmock.params('name=john&mode=formal')] << b'Greetings, John!'

    # Dicts are checked against params only if they do NOT look like headers.
    kmock[{'name': 'john', 'mode': 'formal'}] << b'Greetings, John!'
```

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

### Strict Mode Error Handling with kmock and pytest in Python

Source: https://kmock.readthedocs.io/assertions

Demonstrates kmock's strict mode for error handling, where errors are raised during test teardown if not explicitly handled. This example uses pytest and shows how an unhandled `ZeroDivisionError` in strict mode causes a test failure during teardown.

```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

```

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

### Implement streaming responses with kmock

Source: https://kmock.readthedocs.io/_sources/streams.rst.txt

Demonstrates how to create replayable streams and streams with continuations using the kmock library. Continuations allow the stream to remain open and accept new data injections.

```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'
```

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

### Respond with JSON data in Python

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Shows how to respond with JSON payloads by using standard Python types like strings, integers, floats, booleans, lists, and dictionaries. It includes an example of sending a complex JSON object and then asserting both the raw response body and the parsed JSON data.

```python
import kmock

async def test_json_responses(kmock: kmock.RawHandler) -> None:
    kmock['/'] << {
        'int': 123,
        'bool': True,
        'float': 123.456,
        'string': 'hello',
        'list': [123, 456],
    }

    resp = await kmock.get('/')
    text = await resp.read()
    data = await resp.json()
    assert text == b'{"int": 123, "bool": true, "float": 123.456, "string": "hello", "list": [123, 456]}'
    assert data == {"int": 123, "bool": True, "float": 123.456, "string": "hello", "list": [123, 456]}
```

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

### Python: Simulate Live Streams with Kmock Feeding Points

Source: https://kmock.readthedocs.io/_sources/streams.rst.txt

Illustrates how to set up and feed live streams using kmock's '...' (Ellipsis) feeding points. This allows simulating real-time data feeds in tests, where the stream can block and wait for new data. It covers both simple and complex feeding scenarios.

```python
import aiohttp
import kmock
import asyncio


# A sample system-under-test that consumes the stream and prints it:
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:
    # Setup the endpoints and streams:
    stream = kmock['get /'] << ('Hello!', ..., 'Good bye!')

    # Initiate the background reading from the stream:
    asyncio.create_task(consume_stream(str(kmock.url)))

    # Slowly feed some simulated payload into the stream:
    stream << b'Countdown:\n' << ...
    for i in range(3, 0, -1):
        # Feed an additional feeding point on every iteration:
        stream << i << ...

        # Sleep to simulate the slowly going process:
        await asyncio.sleep(1)

    # Finish the stream and close the connection (because no new feeding point).
    stream << b''
```

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

### Respond with HTTP status codes and body in Python

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Demonstrates how to set a custom HTTP status code (e.g., 404) and a response body (e.g., b'hello') for a given URL using kmock. It then shows how to make a request to that URL and assert the received status code and body.

```python
import kmock

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

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

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

### Lazy Dynamic Streams with Callables in kmock

Source: https://kmock.readthedocs.io/streams

Shows how to use callables (including lambdas and async functions) within a stream to generate content dynamically per request. These callables can accept a kmock.Request object to customize the output based on request parameters. This example includes simulating a delay using asyncio.sleep.

```python
import kmock
import asyncio

async def test_callables_in_streams(kmock: kmock.RawHandler) -> None:
    kmock['/greetings'] << (
        lambda: asyncio.sleep(1),  # some delay to simulate heavy thinking
        b"Hello, "
        lambda req: req.params.get('name', 'user'),
        b"!",
    )

    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!"'

```

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

### Match Kubernetes resources

Source: https://kmock.readthedocs.io/requests

Illustrates matching Kubernetes requests by resource type using kmock.resource or shorthand string notations. Supports various API group/version/plural formats.

```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
```

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

### Respond with wrapped headers in Python

Source: https://kmock.readthedocs.io/_sources/responses.rst.txt

Explains how to explicitly wrap response headers using `kmock.headers` to ensure they are sent as headers, regardless of their names. This method is useful for custom headers or when header names might not follow standard conventions. It includes an example of making a request and asserting a wrapped header.

```python
import kmock

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

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

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

### Mock Kubernetes resource listing with KMock

Source: https://kmock.readthedocs.io/_sources/intro.rst.txt

Shows how to use the KubernetesEmulator to manage in-memory state of K8s objects. It demonstrates populating objects across namespaces and verifying list responses using partial dictionary comparison.

```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}]}
```