# openshift-python-wrapper

> Manage OpenShift and Kubernetes resources with clean, Pythonic code instead of raw API calls

---

Source: quickstart.md

Let me explore the repository structure and key files to understand the codebase before writing the documentation.

Now let me look at the `NamespacedResource` class and the `get` classmethod for listing resources:

Now let me check the `get_client` with the `utils/client_config.py` for understanding connection options:

Now I have a thorough understanding of the codebase. Let me write the documentation:

Now I have a comprehensive understanding of the codebase. Let me write the guide:

# Installing and Creating Your First Resource

Get up and running with openshift-python-wrapper so you can programmatically create, query, and delete Kubernetes and OpenShift resources from Python. This guide takes you from installation to your first working resource in under 5 minutes.

## Prerequisites

- Python 3.10 or later
- Access to a Kubernetes or OpenShift cluster (with a valid kubeconfig)
- `pip` or another Python package manager

## Quick Example

```bash
pip install openshift-python-wrapper
```

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

with Namespace(client=client, name="my-first-namespace") as ns:
    ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
    print(f"Namespace {ns.name} is active: {ns.exists}")
# Namespace is automatically deleted when the block exits
```

That's it — the namespace is created when entering the `with` block and cleaned up when leaving it.

## Step 1: Install the Package

```bash
pip install openshift-python-wrapper
```

> **Tip:** For isolated environments, use a virtual environment:
> ```bash
> python -m venv .venv && source .venv/bin/activate
> pip install openshift-python-wrapper
> ```

## Step 2: Connect to Your Cluster

The `get_client()` function returns a dynamic client connected to your cluster. By default it reads your kubeconfig from the `KUBECONFIG` environment variable or `~/.kube/config`:

```python
from ocp_resources.resource import get_client

client = get_client()
```

You can also point to a specific kubeconfig file:

```python
client = get_client(config_file="/path/to/kubeconfig")
```

Or connect with a token and host directly:

```python
client = get_client(host="https://api.mycluster.example.com:6443", token="sha256~...")
```

> **Note:** For the full set of connection options — including basic auth, in-cluster config, and environment variables — see [Connecting to Clusters](connecting-to-clusters.html).

## Step 3: Create a Resource

There are two ways to create a resource: explicitly with `deploy()`/`clean_up()`, or automatically with a context manager (`with` statement).

### Option A: Context Manager (Recommended)

The context manager automatically deletes the resource when the block exits, which prevents leftover resources:

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

with Namespace(client=client, name="namespace-example-2") as ns:
    ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
    assert ns.exists
# Resource is automatically cleaned up here
```

### Option B: Explicit Deploy and Clean Up

Use `deploy()` and `clean_up()` when you need more control over the lifecycle:

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

ns = Namespace(client=client, name="namespace-example-1")
ns.deploy()
assert ns.exists
ns.clean_up()
```

> **Tip:** Set `teardown=False` on the resource to prevent `clean_up()` from being called when using a context manager. This is useful when you want the resource to persist after the block exits.

## Step 4: Create a Namespaced Resource

Most Kubernetes resources (Pods, ConfigMaps, Deployments, etc.) live inside a namespace. These require both `name` and `namespace`:

```python
from ocp_resources.config_map import ConfigMap
from ocp_resources.resource import get_client

client = get_client()

with ConfigMap(
    client=client,
    name="app-config",
    namespace="default",
    data={"app.properties": "debug=true\nport=8080"},
) as cm:
    assert cm.exists
    print(f"ConfigMap {cm.name} created in namespace {cm.namespace}")
```

## Step 5: Query Resources

Use the `get()` class method to list resources from the cluster. It returns a generator you can iterate over:

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

for pod in Pod.get(client=client, namespace="default"):
    print(pod.name)
```

Filter by labels using `label_selector`:

```python
for pod in Pod.get(client=client, label_selector="app=nginx"):
    node = pod.node
    print(f"Pod {pod.name} is running on node {node.name}")
```

Check if a specific resource exists:

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

ns = Namespace(client=client, name="default")
if ns.exists:
    print("Namespace exists!")
```

> **Note:** For more on querying, filtering, and watching resources, see [Querying and Watching Resources](querying-resources.html).

## Step 6: Delete a Resource

If you didn't use a context manager, call `clean_up()` to delete the resource:

```python
ns = Namespace(client=client, name="my-temp-namespace")
ns.deploy()
# ... do work ...
ns.clean_up()
```

The `clean_up()` method waits for the resource to be fully deleted by default. Pass `wait=False` to return immediately:

```python
ns.clean_up(wait=False)
```

> **Note:** For full details on resource lifecycle management, see [Creating and Managing Resources](creating-and-managing-resources.html).

## Putting It All Together

Here's a complete script that creates a namespace, deploys a ConfigMap inside it, reads it back, and cleans everything up:

```python
from ocp_resources.config_map import ConfigMap
from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

with Namespace(client=client, name="quickstart-demo") as ns:
    ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)

    with ConfigMap(
        client=client,
        name="demo-config",
        namespace="quickstart-demo",
        data={"greeting": "hello from openshift-python-wrapper"},
    ) as cm:
        assert cm.exists
        print(f"Created ConfigMap '{cm.name}' in namespace '{ns.name}'")

        # Query it back
        for config_map in ConfigMap.get(client=client, namespace="quickstart-demo"):
            print(f"  Found: {config_map.name}")
    # ConfigMap cleaned up here
# Namespace cleaned up here
```

## Advanced Usage

### Creating Resources from YAML

You can create any resource from an existing YAML file instead of specifying parameters in Python:

```python
from ocp_resources.config_map import ConfigMap
from ocp_resources.resource import get_client

client = get_client()

cm = ConfigMap(client=client, yaml_file="my-configmap.yaml")
cm.deploy()
```

### Using a Fake Client for Testing

You can run your code without a live cluster by using a fake client — useful for unit tests and local development:

```python
from ocp_resources.config_map import ConfigMap
from ocp_resources.resource import get_client

client = get_client(fake=True)

with ConfigMap(
    client=client,
    name="test-config",
    namespace="default",
    data={"key": "value"},
) as cm:
    cm.create()
    print(f"Created (fake): {cm.name}")
```

> **Note:** See [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html) for full details on the fake client.

### Schema Validation Before Creating

Enable schema validation to catch configuration errors before sending requests to the cluster:

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

pod = Pod(
    client=client,
    name="validated-pod",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
    schema_validation_enabled=True,  # Validates on create
)
pod.deploy()
```

You can also validate manually without creating:

```python
pod = Pod(
    name="check-me",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
)
pod.validate()  # Raises ValidationError if invalid
```

> **Note:** See [Validating Resources Against OpenAPI Schemas](validating-resources.html) for more validation options.

### Waiting for Resource Conditions

After creating a resource, you can wait for it to reach a specific status or condition:

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

with Namespace(client=client, name="wait-demo") as ns:
    ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
```

> **Note:** See [Waiting for Resource Conditions and Status](waiting-for-conditions.html) for advanced waiting patterns.

## Troubleshooting

### "KUBECONFIG not found" or connection errors

Make sure your kubeconfig is in the expected location or explicitly set the path:

```bash
export KUBECONFIG=/path/to/kubeconfig
```

Or pass it directly:

```python
client = get_client(config_file="/path/to/kubeconfig")
```

### Resource not found after creation

Some resources take time to become available. Use `wait_for_status()` or `wait()` to confirm the resource is ready before interacting with it:

```python
ns.deploy()
ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
```

### Missing `name` or `namespace` errors

- Cluster-scoped resources (like `Namespace`) require only `name`.
- Namespaced resources (like `Pod`, `ConfigMap`, `Deployment`) require both `name` and `namespace`.

```python
# Cluster-scoped — only name needed
Namespace(client=client, name="my-ns")

# Namespaced — both name and namespace required
ConfigMap(client=client, name="my-cm", namespace="my-ns")
```

## Related Pages

- [Connecting to Clusters](connecting-to-clusters.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Querying and Watching Resources](querying-resources.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html)

---

Source: connecting-to-clusters.md

Let me start by exploring the repository structure and the knowledge graph.

Now let me explore the client configuration and authentication code:

Now let me look at how the `client` parameter is used in the Resource class:

Now let me look at how resources accept the client parameter:

Now let me look at the NamespacedResource and how it handles client:

Now let me look at the tests for the client configuration:

Let me also check how environment variables relate to client configuration:

Now let me check how `get_client` is used with `generate_kubeconfig`:

Now let me look at the `conftest.py` for the tests to understand how fake_client is set up:

Let me also check the test file for more patterns of client usage:

Now let me check what environment variables are documented:

Now let me check the `generate_kubeconfig` related tests more carefully and how `DynamicClientWithKubeconfig` is used:

Now let me also look at how `get_client` is imported/used in other contexts to understand its full interface:

Let me look at some example files to see real usage patterns:

Now let me check the `get` classmethod for how client is used:

Now let me check for the `client_configuration` parameter and how `verify_ssl` is used:

Now I have all the information I need. Let me write the documentation page.

# Connecting to Clusters

Establish authenticated connections to Kubernetes and OpenShift clusters so you can create, query, and manage resources with openshift-python-wrapper.

## Prerequisites

- openshift-python-wrapper installed (see [Installing and Creating Your First Resource](quickstart.html))
- Access to a Kubernetes or OpenShift cluster (kubeconfig file, token, or in-cluster credentials)

## Quick Example

```python
from ocp_resources.resource import get_client
from ocp_resources.namespace import Namespace

# Connect using default kubeconfig (~/.kube/config or $KUBECONFIG)
client = get_client()

# Use the client with any resource
for ns in Namespace.get(client=client):
    print(ns.name)
```

## Creating a Client

All cluster connections go through `get_client()`, which returns a Kubernetes `DynamicClient`. Pass this client to every resource you create or query.

### Using the Default Kubeconfig

If you have a kubeconfig at `~/.kube/config` or the `KUBECONFIG` environment variable is set, no arguments are needed:

```python
from ocp_resources.resource import get_client

client = get_client()
```

The lookup order is:

1. The `config_file` parameter (if provided)
2. The `KUBECONFIG` environment variable
3. `~/.kube/config`

### Using a Specific Kubeconfig File

```python
client = get_client(config_file="/path/to/my/kubeconfig")
```

### Selecting a Context

If your kubeconfig has multiple contexts, select one by name:

```python
client = get_client(config_file="/path/to/kubeconfig", context="my-staging-cluster")
```

### Authenticating with a Bearer Token

Connect to a cluster API server directly using a host URL and token:

```python
client = get_client(
    host="https://api.my-cluster.example.com:6443",
    token="sha256~my-bearer-token",
)
```

### Authenticating with Username and Password (Basic Auth)

For OpenShift clusters that support OAuth-based basic authentication:

```python
client = get_client(
    host="https://api.my-cluster.example.com:6443",
    username="my-user",
    password="my-password",
)
```

> **Note:** Basic auth requires all three parameters: `host`, `username`, and `password`. This uses OpenShift's OAuth flow internally — it is not plain HTTP Basic Authentication.

### Using a Kubeconfig Dictionary

Pass a kubeconfig as a Python dictionary instead of a file path:

```python
config_dict = {
    "apiVersion": "v1",
    "kind": "Config",
    "clusters": [{"name": "my-cluster", "cluster": {"server": "https://api.example.com:6443"}}],
    "users": [{"name": "my-user", "user": {"token": "my-token"}}],
    "contexts": [{"name": "my-context", "context": {"cluster": "my-cluster", "user": "my-user"}}],
    "current-context": "my-context",
}

client = get_client(config_dict=config_dict)
```

## Passing the Client to Resources

Once you have a client, pass it to resource constructors and class methods:

```python
from ocp_resources.resource import get_client
from ocp_resources.namespace import Namespace
from ocp_resources.pod import Pod

client = get_client()

# Creating a resource
ns = Namespace(client=client, name="my-namespace")
ns.deploy()

# Querying resources
for pod in Pod.get(client=client, namespace="my-namespace"):
    print(pod.name)

# Using context managers
with Namespace(client=client, name="temp-namespace") as ns:
    ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
```

> **Warning:** The `client` parameter will become mandatory in the next major release. Always pass it explicitly. If omitted, the library creates a client automatically using the default kubeconfig, but this triggers a `FutureWarning`.

## Environment Variables

`get_client()` respects several environment variables automatically:

| Variable | Purpose | Default |
|---|---|---|
| `KUBECONFIG` | Path to the kubeconfig file | `~/.kube/config` |
| `HTTPS_PROXY` | HTTPS proxy for cluster connections | None |
| `HTTP_PROXY` | HTTP proxy (used if `HTTPS_PROXY` is not set) | None |

> **Tip:** `HTTPS_PROXY` takes precedence over `HTTP_PROXY` when both are set.

For other environment variables that control logging, resource reuse, and teardown behavior, see [Environment Variables and Configuration](environment-variables.html).

## Advanced Usage

### Disabling TLS Verification

For development clusters with self-signed certificates:

```python
client = get_client(
    host="https://api.dev-cluster.example.com:6443",
    token="sha256~my-token",
    verify_ssl=False,
)
```

> **Warning:** Never disable TLS verification in production environments.

### In-Cluster Configuration

When your code runs inside a pod on the cluster (for example, in a CI/CD pipeline or operator), `get_client()` automatically falls back to in-cluster configuration if the kubeconfig-based connection fails:

```python
# Inside a pod — no config_file or host needed
client = get_client()
```

The library first attempts to connect using the kubeconfig. If that fails with a connection error, it loads credentials from the pod's service account token mounted at `/var/run/secrets/kubernetes.io/serviceaccount/`.

### Generating a Kubeconfig File from a Token Connection

When you connect with `host` and `token`, you might need a kubeconfig file on disk (for example, to pass to external tools). Use `generate_kubeconfig=True`:

```python
client = get_client(
    host="https://api.my-cluster.example.com:6443",
    token="sha256~my-token",
    generate_kubeconfig=True,
)

# Access the generated kubeconfig path
print(client.kubeconfig)
```

The generated kubeconfig file:
- Is written to a temporary file with `0o600` permissions
- Is automatically cleaned up when the process exits

If you already passed a `config_file`, the `generate_kubeconfig` flag reuses that file path instead of creating a new one.

### Passing a Custom Client Configuration

For fine-grained control over TLS settings, timeouts, or proxy configuration, pass a `kubernetes.client.Configuration` object:

```python
import kubernetes
from ocp_resources.resource import get_client

config = kubernetes.client.Configuration()
config.verify_ssl = False
config.proxy = "http://my-proxy:8080"

client = get_client(
    host="https://api.my-cluster.example.com:6443",
    token="sha256~my-token",
    client_configuration=config,
)
```

### Connecting to Multiple Clusters

Create separate clients for each cluster and pass them to different resources:

```python
from ocp_resources.resource import get_client
from ocp_resources.namespace import Namespace

prod_client = get_client(config_file="/path/to/prod-kubeconfig")
staging_client = get_client(config_file="/path/to/staging-kubeconfig")

# Query namespaces on both clusters
prod_namespaces = list(Namespace.get(client=prod_client))
staging_namespaces = list(Namespace.get(client=staging_client))
```

### Using the Fake Client for Testing

For unit tests that don't need a real cluster, pass `fake=True`:

```python
client = get_client(fake=True)
```

This returns a fake client that simulates Kubernetes API operations in memory. See [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html) for full details.

## `get_client()` API Reference

```python
from ocp_resources.resource import get_client

get_client(
    config_file: str | None = None,
    config_dict: dict | None = None,
    context: str | None = None,
    client_configuration: kubernetes.client.Configuration | None = None,
    persist_config: bool = True,
    temp_file_path: str | None = None,
    try_refresh_token: bool = True,
    username: str | None = None,
    password: str | None = None,
    host: str | None = None,
    verify_ssl: bool | None = None,
    token: str | None = None,
    fake: bool = False,
    generate_kubeconfig: bool = False,
) -> DynamicClient
```

| Parameter | Type | Description |
|---|---|---|
| `config_file` | `str \| None` | Path to a kubeconfig file |
| `config_dict` | `dict \| None` | Kubeconfig as a Python dictionary |
| `context` | `str \| None` | Name of the kubeconfig context to use |
| `client_configuration` | `Configuration \| None` | Custom `kubernetes.client.Configuration` object |
| `persist_config` | `bool` | Whether to persist config changes back to the kubeconfig file |
| `temp_file_path` | `str \| None` | Path for temporary kubeconfig file (used with `config_dict`) |
| `try_refresh_token` | `bool` | Attempt to refresh the authentication token (for in-cluster config) |
| `username` | `str \| None` | Username for basic authentication |
| `password` | `str \| None` | Password for basic authentication |
| `host` | `str \| None` | Cluster API server URL (e.g., `https://api.example.com:6443`) |
| `verify_ssl` | `bool \| None` | Set to `False` to skip TLS certificate verification |
| `token` | `str \| None` | Bearer token for authentication |
| `fake` | `bool` | Return a fake in-memory client for testing |
| `generate_kubeconfig` | `bool` | Save connection info to a temporary kubeconfig file accessible via `client.kubeconfig` |

**Returns:** `DynamicClient` — a Kubernetes dynamic client ready for resource operations.

**Raises:**
- `kubernetes.config.ConfigException` — if no valid kubeconfig is found and in-cluster config is unavailable
- `urllib3.exceptions.MaxRetryError` — if the cluster is unreachable (before falling back to in-cluster config)
- `ClientWithBasicAuthError` — if basic auth (username/password) fails

## Authentication Method Priority

When multiple authentication parameters are provided, `get_client()` uses this priority order:

1. **Basic auth** — if `username`, `password`, and `host` are all provided
2. **Token auth** — if `host` and `token` are provided
3. **Config dict** — if `config_dict` is provided
4. **Kubeconfig file** — uses `config_file`, `KUBECONFIG` env var, or `~/.kube/config`
5. **In-cluster config** — automatic fallback if the kubeconfig connection fails

## Troubleshooting

**"FutureWarning: 'client' arg will be mandatory"**
You are creating a resource without passing a `client`. Always call `get_client()` first and pass the result:

```python
client = get_client()
ns = Namespace(client=client, name="my-ns")  # Pass client explicitly
```

**Connection refused or timeout errors**
Verify your cluster is reachable and the kubeconfig or token is correct. Check that `HTTPS_PROXY` / `HTTP_PROXY` environment variables are set correctly if you are behind a proxy.

**"KUBECONFIG environment variable set in code is not detected"**
The Kubernetes Python client reads `KUBECONFIG` at module import time. If you set it dynamically in your code, pass the path directly instead:

```python
import os
# Don't rely on: os.environ["KUBECONFIG"] = "/path/to/config"
# Instead, pass it explicitly:
client = get_client(config_file=os.environ.get("KUBECONFIG", "/path/to/config"))
```

**Basic auth fails with `ClientWithBasicAuthError`**
Basic auth uses the OpenShift OAuth flow. Ensure your cluster has a well-known OAuth endpoint at `/.well-known/oauth-authorization-server`. This method does not work with vanilla Kubernetes clusters.

## Related Pages

- [Installing and Creating Your First Resource](quickstart.html)
- [Environment Variables and Configuration](environment-variables.html)
- [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Error Handling and Exception Reference](error-handling.html)

---

Source: creating-and-managing-resources.md

Let me start by exploring the repository structure and the knowledge graph.

Now let me look at the YAML-related functions and some example resource files:

Now let me look at some concrete resource examples and tests:

Now let me look at tests for create, update, delete operations and the exceptions:

Now let me look at the exceptions module and the project resource:

Now let me look at how `to_yaml`, `yaml_file`, and `kind_dict` usage works, and the `dry_run` option:

Now let me check how `ensure_exists` and `wait_for_resource` work:

Now let me look at how the `service.py` and `route.py` look to understand namespaced resources better:

Let me also look at the `node_network_configuration_policy.py` for a more complex `deploy/clean_up` example:

Now let me look at the `to_yaml` method and some environment variable usage:

Let me also check the `dry_run` behavior and check for how `label` and `annotations` work:

Now I have a thorough understanding of the codebase. Let me write the documentation.

# Creating and Managing Resources

Deploy Kubernetes and OpenShift resources to your cluster, update them in place, and clean them up reliably using the openshift-python-wrapper's resource lifecycle methods.

## Prerequisites

- `openshift-python-wrapper` installed (see [Installing and Creating Your First Resource](quickstart.html))
- A connected client — either a real cluster connection or a fake client for testing (see [Connecting to Clusters](connecting-to-clusters.html))

## Quick Example

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

# Create, use, and automatically delete a Namespace
with Namespace(client=client, name="my-app") as ns:
    print(f"{ns.name} is ready")
# Namespace is deleted when the block exits
```

## Creating Resources

Every resource provides three ways to create it: calling `deploy()` directly, using a context manager, or loading from a YAML file.

### Using `deploy()` and `clean_up()`

```python
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="my-namespace")
ns.deploy()

# ... do work ...

ns.clean_up()  # Deletes the resource and waits for deletion
```

`deploy()` returns the resource instance, so you can chain:

```python
ns = Namespace(client=client, name="my-namespace").deploy()
```

Pass `wait=True` to `deploy()` to block until the resource reaches its ready state:

```python
ns = Namespace(client=client, name="my-namespace")
ns.deploy(wait=True)
```

### Using a Context Manager

The context manager calls `deploy()` on entry and `clean_up()` on exit, guaranteeing cleanup even if an exception occurs:

```python
from ocp_resources.config_map import ConfigMap

with ConfigMap(
    client=client,
    name="app-config",
    namespace="my-namespace",
    data={"database_url": "postgres://db:5432/myapp"},
) as cm:
    print(cm.instance.data)
# ConfigMap is automatically deleted here
```

> **Note:** If cleanup fails during context manager exit, a `ResourceTeardownError` is raised so failures are never silently ignored.

### Namespaced vs. Cluster-Scoped Resources

Resources that live within a namespace inherit from `NamespacedResource` and require both `name` and `namespace`:

```python
from ocp_resources.pod import Pod

pod = Pod(
    client=client,
    name="nginx",
    namespace="my-namespace",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
)
pod.deploy()
```

Cluster-scoped resources (like `Namespace`) inherit from `Resource` and need only `name`:

```python
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="my-namespace")
ns.deploy()
```

### Creating Resources from a YAML File

Pass `yaml_file` to create any resource directly from a YAML manifest. The resource name and namespace are read from the file:

```python
from ocp_resources.config_map import ConfigMap

cm = ConfigMap(client=client, yaml_file="manifests/configmap.yaml")
cm.deploy()
```

`yaml_file` also accepts a `StringIO` object for in-memory YAML:

```python
from io import StringIO
from ocp_resources.config_map import ConfigMap

yaml_content = StringIO("""
apiVersion: v1
kind: ConfigMap
metadata:
  name: dynamic-config
  namespace: my-namespace
data:
  setting: "enabled"
""")

cm = ConfigMap(client=client, yaml_file=yaml_content)
cm.deploy()
```

### Creating Resources from a Dictionary

Use `kind_dict` to pass a pre-built dictionary. When `kind_dict` is provided, the resource class's `to_dict()` logic is bypassed entirely:

```python
from ocp_resources.config_map import ConfigMap

resource_dict = {
    "apiVersion": "v1",
    "kind": "ConfigMap",
    "metadata": {
        "name": "from-dict",
        "namespace": "my-namespace",
    },
    "data": {"key": "value"},
}

cm = ConfigMap(client=client, kind_dict=resource_dict)
cm.deploy()
```

> **Warning:** `yaml_file` and `kind_dict` are mutually exclusive. Passing both raises a `ValueError`.

### Adding Labels and Annotations

Pass `label` and `annotations` at construction time:

```python
from ocp_resources.namespace import Namespace

ns = Namespace(
    client=client,
    name="labeled-ns",
    label={"team": "platform", "env": "staging"},
    annotations={"description": "Staging environment namespace"},
)
ns.deploy()
```

## Updating Resources

### Patching with `update()`

`update()` sends a merge patch — it adds or modifies fields without removing existing ones:

```python
cm = ConfigMap(
    client=client,
    name="app-config",
    namespace="my-namespace",
    ensure_exists=True,
)

cm.update(resource_dict={
    "data": {"new_key": "new_value"},
})
```

### Replacing with `update_replace()`

`update_replace()` performs a full replacement of the resource. Use this when you need to **remove** fields that `update()` would leave untouched:

```python
# Get the current state
current = cm.instance.to_dict()

# Remove a key
current["data"].pop("old_key", None)

# Replace the entire resource
cm.update_replace(resource_dict=current)
```

| Method | HTTP Verb | Behavior | Use When |
|--------|-----------|----------|----------|
| `update()` | PATCH | Merges fields into existing resource | Adding or changing fields |
| `update_replace()` | PUT | Replaces entire resource | Removing fields or doing full replacements |

## Deleting Resources

### Using `clean_up()`

```python
ns.clean_up()  # Deletes and waits for removal (default: wait=True)
```

Control the deletion behavior:

```python
# Delete without waiting
ns.clean_up(wait=False)

# Delete with a custom timeout (in seconds)
ns.clean_up(timeout=120)
```

`clean_up()` returns `True` if the resource was successfully deleted, `False` otherwise.

### Using `delete()`

`delete()` is the lower-level method called by `clean_up()`:

```python
ns.delete(wait=True, timeout=240)
```

You can also pass a custom body for delete options:

```python
ns.delete(body={"propagationPolicy": "Background"})
```

> **Tip:** If you delete a resource that doesn't exist, `delete()` logs a warning and returns `True` instead of raising an error.

### Controlling Teardown

Set `teardown=False` at construction to prevent automatic deletion in context managers:

```python
with Namespace(client=client, name="persistent-ns", teardown=False) as ns:
    print("This namespace will NOT be deleted on exit")
```

## Checking Resource State

### Check If a Resource Exists

```python
ns = Namespace(client=client, name="my-namespace")
if ns.exists:
    print("Namespace exists")
```

### Ensure a Resource Already Exists

Use `ensure_exists=True` to raise `ResourceNotFoundError` if the resource is not present on the cluster:

```python
from ocp_resources.config_map import ConfigMap

# Raises ResourceNotFoundError if the ConfigMap doesn't exist
cm = ConfigMap(
    client=client,
    name="must-exist",
    namespace="default",
    ensure_exists=True,
)
```

### Get the Live Instance

The `instance` property fetches the current state from the API server:

```python
ns = Namespace(client=client, name="my-namespace", ensure_exists=True)
print(ns.instance.metadata.labels)
print(ns.status)
```

### Export as YAML

```python
ns = Namespace(client=client, name="my-namespace")
print(ns.to_yaml())
```

For more on querying, watching, and status checks, see [Querying and Watching Resources](querying-resources.html).

## Dry Run Mode

Validate a resource against the API server without actually creating it:

```python
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="dry-run-ns", dry_run=True)
ns.deploy()  # Sends the request with ?dryRun=All — no resource is created
```

## Advanced Usage

### Constructor Parameters Reference

All resource classes accept these common parameters:

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `client` | `DynamicClient` | `None` | Cluster client connection (will be required in next major release) |
| `name` | `str` | `None` | Resource name |
| `namespace` | `str` | `None` | Namespace (namespaced resources only) |
| `teardown` | `bool` | `True` | Whether `clean_up()` deletes the resource |
| `yaml_file` | `str` | `None` | Path to a YAML manifest file |
| `kind_dict` | `dict` | `None` | Dictionary representation of the resource |
| `delete_timeout` | `int` | `240` | Timeout in seconds for delete operations |
| `dry_run` | `bool` | `False` | Server-side dry run without persisting |
| `label` | `dict` | `None` | Labels to set on the resource |
| `annotations` | `dict` | `None` | Annotations to set on the resource |
| `ensure_exists` | `bool` | `False` | Raise if resource doesn't already exist |
| `wait_for_resource` | `bool` | `False` | Wait for resource creation in context manager |
| `schema_validation_enabled` | `bool` | `False` | Validate against OpenAPI schema on create/replace |
| `hash_log_data` | `bool` | `True` | Mask sensitive data in logs |

For the full API reference, see [Resource and NamespacedResource API](resource-api.html).

### Schema Validation on Create and Replace

Enable automatic schema validation so `create()` and `update_replace()` check your resource against its OpenAPI schema before sending it to the API server:

```python
from ocp_resources.config_map import ConfigMap

cm = ConfigMap(
    client=client,
    name="validated-cm",
    namespace="default",
    data={"key": "value"},
    schema_validation_enabled=True,  # Validates before create()
)
cm.deploy()
```

You can also validate on demand or validate a dictionary directly:

```python
# Validate an existing resource instance
cm.validate()

# Validate a dictionary without creating a resource
ConfigMap.validate_dict({
    "apiVersion": "v1",
    "kind": "ConfigMap",
    "metadata": {"name": "test"},
    "data": {"key": "value"},
})
```

Both methods raise `ValidationError` if validation fails. For more details, see [Validating Resources Against OpenAPI Schemas](validating-resources.html).

### Temporary Edits with ResourceEditor

`ResourceEditor` applies patches to existing resources and automatically restores original values when used as a context manager:

```python
from ocp_resources.resource import ResourceEditor

ns = Namespace(client=client, name="my-namespace", ensure_exists=True)

with ResourceEditor(
    patches={ns: {"metadata": {"labels": {"temporary-label": "true"}}}}
):
    # Label is applied
    assert ns.labels["temporary-label"] == "true"
# Label is automatically removed
```

`ResourceEditor` supports both `update` (merge patch) and `replace` actions:

```python
with ResourceEditor(
    patches={ns: {"metadata": {"labels": {"keep-only-this": "true"}}}},
    action="replace",
):
    pass  # Full replacement applied, then restored on exit
```

For complete coverage, see [Editing Resources Temporarily with ResourceEditor](editing-resources-temporarily.html).

### Managing Multiple Resources

#### ResourceList — Create N Copies

Create multiple instances of the same resource type with indexed names:

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import ResourceList

with ResourceList(
    client=client,
    resource_class=Namespace,
    name="test-ns",
    num_resources=3,
) as namespaces:
    # Creates test-ns-1, test-ns-2, test-ns-3
    for ns in namespaces:
        print(ns.name)
# All three namespaces are deleted on exit
```

#### NamespacedResourceList — One Per Namespace

Create one resource in each namespace from a `ResourceList`:

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import NamespacedResourceList, ResourceList

namespaces = ResourceList(
    client=client,
    resource_class=Namespace,
    name="env",
    num_resources=3,
)

with NamespacedResourceList(
    client=client,
    resource_class=Pod,
    namespaces=namespaces,
    name="worker",
    containers=[{"name": "app", "image": "myapp:latest"}],
) as pods:
    for pod in pods:
        print(f"{pod.name} in {pod.namespace}")
```

For detailed patterns, see [Managing Bulk Resources with ResourceList](managing-resource-lists.html).

### Debugging with Environment Variables

These environment variables control resource lifecycle behavior at runtime without code changes:

| Variable | Effect |
|----------|--------|
| `REUSE_IF_RESOURCE_EXISTS` | Skip `deploy()` if the resource already exists |
| `SKIP_RESOURCE_TEARDOWN` | Skip `clean_up()` to keep resources for debugging |

Both accept a YAML-formatted dictionary. Spaces are significant:

```bash
# Skip creation of all Pods
export REUSE_IF_RESOURCE_EXISTS="{Pod: {}}"

# Skip creation of a specific Pod in a specific namespace
export REUSE_IF_RESOURCE_EXISTS="{Pod: {my-pod: my-namespace}}"

# Skip teardown for a Namespace and a Pod
export SKIP_RESOURCE_TEARDOWN="{Namespace: {my-ns:}, Pod: {my-pod: my-namespace}}"
```

For the full list of environment variables, see [Environment Variables and Configuration](environment-variables.html).

## Troubleshooting

**`MissingRequiredArgumentError` on construction**
You must provide either `name`, `yaml_file`, or `kind_dict`. For namespaced resources, both `name` and `namespace` are required unless using `yaml_file` or `kind_dict`.

**`ResourceTeardownError` in context manager**
The resource's `clean_up()` returned `False`, meaning deletion failed. Check cluster permissions and resource finalizers.

**`ValueError: yaml_file and resource_dict are mutually exclusive`**
You passed both `yaml_file` and `kind_dict` to a constructor. Use only one.

**`ResourceNotFoundError` with `ensure_exists=True`**
The resource doesn't exist on the cluster. Verify the name (and namespace for namespaced resources) are correct.

**Resource not deleted after context manager exits**
Check if `teardown=False` was passed. Also verify the `SKIP_RESOURCE_TEARDOWN` environment variable is not set.

## Related Pages

- [Installing and Creating Your First Resource](quickstart.html)
- [Querying and Watching Resources](querying-resources.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Editing Resources Temporarily with ResourceEditor](editing-resources-temporarily.html)
- [Resource and NamespacedResource API](resource-api.html)

---

Source: querying-resources.md

Let me start by reading the knowledge graph and understanding the codebase structure.

Now let me explore the key source files related to querying, listing, watching, existence checks, and status retrieval.

Now let me look at the `_prepare_resources` method and the `get` method more closely to understand the filtering parameters:

Now let me look at the event module and test files for querying:

Now let me look at the test files for get/list/watch operations:

Now let me check the conftest to understand fake_client and look at the `Condition` attribute and `Status` attribute patterns:

Now let me check how `get` is called with selectors:

Let me look at how `get` passes `field_selector`, `label_selector`, and `namespace` through kwargs:

Now let me look at more examples of how get() is used with various kwargs, and check what other files show:

Let me check how `_build_event_field_selector` and events work:

Now let me check the namespace.py to see how get() is used in a simple resource:

Now let me look at the node.py for a cluster-scoped resource usage of get:

Now let me check the exceptions module for ConditionError:

Let me also check `ensure_exists`:

Let me also look for properties that relate to querying the resource (uuid, labels):

Now let me check the test for events and any resource-specific tests about querying:

Let me also look at the fake_client to understand what kwargs `get` passes through:

Now I have a thorough understanding of the codebase. Let me write the documentation page.

# Querying and Watching Resources

Find, filter, and monitor Kubernetes and OpenShift resources in your cluster — whether you need to list all pods by label, check if a resource exists, watch for real-time changes, or inspect status and conditions.

## Prerequisites

- `openshift-python-wrapper` installed (see [Installing and Creating Your First Resource](quickstart.html))
- A connected `DynamicClient` instance (see [Connecting to Clusters](connecting-to-clusters.html))

## Quick Example

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

# List all pods in a namespace
for pod in Pod.get(client=client, namespace="my-namespace"):
    print(f"{pod.name} — {pod.status}")
```

## Listing Resources

Every resource class provides a `get()` class method that returns a generator of resource objects. The method accepts standard Kubernetes query parameters as keyword arguments.

### List all resources of a type

```python
from ocp_resources.namespace import Namespace

for ns in Namespace.get(client=client):
    print(ns.name)
```

### Filter by namespace

Namespaced resources (like `Pod`, `Service`, `ConfigMap`) accept a `namespace` keyword:

```python
from ocp_resources.pod import Pod

for pod in Pod.get(client=client, namespace="default"):
    print(pod.name)
```

### Filter by label selector

Use `label_selector` with standard Kubernetes label selector syntax:

```python
from ocp_resources.pod import Pod

# Single label
for pod in Pod.get(client=client, namespace="my-app", label_selector="app=nginx"):
    print(pod.name)

# Multiple labels (comma-separated)
for pod in Pod.get(client=client, label_selector="app=nginx,tier=frontend"):
    print(pod.name)
```

### Filter by field selector

Use `field_selector` with standard Kubernetes field selector syntax:

```python
from ocp_resources.pod import Pod

# Find pods on a specific node
for pod in Pod.get(client=client, field_selector="spec.nodeName=worker-1"):
    print(pod.name)

# Find pods by status phase
for pod in Pod.get(client=client, namespace="default", field_selector="status.phase=Running"):
    print(pod.name)
```

### Get raw resource objects

By default, `get()` returns wrapper objects. Pass `raw=True` to get the underlying `ResourceInstance` or `ResourceField` objects directly:

```python
from ocp_resources.pod import Pod

for pod in Pod.get(client=client, namespace="default", raw=True):
    # Returns raw ResourceField objects instead of Pod instances
    print(pod.metadata.name, pod.status.phase)
```

### List all cluster resources

To iterate over every resource type in the cluster:

```python
from ocp_resources.resource import Resource

for resource in Resource.get_all_cluster_resources(client=client):
    print(f"{resource.kind}: {resource.metadata.name}")

# Filter with label selector
for resource in Resource.get_all_cluster_resources(client=client, label_selector="my-label=value"):
    print(f"{resource.kind}: {resource.metadata.name}")
```

## `get()` Method Reference

```python
@classmethod
def get(
    cls,
    client: DynamicClient | None = None,
    singular_name: str = "",
    exceptions_dict: dict[type[Exception], list[str]] = DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
    raw: bool = False,
    *args,
    **kwargs,
) -> Generator
```

| Parameter | Type | Description |
|---|---|---|
| `client` | `DynamicClient` | Kubernetes dynamic client (required) |
| `singular_name` | `str` | Resource kind in lowercase; used when multiple resources match the same kind |
| `raw` | `bool` | If `True`, return raw `ResourceInstance`/`ResourceField` objects instead of wrapper instances |
| `exceptions_dict` | `dict` | Exceptions to retry on during API calls |
| `**kwargs` | | Passed through to the Kubernetes API: `namespace`, `label_selector`, `field_selector`, `limit`, `resource_version`, `timeout_seconds`, etc. |

**Returns:** A generator yielding resource objects.

## Checking Resource Existence

The `exists` property queries the API server and returns the resource instance if it exists, or `None` if it does not.

```python
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="my-namespace")

if ns.exists:
    print("Namespace exists")
else:
    print("Namespace not found")
```

### Require existence at initialization

Use `ensure_exists=True` to raise `ResourceNotFoundError` immediately if the resource is not found:

```python
from ocp_resources.pod import Pod

# Raises ResourceNotFoundError if the pod doesn't exist
pod = Pod(client=client, name="my-pod", namespace="default", ensure_exists=True)
```

> **Tip:** Use `ensure_exists=True` when you need a reference to a resource that must already be on the cluster, such as a pre-existing ConfigMap or Secret.

## Getting the Resource Instance

The `instance` property fetches the full live resource from the API server:

```python
pod = Pod(client=client, name="my-pod", namespace="default")

# Get the full resource instance
inst = pod.instance
print(inst.metadata.name)
print(inst.metadata.uid)
print(inst.metadata.resourceVersion)

# Convert to a plain dictionary
resource_dict = pod.instance.to_dict()
```

> **Note:** Each access to `.instance` makes an API call. Cache the result in a local variable if you need multiple fields from the same snapshot.

## Retrieving Labels

The `labels` property returns the resource's labels:

```python
ns = Namespace(client=client, name="my-namespace")
print(ns.labels)  # {'kubernetes.io/metadata.name': 'my-namespace', ...}

# Check a specific label
if "app" in ns.labels:
    print(f"App label: {ns.labels['app']}")
```

## Retrieving Resource Status

### Status phase

The `status` property returns the current phase (e.g., `Running`, `Pending`, `Active`):

```python
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="my-namespace")
print(ns.status)  # "Active"
```

Each resource class defines status constants you can compare against:

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-pod", namespace="default")
if pod.status == Pod.Status.RUNNING:
    print("Pod is running")
```

Common status constants available on all resources:

| Constant | Value |
|---|---|
| `Status.RUNNING` | `"Running"` |
| `Status.PENDING` | `"Pending"` |
| `Status.SUCCEEDED` | `"Succeeded"` |
| `Status.FAILED` | `"Failed"` |
| `Status.ACTIVE` | `"Active"` |
| `Status.TERMINATING` | `"Terminating"` |
| `Status.COMPLETED` | `"Completed"` |
| `Status.ERROR` | `"Error"` |

### Condition messages

Use `get_condition_message()` to retrieve the message for a specific condition type:

```python
pod = Pod(client=client, name="my-pod", namespace="default")

# Get message for a condition type
msg = pod.get_condition_message(condition_type="Ready")
print(msg)

# Optionally filter by condition status
msg = pod.get_condition_message(
    condition_type="Ready",
    condition_status="False",
)
print(msg)  # Returns empty string if status doesn't match
```

**Signature:**
```python
def get_condition_message(
    self,
    condition_type: str,
    condition_status: str = "",
) -> str
```

| Parameter | Type | Description |
|---|---|---|
| `condition_type` | `str` | The condition type to look up (e.g., `"Ready"`, `"Available"`) |
| `condition_status` | `str` | If provided, only return the message when the condition has this status |

**Returns:** The condition message string, or `""` if not found or status doesn't match.

### Condition constants

Resources provide built-in condition constants:

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-pod", namespace="default")

# Use condition constants
msg = pod.get_condition_message(
    condition_type=Pod.Condition.READY,
    condition_status=Pod.Condition.Status.TRUE,
)
```

Common condition constants:

| Constant | Value |
|---|---|
| `Condition.READY` | `"Ready"` |
| `Condition.AVAILABLE` | `"Available"` |
| `Condition.PROGRESSING` | `"Progressing"` |
| `Condition.DEGRADED` | `"Degraded"` |
| `Condition.UPGRADEABLE` | `"Upgradeable"` |
| `Condition.Status.TRUE` | `"True"` |
| `Condition.Status.FALSE` | `"False"` |
| `Condition.Status.UNKNOWN` | `"Unknown"` |

## Watching for Changes

### Watch a specific resource

The `watcher()` method streams events for a specific resource instance:

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-pod", namespace="default")

for event in pod.watcher(timeout=60):
    print(f"Event type: {event['type']}")       # ADDED, MODIFIED, DELETED
    print(f"Object: {event['object'].metadata.name}")
    print(f"Raw: {event['raw_object']}")
```

**Signature:**
```python
def watcher(
    self,
    timeout: int,
    resource_version: str = "",
) -> Generator[dict[str, Any], None, None]
```

| Parameter | Type | Description |
|---|---|---|
| `timeout` | `int` | How long to watch (in seconds) |
| `resource_version` | `str` | Only receive events newer than this version. Defaults to the resource version at creation time. |

Each yielded event is a dict with these keys:

| Key | Description |
|---|---|
| `type` | Event type: `"ADDED"`, `"MODIFIED"`, or `"DELETED"` |
| `object` | A `ResourceInstance` wrapping the resource |
| `raw_object` | A plain dict representing the resource |

## Querying Events

### Stream events for a resource

The `events()` method watches Kubernetes events associated with a resource:

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-pod", namespace="default")

for event in pod.events(timeout=30):
    print(f"Reason: {event.object.reason}")
    print(f"Message: {event.object.message}")
```

**Signature:**
```python
def events(
    self,
    name: str = "",
    label_selector: str = "",
    field_selector: str = "",
    resource_version: str = "",
    timeout: int = 240,
) -> Generator
```

| Parameter | Type | Description |
|---|---|---|
| `name` | `str` | Filter by event name |
| `label_selector` | `str` | Filter by labels (comma-separated `key=value`) |
| `field_selector` | `str` | Additional field filters (merged with auto-generated `involvedObject.name` filter) |
| `resource_version` | `str` | Only return events newer than this version |
| `timeout` | `int` | Watch timeout in seconds (default: 240) |

```python
# Example: filter for Warning events with a specific reason
for event in pod.events(
    field_selector="type==Warning,reason==BackOff",
    timeout=10,
):
    print(event.object.message)
```

### List existing events (non-streaming)

Use `Event.list()` to get a snapshot of existing events without streaming:

```python
from ocp_resources.event import Event

# List Warning events from the last 5 minutes
events = Event.list(
    client=client,
    namespace="my-namespace",
    field_selector="type==Warning",
)
for event in events:
    print(f"{event.reason}: {event.message}")

# List events from the last 10 minutes
events = Event.list(
    client=client,
    namespace="my-namespace",
    since_seconds=600,
)
```

**Signature:**
```python
@classmethod
def Event.list(
    cls,
    client: DynamicClient,
    namespace: str | None = None,
    field_selector: str | None = None,
    label_selector: str | None = None,
    since_seconds: int = 300,
) -> list
```

Events are returned sorted by timestamp (most recent first).

### Delete events

Clean up events before tests to avoid false positives:

```python
from ocp_resources.event import Event

Event.delete_events(
    client=client,
    namespace="my-namespace",
    field_selector="reason=AnEventReason",
)
```

## Advanced Usage

### Handling resources with duplicate API groups

Some resource kinds exist in multiple API groups. Use `singular_name` to disambiguate:

```python
from ocp_resources.dns import DNS

# When a kind exists in multiple API groups, pass singular_name
for dns in DNS.get(client=client, singular_name="dns"):
    print(dns.name)
```

### Using `full_api()` for low-level access

The `full_api()` method returns the raw Kubernetes API resource object, giving you direct access to the API:

```python
ns = Namespace(client=client, name="my-namespace")
api = ns.full_api()

# Use the API directly for custom queries
result = api.get(
    label_selector="app=nginx",
    field_selector="status.phase=Active",
    limit=10,
)
```

`full_api()` accepts these keyword arguments:

| Parameter | Description |
|---|---|
| `pretty` | Pretty-print output |
| `_continue` | Continuation token for paginated results |
| `field_selector` | Filter by fields |
| `label_selector` | Filter by labels |
| `limit` | Maximum number of results |
| `resource_version` | Filter by resource version |
| `timeout_seconds` | Server-side timeout |
| `watch` | Enable watch mode |

### Converting resources to YAML

```python
ns = Namespace(client=client, name="my-namespace")
print(ns.to_yaml())
```

### Wait for conditions and status

For polling-based waiting (e.g., wait until a pod is `Running` or a condition is `True`), see [Waiting for Resource Conditions and Status](waiting-for-conditions.html).

### Querying during tests with the fake client

You can use all querying methods with the fake client for unit testing:

```python
from ocp_resources.resource import get_client
from ocp_resources.namespace import Namespace

client = get_client(fake=True)
ns = Namespace(client=client, name="test-ns")
ns.deploy()

# All query methods work the same way
assert ns.exists
assert ns.instance
assert ns.status == Namespace.Status.ACTIVE

# Listing works too
for ns in Namespace.get(client=client):
    print(ns.name)
```

See [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html) for details.

## Troubleshooting

**`ResourceNotFoundError` when using `ensure_exists=True`**
The resource does not exist on the cluster. Verify the name, namespace, and that the resource has been created before initializing the wrapper.

**`NotImplementedError: Couldn't find <Kind> in <api_group>`**
The resource kind is not available on your cluster. This typically means the CRD has not been installed or the API group is incorrect.

**`get()` returns no results**
- Verify the `namespace` is correct (omit it for cluster-scoped resources).
- Check your `label_selector` and `field_selector` syntax — use `==` for equality (e.g., `field_selector="status.phase==Running"`).
- Ensure the client has permissions to list the resource type.

**`status` property raises `AttributeError`**
Not all resources have a `status.phase` field. Access `instance.status` directly to inspect what status fields are available for your resource type.

## Related Pages

- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Working with Kubernetes Events](working-with-events.html)
- [Common Resource Patterns](common-patterns.html)
- [Resource and NamespacedResource API](resource-api.html)

---

Source: waiting-for-conditions.md

Let me start by reading the knowledge graph and page manifest, then explore the relevant source code.

Now let me examine the resource-specific wait methods, the `wait_for_condition` more closely, and the constants used:

Now let me look at the `Status` class, `Condition` class, and how `create` integrates with `wait`:

Now let me look at how Status/Condition are exposed on the base Resource class and how `deploy` / `create` uses wait:

Now let me see the `deploy` method more fully and how `wait` is passed through:

Now let me check the `sriov_network_node_state.py` and `virtual_machine_restore.py` for more wait patterns, and also look at the `TimeoutExpiredError` import:

Now let me check the `cdi_config.py` wait method and look at the `wait_for_resource` parameter in the constructor:

Now let me look at how `create` uses `wait`:

Now let me look at more of the `status` property:

Now I have all the information I need. Let me also check the `TimeoutSampler` import to understand the exception:

Let me check if there are any virtual machine import wait patterns with more stop_status details:

Now let me look at the `node_network_state.py` wait patterns:

I now have comprehensive understanding of all the wait methods. Let me compile the documentation.

# Waiting for Resource Conditions and Status

Wait for Kubernetes and OpenShift resources to reach a desired status, condition, or lifecycle state using built-in polling methods with configurable timeouts and early exit on failure.

## Prerequisites

- `openshift-python-wrapper` installed in your project
- A connected client (see [Connecting to Clusters](connecting-to-clusters.html))
- A resource instance to wait on (see [Creating and Managing Resources](creating-and-managing-resources.html))

## Quick Example

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-pod", namespace="default")
pod.create(wait=True)  # Creates the pod and waits for it to exist

# Wait for the pod to reach "Running" status
pod.wait_for_status(status=Pod.Status.RUNNING)
```

## Waiting for a Resource to Exist

The `wait()` method polls until a resource exists on the cluster. This is useful after creating a resource or when waiting for an external controller to produce one.

```python
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="my-namespace")
ns.create()
ns.wait(timeout=120, sleep=2)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `timeout` | `int` | `240` (4 minutes) | Maximum seconds to wait |
| `sleep` | `int` | `1` | Seconds between polls |

**Raises:** `TimeoutExpiredError` if the resource does not appear before the timeout.

> **Tip:** Pass `wait=True` to `create()` or `deploy()` to combine creation and existence waiting in one call:
> ```python
> pod.create(wait=True)
> # or
> pod.deploy(wait=True)
> ```

## Waiting for Deletion

The `wait_deleted()` method polls until a resource no longer exists:

```python
pod.delete(wait=True, timeout=120)

# Or call wait_deleted() separately:
pod.delete()
success = pod.wait_deleted(timeout=120)
if not success:
    print("Resource still exists after timeout")
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `timeout` | `int` | `240` (4 minutes) | Maximum seconds to wait |

**Returns:** `True` if the resource was deleted, `False` if the timeout was reached.

> **Note:** Unlike most wait methods, `wait_deleted()` returns `False` on timeout rather than raising an exception.

## Waiting for Status (Phase)

The `wait_for_status()` method polls `status.phase` until it matches the expected value:

```python
from ocp_resources.datavolume import DataVolume

dv = DataVolume(client=client, name="my-dv", namespace="default")
dv.wait_for_status(status=DataVolume.Status.SUCCEEDED, timeout=600)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `status` | `str` | *(required)* | Expected `status.phase` value |
| `timeout` | `int` | `240` (4 minutes) | Maximum seconds to wait |
| `stop_status` | `str \| None` | `Status.FAILED` | Status that aborts the wait immediately |
| `sleep` | `int` | `1` | Seconds between polls |
| `exceptions_dict` | `dict` | Protocol + cluster retry errors | Exceptions to catch and retry |

**Raises:** `TimeoutExpiredError` if the resource does not reach the desired status, or if `stop_status` is encountered.

### Using `stop_status` for early failure

By default, `wait_for_status()` aborts immediately if the resource enters a `"Failed"` phase. Override this to detect a different failure phase:

```python
from ocp_resources.virtual_machine_instance import VirtualMachineInstance

vmi = VirtualMachineInstance(client=client, name="my-vmi", namespace="default")
vmi.wait_for_status(
    status=VirtualMachineInstance.Status.RUNNING,
    stop_status="ErrorUnschedulable",
    timeout=300,
)
```

### Built-in status constants

Every resource inherits status constants from its base class. Resource-specific subclasses extend these with additional values:

```python
# Base status constants (available on all resources)
from ocp_resources.resource import Resource

Resource.Status.SUCCEEDED    # "Succeeded"
Resource.Status.FAILED       # "Failed"
Resource.Status.RUNNING      # "Running"
Resource.Status.PENDING      # "Pending"

# Resource-specific statuses
from ocp_resources.virtual_machine import VirtualMachine

VirtualMachine.Status.MIGRATING     # "Migrating"
VirtualMachine.Status.STOPPED       # "Stopped"
VirtualMachine.Status.STARTING      # "Starting"
VirtualMachine.Status.PAUSED        # "Paused"

from ocp_resources.persistent_volume_claim import PersistentVolumeClaim

PersistentVolumeClaim.Status.BOUND  # "Bound"
```

## Waiting for Conditions

The `wait_for_condition()` method waits for a specific entry in `status.conditions[]` to match a desired type, status, and optionally a reason or message:

```python
from ocp_resources.user_defined_network import UserDefinedNetwork

udn = UserDefinedNetwork(client=client, name="my-net", namespace="default")
udn.wait_for_condition(
    condition=UserDefinedNetwork.Condition.NETWORK_READY,
    status=UserDefinedNetwork.Condition.Status.TRUE,
    timeout=30,
)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `condition` | `str` | *(required)* | Condition type to match (e.g. `"Ready"`, `"Available"`) |
| `status` | `str` | *(required)* | Expected condition status (`"True"`, `"False"`, `"Unknown"`) |
| `timeout` | `int` | `300` (5 minutes) | Maximum seconds to wait |
| `sleep_time` | `int` | `1` | Seconds between polls |
| `reason` | `str \| None` | `None` | If set, the condition's `reason` field must also match |
| `message` | `str` | `""` | Text that must appear in the condition's `message` field |
| `stop_condition` | `str \| None` | `None` | A condition type that aborts the wait immediately |
| `stop_status` | `str` | `"True"` | The status of the stop condition that triggers early abort |

**Raises:**
- `TimeoutExpiredError` — if the condition is not met before the timeout
- `ConditionError` — if the `stop_condition` is detected before the desired condition

### Matching by reason and message

```python
pod.wait_for_condition(
    condition="Ready",
    status="True",
    reason="PodCompleted",
    message="containers with",
    timeout=120,
)
```

The `reason` must be an exact match. The `message` is checked using substring inclusion — the condition's message field must *contain* the provided string.

### Using `stop_condition` for early abort

Stop conditions let you fail fast when an unrecoverable condition appears:

```python
from ocp_resources.exceptions import ConditionError

try:
    resource.wait_for_condition(
        condition="Ready",
        status="True",
        stop_condition="Failed",
        stop_status="True",
        timeout=300,
    )
except ConditionError as e:
    print(f"Resource hit a failure condition: {e}")
```

> **Note:** The `stop_condition` matching only checks the condition `type` and `status`. The `reason` and `message` parameters are ignored when evaluating stop conditions.

### Built-in condition constants

```python
from ocp_resources.resource import Resource

# Condition types
Resource.Condition.READY         # "Ready"
Resource.Condition.AVAILABLE     # "Available"
Resource.Condition.DEGRADED      # "Degraded"
Resource.Condition.PROGRESSING   # "Progressing"
Resource.Condition.UPGRADEABLE   # "Upgradeable"
Resource.Condition.NETWORK_READY # "NetworkReady"

# Condition statuses
Resource.Condition.Status.TRUE    # "True"
Resource.Condition.Status.FALSE   # "False"
Resource.Condition.Status.UNKNOWN # "Unknown"

# Condition reasons
Resource.Condition.Reason.ALL_REQUIREMENTS_MET  # "AllRequirementsMet"
Resource.Condition.Reason.INSTALL_SUCCEEDED     # "InstallSucceeded"
```

## Waiting with Context Managers

When using a resource as a context manager, set `wait_for_resource=True` to automatically wait for the resource to exist after creation:

```python
from ocp_resources.pod import Pod

with Pod(
    client=client,
    name="my-pod",
    namespace="default",
    wait_for_resource=True,
) as pod:
    # Pod exists when you reach this line
    pod.wait_for_status(status=Pod.Status.RUNNING)
```

See [Creating and Managing Resources](creating-and-managing-resources.html) for more on context manager usage.

## Advanced Usage

### Waiting for any conditions to appear

The `wait_for_conditions()` method waits up to 30 seconds for the resource's `status.conditions` list to be populated (non-empty). This is useful when you need to inspect conditions but aren't sure they've been set yet:

```python
resource.wait_for_conditions()
# Now resource.instance.status.conditions is available
```

### Resource-specific wait methods

Many resource types provide specialized wait methods beyond the base `wait_for_status()` and `wait_for_condition()`. These methods encapsulate domain-specific logic.

#### Deployments

```python
from ocp_resources.deployment import Deployment

deploy = Deployment(client=client, name="my-app", namespace="default")
deploy.wait_for_replicas(deployed=True, timeout=240)   # All replicas ready
deploy.wait_for_replicas(deployed=False, timeout=120)   # Scale to zero complete
```

#### DaemonSets

```python
from ocp_resources.daemonset import DaemonSet

ds = DaemonSet(client=client, name="my-ds", namespace="default")
ds.wait_until_deployed(timeout=240)  # All desired pods are ready
```

#### VirtualMachines (KubeVirt)

```python
from ocp_resources.virtual_machine import VirtualMachine
from ocp_resources.virtual_machine_instance import VirtualMachineInstance

vm = VirtualMachine(client=client, name="my-vm", namespace="default")

# Wait for VM ready status (True = running, None = stopped)
vm.wait_for_ready_status(status=True, timeout=240)

# Wait for a specific status field to become None
vm.wait_for_status_none(status="snapshotInProgress", timeout=240)

# Wait for VMI to be running
vmi = VirtualMachineInstance(client=client, name="my-vm", namespace="default")
vmi.wait_until_running(timeout=240, logs=True)

# Wait for VMI pause state
vmi.wait_for_pause_status(pause=True, timeout=240)
```

See [Working with Virtual Machines (KubeVirt)](working-with-virtual-machines.html) for the full VM lifecycle.

#### DataVolumes

```python
from ocp_resources.datavolume import DataVolume

dv = DataVolume(client=client, name="my-dv", namespace="default")
dv.wait_for_dv_success(
    timeout=600,
    failure_timeout=120,
    pvc_wait_for_bound_timeout=60,
)
```

The `wait_for_dv_success()` method handles the full DataVolume lifecycle: it waits for the status to leave `Pending`, then waits for `Succeeded`, and finally confirms the PVC is `Bound`.

You can pass a `stop_status_func` callback to abort early on custom conditions:

```python
def dv_is_stuck(dv):
    return dv.instance.status.conditions.restartCount > 3

dv.wait_for_dv_success(
    stop_status_func=dv_is_stuck,
    dv=dv,  # passed as keyword argument to the function
)
```

#### VirtualMachineSnapshot / VirtualMachineRestore

```python
from ocp_resources.virtual_machine_snapshot import VirtualMachineSnapshot
from ocp_resources.virtual_machine_restore import VirtualMachineRestore

snapshot = VirtualMachineSnapshot(client=client, name="snap-1", namespace="default")
snapshot.wait_snapshot_done(timeout=240)    # readyToUse + VM snapshotInProgress is None

restore = VirtualMachineRestore(client=client, name="restore-1", namespace="default")
restore.wait_restore_done(timeout=240)     # complete + VM restoreInProgress is None
```

#### MachineSet

```python
from ocp_resources.machine_set import MachineSet

ms = MachineSet(client=client, name="worker-set", namespace="openshift-machine-api")
ms.wait_for_replicas(timeout=300, sleep=1)
```

Returns `True` when ready replicas match desired replicas, `False` on timeout.

#### NodeNetworkConfigurationPolicy

```python
from ocp_resources.node_network_configuration_policy import NodeNetworkConfigurationPolicy

nncp = NodeNetworkConfigurationPolicy(client=client, name="br1-policy")
nncp.wait_for_status_success()  # Uses the resource's own success_timeout
```

Raises `NNCPConfigurationFailed` if the policy fails or finds no matching nodes.

### Timeout constants

The library provides predefined timeout constants you can use with any wait method:

```python
from ocp_resources.utils.constants import (
    TIMEOUT_1SEC,       # 1
    TIMEOUT_5SEC,       # 5
    TIMEOUT_10SEC,      # 10
    TIMEOUT_30SEC,      # 30
    TIMEOUT_1MINUTE,    # 60
    TIMEOUT_2MINUTES,   # 120
    TIMEOUT_4MINUTES,   # 240
    TIMEOUT_10MINUTES,  # 600
)

pod.wait_for_status(status=Pod.Status.RUNNING, timeout=TIMEOUT_10MINUTES)
```

### Custom exception retries

Wait methods handle transient cluster errors automatically (connection resets, etcd leader changes, protocol errors). You can customize which exceptions are retried:

```python
from ocp_resources.utils.constants import (
    PROTOCOL_ERROR_EXCEPTION_DICT,
    DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
)

# Combine default retry exceptions
pod.wait_for_status(
    status=Pod.Status.RUNNING,
    exceptions_dict={
        **PROTOCOL_ERROR_EXCEPTION_DICT,
        **DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
    },
)
```

The default retry exceptions cover:
- `ProtocolError` — network-level failures
- `MaxRetryError`, `ConnectionAbortedError`, `ConnectionResetError` — connection failures
- `InternalServerError` — etcd leader changes, webhook failures, RPC errors
- `ServerTimeoutError` — API server timeouts
- `ForbiddenError` — context deadline exceeded errors

## Troubleshooting

### `TimeoutExpiredError` is raised unexpectedly

- **Increase the timeout.** The default is 240 seconds (4 minutes), which may not be enough for slow operations like DataVolume imports or VM migrations.
- **Check `stop_status`.** `wait_for_status()` defaults `stop_status` to `Status.FAILED`. If your resource legitimately passes through a `"Failed"` phase, set `stop_status` to a value that won't match, or set it to `None`.
- **Inspect the current status.** The error log includes the last observed status. Read it to understand where the resource got stuck.

### `ConditionError` on `wait_for_condition()`

This means the `stop_condition` was matched before the desired condition. The error message includes the full condition details. Review your `stop_condition` and `stop_status` settings.

### Resource exists but `wait_for_status()` times out

The `wait_for_status()` method reads from `status.phase`. Some resources use different status fields (e.g., `status.ready`, `status.readyToUse`, `status.syncStatus`). For those resources, use the resource-specific wait methods documented above, or use `wait_for_condition()` to match on `status.conditions[]` entries instead.

## Related Pages

- [Querying and Watching Resources](querying-resources.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Working with Virtual Machines (KubeVirt)](working-with-virtual-machines.html)
- [Error Handling and Exception Reference](error-handling.html)
- [Resource and NamespacedResource API](resource-api.html)

---

Source: validating-resources.md

Let me start by reading the exclusions file and pages manifest, then explore the codebase.

Now let me look at how the Resource base class integrates validation:

Now let me check the exceptions module:

Now let me check how `deploy` and context managers interact with validation:

Now I have a comprehensive understanding of the validation feature. Let me write the documentation.

# Validating Resources Against OpenAPI Schemas

Catch configuration mistakes before they reach the Kubernetes API by validating resource definitions against OpenAPI schemas. This saves debugging time and prevents failed deployments caused by typos, missing fields, or incorrect types.

## Prerequisites

- openshift-python-wrapper installed (see [Installing and Creating Your First Resource](quickstart.html))
- A connection to a cluster, or use the fake client for offline validation (see [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html))

## Quick Example

```python
from ocp_resources.pod import Pod

pod = Pod(
    name="nginx-pod",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
)

# Validate against the OpenAPI schema — raises ValidationError if invalid
pod.validate()
```

If the resource is valid, `validate()` returns silently. If something is wrong, it raises a `ValidationError` with a clear message pointing to the problematic field.

## Manual Validation

Call `.validate()` on any resource instance to check it against the OpenAPI schema for its kind:

```python
from ocp_resources.exceptions import ValidationError
from ocp_resources.service import Service

service = Service(
    name="nginx-service",
    namespace="default",
    selector={"app": "nginx"},
    ports=[{"port": 80, "targetPort": 80, "protocol": "TCP"}],
)

try:
    service.validate()
    print("Service configuration is valid")
except ValidationError as e:
    print(f"Validation failed: {e}")
```

> **Tip:** Use manual validation in CI pipelines or pre-commit hooks to catch errors before applying resources to a cluster.

## Validating Dictionaries Without Creating Objects

Use the `validate_dict()` class method to validate a raw resource dictionary without instantiating a resource object:

```python
from ocp_resources.deployment import Deployment
from ocp_resources.exceptions import ValidationError

deployment_dict = {
    "apiVersion": "apps/v1",
    "kind": "Deployment",
    "metadata": {"name": "nginx-deployment", "namespace": "default"},
    "spec": {
        "replicas": 3,
        "selector": {"matchLabels": {"app": "nginx"}},
        "template": {
            "metadata": {"labels": {"app": "nginx"}},
            "spec": {
                "containers": [
                    {"name": "nginx", "image": "nginx:1.21", "ports": [{"containerPort": 80}]}
                ],
            },
        },
    },
}

try:
    Deployment.validate_dict(resource_dict=deployment_dict)
    print("Deployment configuration is valid")
except ValidationError as e:
    print(f"Invalid: {e}")
```

This is useful when you have a dictionary from a YAML file or external source and want to validate it before creating the resource.

## Enabling Auto-Validation on Create

Set `schema_validation_enabled=True` when creating a resource instance to automatically validate before every `create()`, `deploy()`, or `update_replace()` call:

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ValidationError

pod = Pod(
    client=client,
    name="auto-validated-pod",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
    schema_validation_enabled=True,
)

# Validation runs automatically before the resource is sent to the API
pod.deploy()
```

If validation fails, the resource is **not** created and a `ValidationError` is raised:

```python
invalid_pod = Pod(
    client=client,
    name="bad-pod",
    namespace="default",
    containers=[{"name": "nginx"}],  # Missing required 'image' field
    schema_validation_enabled=True,
)

try:
    invalid_pod.deploy()
except ValidationError as e:
    print(f"Blocked: {e}")
```

> **Note:** Auto-validation is **disabled by default** (`schema_validation_enabled=False`). You must opt in per instance.

### When Auto-Validation Runs

| Operation          | Auto-validates? | Notes                                      |
|--------------------|-----------------|---------------------------------------------|
| `create()` / `deploy()` | ✅ Yes          | Validates the full resource before sending |
| `update_replace()`      | ✅ Yes          | Validates the replacement dictionary       |
| `update()`              | ❌ No           | Sends a partial patch, not a full resource |

Auto-validation also applies when using context managers, since they call `deploy()` internally. See [Creating and Managing Resources](creating-and-managing-resources.html) for details on resource lifecycle methods.

### Toggling Validation at Runtime

You can enable or disable validation on an existing instance:

```python
pod = Pod(
    client=client,
    name="my-pod",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
)

# Enable after creation
pod.schema_validation_enabled = True
pod.deploy()  # Will validate first

# Disable for a subsequent operation
pod.schema_validation_enabled = False
```

## Advanced Usage

### Validating Resources with Duplicate API Groups

Some resource kinds (like `DNS`) exist in multiple API groups. The validator uses the resource's `api_group` to select the correct schema automatically:

```python
# These two resources share the kind "DNS" but use different API groups.
# Each is validated against its own schema.
dns_config.validate()       # Uses config.openshift.io schema
dns_operator.validate()     # Uses operator.openshift.io schema
```

See [Common Resource Patterns](common-patterns.html) for more on working with resources that share a kind name.

### Pre-Warming the Schema Cache

The first validation for a resource kind loads and resolves the schema (~25ms). Subsequent validations for the same kind use a cache and are much faster (~2ms). Pre-warm the cache at application startup for critical resource types:

```python
from ocp_resources.pod import Pod
from ocp_resources.deployment import Deployment
from ocp_resources.service import Service
from ocp_resources.exceptions import ValidationError

for resource_cls in [Pod, Deployment, Service]:
    try:
        resource_cls.validate_dict({
            "apiVersion": "v1",
            "kind": resource_cls.kind,
            "metadata": {"name": "warmup"}
        })
    except ValidationError:
        pass  # Expected — we just want to load the schemas
```

### Disabling Validation for Bulk Operations

When creating many resources in a loop, disable auto-validation to avoid repeated checks if you've already validated your templates:

```python
# Validate the template once
Pod.validate_dict(resource_dict=pod_template)

# Then create many instances without per-instance validation
for i in range(100):
    pod = Pod(
        client=client,
        name=f"worker-{i}",
        namespace="default",
        containers=[{"name": "app", "image": "myapp:latest"}],
        schema_validation_enabled=False,  # Skip — already validated
    )
    pod.deploy()
```

## Troubleshooting

### Reading Validation Error Messages

Validation errors include three key pieces of information:

```
Resource validation failed for Pod/my-pod
  Field: spec.containers[0].image
  Error: 123 is not of type 'string'
  Expected type: string
```

- **Resource identifier** — which resource failed (kind and name)
- **Field path** — the JSON path to the problematic field
- **Error details** — what went wrong and what was expected

### Common Validation Error Patterns

| Error message pattern                        | Cause                              | Fix                                              |
|----------------------------------------------|-------------------------------------|--------------------------------------------------|
| `'X' is a required property`                 | Missing a required field           | Add the missing field to your resource definition |
| `'X' is not of type 'Y'`                     | Wrong data type                    | Use the correct type (e.g., `3` not `"3"`)       |
| `'X' does not match pattern`                 | Invalid format                     | Follow Kubernetes naming rules (lowercase, alphanumeric, hyphens) |
| `'X' is not one of [...]`                    | Invalid enum value                 | Use one of the allowed values listed in the error |
| `Additional properties are not allowed`       | Unknown field name                 | Check for typos in field names                   |

### Validation Passes But API Rejects the Resource

OpenAPI schema validation catches structural issues (wrong types, missing fields, invalid formats) but cannot check everything the API server validates. For example:

- Unique name constraints within a namespace
- Resource quota limits
- Admission webhook rules
- Cross-resource references (e.g., a ConfigMap that doesn't exist)

If validation passes but the API still rejects your resource, the issue is likely a server-side constraint.

### No Schema Available for a Resource

If no schema is bundled for a resource kind, `validate()` silently succeeds instead of raising an error. A debug-level log message is emitted:

```
No schema found for MyCustomResource, skipping validation
```

To update bundled schemas, see [class-generator CLI Reference](class-generator-cli.html).

## Related Pages

- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Schema Validation and Code Generation Architecture](schema-validation-internals.html)
- [Error Handling and Exception Reference](error-handling.html)
- [class-generator CLI Reference](class-generator-cli.html)
- [Resource and NamespacedResource API](resource-api.html)

---

Source: managing-resource-lists.md

Let me explore the codebase to understand ResourceList and NamespacedResourceList.

Now let me check the example file more carefully and also look at how the `NamespacedResourceList` actually works in the example vs the source code — I noticed the example passes `namespaces=[ns.name for ns in namespaces]` (a list of strings) but the source code expects a `ResourceList` object. Let me verify.

The example file seems to have an outdated API for `NamespacedResourceList` (passing a list of strings), but the actual source code expects a `ResourceList` object. The tests confirm it passes a `ResourceList`. Let me check if the example is outdated:

The example file is inconsistent with the actual source code. The source code validates that each element in `namespaces` has `ns.kind == "Namespace"`, meaning it expects actual Namespace resource objects (from a ResourceList), not strings. The test confirms this by passing a `ResourceList` fixture. I'll document based on the actual source code, not the potentially outdated example.

Let me also check what methods and properties are available:

Now I have all the information I need. Let me write the documentation.

# Managing Bulk Resources with ResourceList

When you need to create several copies of the same resource or deploy a resource across multiple namespaces, doing it one at a time is tedious and error-prone. `ResourceList` and `NamespacedResourceList` let you create, deploy, and clean up groups of resources in a single operation.

## Prerequisites

- A working cluster connection (see [Connecting to Clusters](connecting-to-clusters.html))
- Familiarity with creating individual resources (see [Creating and Managing Resources](creating-and-managing-resources.html))

## Quick Example

Create three namespaces and deploy them in two lines:

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import ResourceList, get_client

client = get_client()

namespaces = ResourceList(resource_class=Namespace, num_resources=3, client=client, name="my-ns")
namespaces.deploy()
# Creates: my-ns-1, my-ns-2, my-ns-3
```

Or use a context manager for automatic cleanup:

```python
with ResourceList(client=client, resource_class=Namespace, name="my-ns", num_resources=3) as namespaces:
    print(namespaces[0].name)  # "my-ns-1"
# All three namespaces are deleted when the block exits
```

## Creating Multiple Copies of a Resource

`ResourceList` creates N copies of a resource type, each with an auto-generated name based on a base name you provide.

**Step 1:** Choose a resource class and specify how many copies you want:

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import ResourceList, get_client

client = get_client()
namespaces = ResourceList(
    resource_class=Namespace,
    num_resources=3,
    client=client,
    name="test-ns",
)
```

This creates three `Namespace` objects named `test-ns-1`, `test-ns-2`, and `test-ns-3`. They are not yet deployed to the cluster.

**Step 2:** Deploy all resources:

```python
namespaces.deploy()
```

**Step 3:** Work with individual resources by index or iteration:

```python
# Access by index
first_ns = namespaces[0]
print(first_ns.name)  # "test-ns-1"

# Iterate over all resources
for ns in namespaces:
    print(ns.name)

# Check how many resources are in the list
print(len(namespaces))  # 3
```

**Step 4:** Clean up all resources when done:

```python
namespaces.clean_up()
```

> **Note:** `clean_up()` deletes resources in reverse order. This ensures dependent resources are removed before the resources they depend on.

## Deploying One Resource Per Namespace

`NamespacedResourceList` creates one instance of a namespaced resource in each namespace from a `ResourceList`.

```python
from ocp_resources.namespace import Namespace
from ocp_resources.pod import Pod
from ocp_resources.resource import NamespacedResourceList, ResourceList, get_client

client = get_client()

# First, create the namespaces
namespaces = ResourceList(resource_class=Namespace, num_resources=3, client=client, name="app-ns")
namespaces.deploy()

# Then create one pod in each namespace
pods = NamespacedResourceList(
    client=client,
    resource_class=Pod,
    namespaces=namespaces,
    name="web-server",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
)
pods.deploy()
```

This creates three pods, all named `web-server`, one in each namespace (`app-ns-1`, `app-ns-2`, `app-ns-3`).

```python
for pod in pods:
    print(f"{pod.name} in {pod.namespace}")
# web-server in app-ns-1
# web-server in app-ns-2
# web-server in app-ns-3
```

> **Warning:** The `namespaces` parameter must be a `ResourceList` containing only Namespace resources. Passing any other resource type raises a `TypeError`.

## Passing Extra Resource Parameters

Any additional keyword arguments you pass are forwarded to each resource's constructor. This lets you configure resources beyond just the name:

```python
from ocp_resources.role import Role
from ocp_resources.resource import NamespacedResourceList

roles = NamespacedResourceList(
    client=client,
    resource_class=Role,
    namespaces=namespaces,
    name="viewer-role",
    rules=[
        {
            "apiGroups": [""],
            "resources": ["pods"],
            "verbs": ["get", "list", "watch"],
        }
    ],
)
roles.deploy()
```

## Advanced Usage

### Context Managers for Automatic Lifecycle

Both `ResourceList` and `NamespacedResourceList` support context managers. Resources are deployed on entry and cleaned up on exit — ideal for tests:

```python
with ResourceList(client=client, resource_class=Namespace, name="test-ns", num_resources=3) as namespaces:
    with NamespacedResourceList(
        client=client,
        resource_class=Pod,
        namespaces=namespaces,
        name="test-pod",
        containers=[{"name": "app", "image": "busybox"}],
    ) as pods:
        # All namespaces and pods are deployed
        assert len(pods) == 3
    # Pods are cleaned up here
# Namespaces are cleaned up here
```

> **Tip:** Nest the context managers so that namespaced resources are cleaned up before their namespaces are deleted.

### Waiting During Deploy and Cleanup

Pass `wait=True` to `deploy()` to wait for each resource to reach a ready state. By default, `clean_up()` already waits for deletion to complete.

```python
namespaces = ResourceList(resource_class=Namespace, num_resources=2, client=client, name="ns")
namespaces.deploy(wait=True)

# Later...
namespaces.clean_up(wait=True)   # wait=True is the default
namespaces.clean_up(wait=False)  # Skip waiting for deletion
```

For more on waiting for specific conditions, see [Waiting for Resource Conditions and Status](waiting-for-conditions.html).

### Comparison: ResourceList vs NamespacedResourceList

| Feature | `ResourceList` | `NamespacedResourceList` |
|---|---|---|
| **Purpose** | Create N copies of any resource | Create one resource per namespace |
| **Naming** | Auto-generates `{name}-1`, `{name}-2`, … | Same name in each namespace |
| **Required params** | `resource_class`, `num_resources`, `name` | `resource_class`, `namespaces`, `name` |
| **Use case** | Multiple namespaces, nodes, cluster-scoped resources | Pods, services, roles — anything namespace-scoped |
| **Context manager** | ✅ | ✅ |

## Troubleshooting

**TypeError: "All the resources in namespaces should be namespaces"**

You passed a `ResourceList` that contains non-Namespace resources to `NamespacedResourceList`. Make sure the `namespaces` parameter is a `ResourceList` created with `resource_class=Namespace`.

**Resources not appearing on the cluster**

Creating a `ResourceList` or `NamespacedResourceList` only instantiates the Python objects — it does not deploy them. Call `.deploy()` or use a context manager (`with ... as`) to actually create the resources on the cluster.

**Cleanup order issues**

Always clean up namespaced resources before their namespaces. When using context managers, nest them so the inner block (namespaced resources) exits first. When calling `clean_up()` manually, clean up dependent resources before their parents.

## Related Pages

- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html)
- [Common Resource Patterns](common-patterns.html)
- [Resource and NamespacedResource API](resource-api.html)

---

Source: editing-resources-temporarily.md

Let me start by reading the knowledge graph and understanding the repository structure.

Now let me find the ResourceEditor source code.

Now let me look at how ResourceEditor is used in the NNCP file and any tests:

Let me look at the import section and any related constants:

Now let me look at the `update` and `update_replace` methods on the Resource class to understand the two actions:

Now let me look at the imports to verify how ResourceEditor is exported:

Now let me check for any additional examples or documentation:

Let me look at the full ResourceEditor code once more to make sure I've captured everything, and also check the `_apply_patches` method details and what `action` values are supported:

Now I have a complete picture of the ResourceEditor class. Let me write the documentation.

# Editing Resources Temporarily with ResourceEditor

Temporarily modify Kubernetes and OpenShift resources during tests and have their original values automatically restored when you're done. This is essential for test isolation—ensuring that changes made during a test don't leak into other tests or persist in the cluster.

## Prerequisites

- `openshift-python-wrapper` installed in your project (see [Installing and Creating Your First Resource](quickstart.html))
- A cluster connection established with a privileged client (see [Connecting to Clusters](connecting-to-clusters.html))
- One or more existing resources on the cluster that you want to temporarily modify

> **Warning:** The client used to retrieve resources must have sufficient privileges. Do not use an unprivileged user client with `ResourceEditor`.

## Quick Example

```python
from ocp_resources.resource import ResourceEditor
from ocp_resources.node import Node

node = Node(name="worker-0")

with ResourceEditor(
    patches={node: {"metadata": {"labels": {"test-label": "true"}}}}
):
    # The node now has the label "test-label": "true"
    # Run your test logic here...
    pass

# The label is automatically removed when the block exits
```

That's it—when the `with` block ends (even if an exception occurs), the original resource state is restored.

## Step-by-Step: Using ResourceEditor as a Context Manager

### 1. Import and identify your target resource

```python
from ocp_resources.resource import ResourceEditor
from ocp_resources.config_map import ConfigMap

cm = ConfigMap(name="my-config", namespace="my-namespace")
```

### 2. Define your patch

Patches are dictionaries that mirror the resource's YAML structure. Only include the fields you want to change:

```python
patch = {"data": {"feature_flag": "enabled", "log_level": "debug"}}
```

### 3. Wrap your test logic in a `with` block

```python
with ResourceEditor(patches={cm: patch}):
    # ConfigMap now has updated data fields
    assert cm.instance.data.feature_flag == "enabled"
    # Run tests that depend on the modified config...
```

### 4. Original values are restored automatically

When the `with` block exits, the original `data` values are written back to the resource. Fields that didn't exist before the patch are removed (set to `None` in the restore payload, which tells the API to delete them).

## Patching Multiple Resources at Once

Pass multiple resource-patch pairs in a single `ResourceEditor`:

```python
from ocp_resources.resource import ResourceEditor
from ocp_resources.namespace import Namespace
from ocp_resources.config_map import ConfigMap

ns = Namespace(name="test-ns")
cm = ConfigMap(name="app-config", namespace="test-ns")

patches = {
    ns: {"metadata": {"labels": {"env": "test"}}},
    cm: {"data": {"database_url": "postgres://test-db:5432"}},
}

with ResourceEditor(patches=patches):
    # Both resources are patched; run tests here
    pass
# Both resources are restored to their original state
```

## Using update() and restore() Without a Context Manager

If you need finer control over when patches are applied and reverted—for example, in `setup`/`teardown` fixtures—use the `update()` and `restore()` methods directly:

```python
from ocp_resources.resource import ResourceEditor
from ocp_resources.config_map import ConfigMap

cm = ConfigMap(name="app-config", namespace="default")

editor = ResourceEditor(
    patches={cm: {"data": {"mode": "maintenance"}}}
)

# Apply the patch and create backups
editor.update(backup_resources=True)

# ... run tests ...

# Restore original values
editor.restore()
```

> **Note:** When calling `update()` manually, pass `backup_resources=True` to ensure original values are captured. Without this flag, no backup is created and `restore()` will have nothing to revert.

## Advanced Usage

### Replace Instead of Patch

By default, `ResourceEditor` uses the `"update"` action, which sends a merge-patch (`application/merge-patch+json`). This merges your changes into the existing resource. If you need to *replace* the entire resource (for example, to remove fields that merge-patch can't delete), use `action="replace"`:

```python
from ocp_resources.resource import ResourceEditor

with ResourceEditor(
    patches={my_resource: {"spec": {"replicas": 3}}},
    action="replace",
):
    # The resource is fully replaced, not merged
    pass
```

| Action | Method Used | Behavior |
|---|---|---|
| `"update"` (default) | Merge patch | Merges patch into existing resource; cannot remove fields by omission |
| `"replace"` | Full replacement | Replaces the entire resource; fields not in the patch are removed |

> **Warning:** With `action="replace"`, the patch must include all required fields for the resource, not just the fields you want to change. The replacement payload automatically includes `metadata.name`, `metadata.namespace`, `metadata.resourceVersion`, `kind`, and `apiVersion`.

### Providing Your Own Backups

If you already know what the restore state should be (or need custom restore logic), pass `user_backups` to skip the automatic backup calculation:

```python
from ocp_resources.resource import ResourceEditor

custom_backup = {my_resource: {"spec": {"replicas": 1}}}

with ResourceEditor(
    patches={my_resource: {"spec": {"replicas": 5}}},
    user_backups=custom_backup,
):
    # Resource scaled to 5 replicas
    pass
# Restored to 1 replica (your custom backup), regardless of what the original value was
```

### Inspecting Backups

Access the automatically generated backup data through the `backups` property:

```python
editor = ResourceEditor(
    patches={my_resource: {"metadata": {"labels": {"env": "staging"}}}}
)
editor.update(backup_resources=True)

# See what was backed up
print(editor.backups)
# {<Resource>: {'metadata': {'labels': {'env': 'production'}}}}

editor.restore()
```

### No-Op Detection

`ResourceEditor` automatically detects when a patch produces no actual changes. If the patch values already match the resource's current state, the resource is skipped entirely and a warning is logged:

```
ResourceEdit: no diff found in patch for my-resource -- skipping
```

This prevents unnecessary API calls and avoids triggering watch events for no-op changes.

### Automatic Retry on Conflicts

All patch and restore operations are wrapped with automatic retry logic. If a `ConflictError` occurs (for example, due to a concurrent update changing the `resourceVersion`), the operation is retried automatically with a 5-second sleep interval and a 30-second timeout.

## API Reference

### Constructor

```python
ResourceEditor(
    patches: dict,
    action: str = "update",
    user_backups: dict | None = None,
)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `patches` | `dict` | *(required)* | Mapping of resource objects to patch dicts: `{resource: {yaml_patch}}` |
| `action` | `str` | `"update"` | Either `"update"` (merge patch) or `"replace"` (full replacement) |
| `user_backups` | `dict \| None` | `None` | Custom backup dicts to use for restore instead of auto-generated ones |

### Properties

| Property | Return Type | Description |
|---|---|---|
| `backups` | `dict` | The backup data captured for each patched resource: `{resource: backup_dict}` |
| `patches` | `dict` | The patches dict provided in the constructor |

### Methods

| Method | Parameters | Description |
|---|---|---|
| `update(backup_resources=False)` | `backup_resources: bool` | Apply patches. Set `backup_resources=True` to capture original values for later restore. |
| `restore()` | — | Re-apply the stored backup dicts to revert all patched resources. |

### Context Manager Protocol

| Method | Behavior |
|---|---|
| `__enter__()` | Calls `update(backup_resources=True)` and returns the `ResourceEditor` instance |
| `__exit__()` | Calls `restore()` to revert all changes |

## Troubleshooting

| Problem | Cause | Solution |
|---|---|---|
| Restore doesn't revert changes | Called `update()` without `backup_resources=True` | Always pass `backup_resources=True` when calling `update()` manually, or use the context manager (`with` block) which does this automatically |
| `NotFoundError` during backup | Resource cannot be found by name (known with some CRDs like `ServiceMonitor`) | `ResourceEditor` handles this automatically by falling back to a field-selector query |
| `ConflictError` during patch | Another process modified the resource concurrently | Automatic retries handle this; if it persists beyond 30 seconds, check for controllers continuously modifying the resource |
| Patch has no effect | The resource already has the values in your patch | This is expected—`ResourceEditor` logs a warning and skips the resource. Check that you're patching the correct fields. |
| `replace` action fails | Patch is missing required fields | When using `action="replace"`, provide all required spec fields since the entire resource is replaced |

> **Tip:** For creating and cleaning up resources (rather than editing existing ones), see [Creating and Managing Resources](creating-and-managing-resources.html). For managing groups of similar resources, see [Managing Bulk Resources with ResourceList](managing-resource-lists.html).

## Related Pages

- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Resource and NamespacedResource API](resource-api.html)
- [Common Resource Patterns](common-patterns.html)
- [Managing Bulk Resources with ResourceList](managing-resource-lists.html)
- [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html)

---

Source: working-with-virtual-machines.md

Let me start by exploring the knowledge graph and understanding the repository structure.

Now let me look at the remaining VM-related resources and check the `RunStrategy` and `Status` constants, as well as the `data_volume.py`:

Now let me check how the `api_request` method works in the base `Resource` class:

Now let me look at the `data_volume.py` to check any VM-related interactions:

Now let me check the `RunStrategy` constants in VirtualMachineInstance:

Now let me look at the `Status` class in the base `Resource` for context and the VMI's `RunStrategy`:

Let me find the Status class definition:

Now let me check the `ApiGroup` constants for KubeVirt:

Now let me check for any tests related to VMs:

Now let me check the `deploy` and `clean_up` method signatures:

Now I have a thorough understanding of the VM-related API. Let me write the documentation:

# Working with Virtual Machines (KubeVirt)

Manage the full lifecycle of KubeVirt virtual machines on OpenShift — create VMs, start and stop them, live-migrate between nodes, take snapshots, and interact with their running instances — all from Python.

## Prerequisites

- `openshift-python-wrapper` installed (see [Installing and Creating Your First Resource](quickstart.html))
- A cluster connection configured (see [Connecting to Clusters](connecting-to-clusters.html))
- KubeVirt / OpenShift Virtualization installed on your cluster

## Quick Example

```python
from ocp_resources.virtual_machine import VirtualMachine

vm = VirtualMachine(
    name="my-vm",
    namespace="my-namespace",
    body={
        "spec": {
            "domain": {
                "devices": {"disks": [{"disk": {"bus": "virtio"}, "name": "rootdisk"}]},
                "resources": {"requests": {"memory": "1Gi"}},
            },
            "volumes": [{"name": "rootdisk", "containerDisk": {"image": "quay.io/containerdisks/fedora"}}],
        }
    },
)

with vm:
    vm.start(wait=True)
    print(f"VM ready: {vm.ready}")       # True
    print(f"Status: {vm.printable_status}")  # "Running"
    vm.stop(wait=True)
# VM is automatically cleaned up when the context manager exits
```

## Creating a VirtualMachine

```python
from ocp_resources.virtual_machine import VirtualMachine

vm = VirtualMachine(
    name="my-vm",
    namespace="my-namespace",
    body={
        "spec": {
            "domain": {
                "devices": {"disks": [{"disk": {"bus": "virtio"}, "name": "rootdisk"}]},
                "resources": {"requests": {"memory": "2Gi"}},
            },
            "volumes": [
                {
                    "name": "rootdisk",
                    "containerDisk": {"image": "quay.io/containerdisks/fedora"},
                }
            ],
        }
    },
)
vm.deploy()
```

The `body` parameter accepts a dictionary that maps to the KubeVirt VM `spec.template` structure. If you omit `body`, a minimal skeleton (`{"template": {"spec": {}}}`) is created automatically.

You can also create a VM from a YAML file:

```python
vm = VirtualMachine(
    name="my-vm",
    namespace="my-namespace",
    yaml_file="vm-definition.yaml",
)
vm.deploy()
```

### Constructor Parameters

| Parameter | Type | Default | Description |
|---|---|---|---|
| `name` | `str` | `None` | Name of the VirtualMachine resource |
| `namespace` | `str` | `None` | Target namespace |
| `client` | `DynamicClient` | `None` | Kubernetes client; auto-resolved if omitted |
| `body` | `dict` | `None` | VM spec body dictionary |
| `teardown` | `bool` | `True` | Whether to delete the resource on cleanup |
| `yaml_file` | `str` | `None` | Path to a YAML file defining the VM |
| `delete_timeout` | `int` | `240` (seconds) | Timeout for delete operations |

> **Tip:** Use the context manager (`with vm:`) to ensure automatic cleanup. See [Creating and Managing Resources](creating-and-managing-resources.html) for details on `deploy()`, `clean_up()`, and context managers.

## Starting, Stopping, and Restarting

### Start a VM

```python
# Fire-and-forget
vm.start()

# Wait until the VM is ready
vm.start(wait=True)

# Custom timeout (seconds)
vm.start(timeout=600, wait=True)
```

### Stop a VM

```python
# Fire-and-forget
vm.stop()

# Wait for the VM to stop and the VMI to be deleted
vm.stop(wait=True)

# Custom timeouts
vm.stop(timeout=300, vmi_delete_timeout=300, wait=True)
```

### Restart a VM

```python
vm.restart(wait=True)
```

When `wait=True`, `restart()` waits for the old virt-launcher pod to be deleted and the new VMI to reach the `Running` state.

## Checking VM Status

```python
# Boolean ready status: True if running, None if stopped
vm.ready  # True

# Human-readable status string
vm.printable_status  # "Running", "Stopped", "Starting", "Migrating", etc.

# Wait for ready status
vm.wait_for_ready_status(status=True, timeout=300)   # Wait for running
vm.wait_for_ready_status(status=None, timeout=300)    # Wait for stopped

# Wait for a specific status field to clear
vm.wait_for_status_none(status="snapshotInProgress")
```

### Run Strategy Constants

Use built-in constants for the `runStrategy` field:

```python
VirtualMachine.RunStrategy.ALWAYS            # "Always"
VirtualMachine.RunStrategy.HALTED            # "Halted"
VirtualMachine.RunStrategy.MANUAL            # "Manual"
VirtualMachine.RunStrategy.RERUNONFAILURE    # "RerunOnFailure"
```

### Status Constants

```python
VirtualMachine.Status.STARTING       # "Starting"
VirtualMachine.Status.RUNNING        # "Running"  (inherited)
VirtualMachine.Status.STOPPED        # "Stopped"
VirtualMachine.Status.STOPPING       # "Stopping"
VirtualMachine.Status.MIGRATING      # "Migrating"
VirtualMachine.Status.PAUSED         # "Paused"
VirtualMachine.Status.PROVISIONING   # "Provisioning"
VirtualMachine.Status.ERROR_UNSCHEDULABLE  # "ErrorUnschedulable"
VirtualMachine.Status.DATAVOLUME_ERROR     # "DataVolumeError"
```

## Working with VirtualMachineInstances

Every running VM has an associated `VirtualMachineInstance` (VMI). Access it through the VM or create one directly.

### Accessing the VMI from a VM

```python
vmi = vm.vmi  # Returns a VirtualMachineInstance object
```

### Creating a Standalone VMI

```python
from ocp_resources.virtual_machine_instance import VirtualMachineInstance

vmi = VirtualMachineInstance(
    name="my-vmi",
    namespace="my-namespace",
    domain={
        "devices": {"disks": [{"disk": {"bus": "virtio"}, "name": "rootdisk"}]},
        "resources": {"requests": {"memory": "1Gi"}},
    },
    volumes=[
        {"name": "rootdisk", "containerDisk": {"image": "quay.io/containerdisks/fedora"}}
    ],
)
vmi.deploy()
```

> **Note:** The `domain` parameter is **required** when creating a VMI programmatically. Omitting it raises `MissingRequiredArgumentError`.

### VMI Constructor Parameters

| Parameter | Type | Required | Description |
|---|---|---|---|
| `domain` | `dict` | **Yes** | Domain spec (CPU, memory, devices) |
| `volumes` | `list` | No | Volumes to mount |
| `networks` | `list` | No | Network interfaces |
| `affinity` | `dict` | No | Scheduling affinity rules |
| `node_selector` | `dict` | No | Node label selector |
| `tolerations` | `list` | No | Toleration rules |
| `eviction_strategy` | `str` | No | `"LiveMigrate"`, `"LiveMigrateIfPossible"`, `"None"`, or `"External"` |
| `termination_grace_period_seconds` | `int` | No | Seconds to wait before force-killing |
| `hostname` | `str` | No | VM hostname |
| `priority_class_name` | `str` | No | Pod priority class |
| `scheduler_name` | `str` | No | Custom scheduler name |
| `start_strategy` | `str` | No | Set to `"Paused"` to start paused |
| `access_credentials` | `list` | No | Public keys to inject |
| `topology_spread_constraints` | `list` | No | Topology spread rules |

### Waiting for a VMI to Run

```python
vmi.wait_until_running(timeout=300)
```

If the VMI fails to start, `wait_until_running` raises `TimeoutExpiredError` and logs the virt-launcher pod status and logs for debugging. Set `logs=False` to suppress this:

```python
vmi.wait_until_running(timeout=300, logs=False)
```

You can also specify a `stop_status` to abort early if the VMI enters a failure state:

```python
vmi.wait_until_running(timeout=300, stop_status="Failed")
```

### Pausing and Unpausing

```python
vmi.pause(wait=True)    # Pause the VMI
vmi.unpause(wait=True)  # Resume the VMI
```

### Resetting a VMI

```python
vmi.reset()  # Hard-reset the VMI (like pressing the reset button)
```

### Getting the VMI Node

```python
node = vmi.get_node()
print(node.name)
```

### Network Interfaces

```python
# All interfaces from the VMI status
interfaces = vmi.interfaces

# Get IP address for a specific interface
ip = vmi.interface_ip("eth0")
```

### Guest Agent Information

When the QEMU guest agent is installed in the VM:

```python
vmi.guest_os_info       # OS information
vmi.guest_fs_info       # Filesystem information
vmi.guest_user_info     # Logged-in user information
vmi.os_version          # OS version string
```

> **Note:** If no guest agent is installed, `os_version` returns an empty dict and logs a warning.

### Accessing the Virt-Launcher Pod

```python
pod = vmi.get_virt_launcher_pod()
print(pod.name)
print(pod.status)

# Access logs from the compute container
logs = pod.log(container="compute")
```

For clusters where a privileged client is required:

```python
pod = vmi.get_virt_launcher_pod(privileged_client=admin_client)
```

### Accessing the Virt-Handler Pod

```python
handler_pod = vmi.get_virt_handler_pod(privileged_client=admin_client)
```

### Executing Virsh Commands

Run virsh commands inside the virt-launcher pod:

```python
# Get the XML definition
xml_output = vmi.get_xml()

# Get the XML as a parsed Python dict
xml_dict = vmi.get_xml_dict()

# Get domain memory stats
memstats = vmi.get_dommemstat()

# Run any arbitrary virsh command
output = vmi.execute_virsh_command(command="dominfo")
```

### Checking Root Status

```python
pod = vmi.get_virt_launcher_pod()

# Check if the virt-launcher pod runs as root
VirtualMachineInstance.is_pod_root(pod=pod)  # True or False

# Get the pod's runAsUser UID
VirtualMachineInstance.get_pod_user_uid(pod=pod)  # int or None
```

## Live Migration

Trigger a live migration by creating a `VirtualMachineInstanceMigration`:

```python
from ocp_resources.virtual_machine_instance_migration import VirtualMachineInstanceMigration

migration = VirtualMachineInstanceMigration(
    name="migrate-my-vm",
    namespace="my-namespace",
    vmi_name="my-vm",
)
migration.deploy()
```

Monitor migration by checking the VMI's status. Once migration completes, the VMI runs on the target node. See [Waiting for Resource Conditions and Status](waiting-for-conditions.html) for condition-based waiting.

### Migration Resource Quotas

Control resource allocation for migrations:

```python
from ocp_resources.virtual_machine_migration_resource_quota import VirtualMachineMigrationResourceQuota

quota = VirtualMachineMigrationResourceQuota(
    name="migration-quota",
    namespace="my-namespace",
    requests_cpu="2",
    requests_memory="4Gi",
    limits_cpu="4",
    limits_memory="8Gi",
)
quota.deploy()
```

## Snapshots and Restores

### Taking a Snapshot

```python
from ocp_resources.virtual_machine_snapshot import VirtualMachineSnapshot

snapshot = VirtualMachineSnapshot(
    name="my-vm-snapshot",
    namespace="my-namespace",
    vm_name="my-vm",
)
snapshot.deploy()

# Wait for the snapshot to be ready
snapshot.wait_ready_to_use(timeout=300)

# Wait for both snapshot readiness AND the VM's snapshotInProgress to clear
snapshot.wait_snapshot_done(timeout=300)
```

### Restoring from a Snapshot

```python
from ocp_resources.virtual_machine_restore import VirtualMachineRestore

restore = VirtualMachineRestore(
    name="my-vm-restore",
    namespace="my-namespace",
    vm_name="my-vm",
    snapshot_name="my-vm-snapshot",
)
restore.deploy()

# Wait for restore to complete
restore.wait_restore_done(timeout=300)
```

The `wait_restore_done()` method checks both the restore's `complete` status and that the VM's `restoreInProgress` field is cleared.

| Parameter | Type | Description |
|---|---|---|
| `vm_name` | `str` | Target VirtualMachine name |
| `snapshot_name` | `str` | Source VirtualMachineSnapshot name |
| `volume_restore_policy` | `str` | Optional policy for volume restoration |

## Cloning VMs

```python
from ocp_resources.virtual_machine_clone import VirtualMachineClone

clone = VirtualMachineClone(
    name="clone-my-vm",
    namespace="my-namespace",
    source_name="my-vm",
    target_name="my-vm-clone",
)
clone.deploy()
```

> **Note:** `source_name` is **required**. If omitted, `MissingRequiredArgumentError` is raised.

### Clone Parameters

| Parameter | Type | Default | Description |
|---|---|---|---|
| `source_name` | `str` | **Required** | Name of the source VM |
| `source_kind` | `str` | `"VirtualMachine"` | Kind of the source resource |
| `target_name` | `str` | Auto-generated | Name for the cloned VM |
| `label_filters` | `list` | `None` | Labels to copy, e.g. `["*", "!someKey/*"]` |
| `annotation_filters` | `list` | `None` | Annotations to copy |
| `new_mac_addresses` | `dict` | `None` | Override MACs: `{"nic0": "00:11:22:33:44:55"}` |
| `new_smbios_serial` | `str` | `None` | Override SMBIOS serial number |
| `volume_name_policy` | `str` | `None` | Volume naming policy for the clone |

## Exporting VMs

```python
from ocp_resources.virtual_machine_export import VirtualMachineExport

export = VirtualMachineExport(
    name="export-my-vm",
    namespace="my-namespace",
    source={"apiGroup": "kubevirt.io", "kind": "VirtualMachine", "name": "my-vm"},
    token_secret_ref="my-export-token",
    ttl_duration="1h",
)
export.deploy()
```

> **Note:** The `source` parameter is **required**. If omitted, `MissingRequiredArgumentError` is raised.

## Advanced Usage

### Instance Types and Preferences

KubeVirt instance types define reusable VM sizing profiles. Preferences define non-sizing defaults (clock, firmware, devices).

#### Namespaced Instance Type

```python
from ocp_resources.virtual_machine_instancetype import VirtualMachineInstancetype

instancetype = VirtualMachineInstancetype(
    name="small",
    namespace="my-namespace",
    cpu={"guest": 2},
    memory={"guest": "4Gi"},
)
instancetype.deploy()
```

Both `cpu` and `memory` are **required**. Omitting either raises `MissingRequiredArgumentError`.

#### Cluster-Scoped Instance Type

```python
from ocp_resources.virtual_machine_cluster_instancetype import VirtualMachineClusterInstancetype

cluster_instancetype = VirtualMachineClusterInstancetype(
    name="large",
    cpu={"guest": 8},
    memory={"guest": "16Gi"},
    gpus=[{"deviceName": "nvidia.com/A100", "name": "gpu0"}],
)
cluster_instancetype.deploy()
```

#### Namespaced Preference

```python
from ocp_resources.virtual_machine_preference import VirtualMachinePreference

preference = VirtualMachinePreference(
    name="linux-pref",
    namespace="my-namespace",
    cpu={"preferredCPUTopology": "preferSockets"},
    firmware={"preferredUseEfi": True},
)
preference.deploy()
```

#### Cluster-Scoped Preference

```python
from ocp_resources.virtual_machine_cluster_preference import VirtualMachineClusterPreference

cluster_preference = VirtualMachineClusterPreference(
    name="windows-pref",
    clock={"preferredTimer": {"hpet": {"present": False}}},
    features={"preferredHyperv": {"spinlocks": {"spinlocks": 8191}}},
)
cluster_preference.deploy()
```

### VMI Replica Sets

Scale out VMIs horizontally:

```python
from ocp_resources.virtual_machine_instance_replica_set import VirtualMachineInstanceReplicaSet

replica_set = VirtualMachineInstanceReplicaSet(
    name="my-vmi-rs",
    namespace="my-namespace",
    replicas=3,
    selector={"matchLabels": {"app": "my-vmi"}},
    template={
        "metadata": {"labels": {"app": "my-vmi"}},
        "spec": {
            "domain": {
                "devices": {},
                "resources": {"requests": {"memory": "1Gi"}},
            }
        },
    },
)
replica_set.deploy()
```

Both `selector` and `template` are **required**.

### Storage Migration

Migrate VM storage between storage classes:

```python
from ocp_resources.virtual_machine_storage_migration_plan import VirtualMachineStorageMigrationPlan
from ocp_resources.virtual_machine_storage_migration import VirtualMachineStorageMigration

# Create a migration plan
plan = VirtualMachineStorageMigrationPlan(
    name="migrate-storage",
    namespace="my-namespace",
    virtual_machines=[
        {"name": "my-vm", "volumes": [{"name": "rootdisk", "target": {"storageClassName": "fast-ssd"}}]}
    ],
    retention_policy="deleteSource",
)
plan.deploy()

# Execute the migration
migration = VirtualMachineStorageMigration(
    name="migrate-storage-exec",
    namespace="my-namespace",
    virtual_machine_storage_migration_plan_ref={"name": "migrate-storage", "namespace": "my-namespace"},
)
migration.deploy()
```

### VM Templates

```python
from ocp_resources.virtual_machine_template import VirtualMachineTemplate

template = VirtualMachineTemplate(
    name="fedora-template",
    namespace="my-namespace",
    virtual_machine={
        "spec": {
            "template": {
                "spec": {
                    "domain": {
                        "devices": {},
                        "resources": {"requests": {"memory": "2Gi"}},
                    }
                }
            }
        }
    },
    parameters=[
        {"name": "VM_NAME", "description": "Name of the VM", "required": True}
    ],
    message="Use this template to create Fedora VMs.",
)
template.deploy()
```

### Listing VMs Across Namespaces

```python
# List all VMs in a namespace
for vm in VirtualMachine.get(namespace="my-namespace"):
    print(f"{vm.name}: {vm.instance.get('status', {}).get('printableStatus')}")

# List with label selectors
for vm in VirtualMachine.get(
    namespace="my-namespace",
    label_selector="app=webserver",
):
    print(vm.name)
```

See [Querying and Watching Resources](querying-resources.html) for more query options.

### Getting VM Network Interfaces

```python
interfaces = vm.get_interfaces()
for iface in interfaces:
    print(f"Name: {iface['name']}, Model: {iface.get('model', 'default')}")
```

## Troubleshooting

### VM Stuck in "Starting" Status

Check the VMI's virt-launcher pod for scheduling or image-pull issues:

```python
vmi = vm.vmi
try:
    pod = vmi.get_virt_launcher_pod()
    print(f"Pod status: {pod.status}")
    print(pod.log(container="compute"))
except ResourceNotFoundError:
    print("Virt-launcher pod not found — VM may not have been scheduled yet")
```

### TimeoutExpiredError on start/stop

Increase the timeout parameter:

```python
vm.start(timeout=600, wait=True)  # 10 minutes instead of default 4
```

### MissingRequiredArgumentError

This is raised when a required constructor parameter is missing. Common cases:

| Resource | Required Parameters |
|---|---|
| `VirtualMachineInstance` | `domain` |
| `VirtualMachineInstancetype` | `cpu`, `memory` |
| `VirtualMachineClone` | `source_name` |
| `VirtualMachineExport` | `source` |
| `VirtualMachineInstanceReplicaSet` | `selector`, `template` |
| `VirtualMachineStorageMigration` | `virtual_machine_storage_migration_plan_ref` |
| `VirtualMachineTemplate` | `virtual_machine` |

### Guest Agent Not Reporting

If `vmi.os_version` returns an empty dict, the QEMU guest agent is not installed or not running inside the VM. Install and start `qemu-guest-agent` in the guest OS.

## Related Pages

- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Querying and Watching Resources](querying-resources.html)
- [Common Resource Patterns](common-patterns.html)
- [Executing Commands in Pods and Retrieving Logs](pod-execution-and-logs.html)

---

Source: testing-without-cluster.md

Let me start by exploring the repository structure and the knowledge graph.

Now let me look at the test files for the fake client:

Let me look at a few more test files to understand patterns used:

Now let me look at how `get_client` works with `fake=True` and how resource classes use the client:

Now let me look at a few more test examples to understand patterns:

Now let me check how resources use the fake client and how `deploy`, `clean_up`, `update` work:

Let me look at additional test files for more patterns - ConfigMap, events, and the configmap resource:

Now let me check how watch and events are used, and how to use the lower-level API:

Now let me check the `events` and `watch` implementation in the Resource class:

Now let me look at a few more resource test files and the low-level API usage:

Let me check for import paths and the `ocp_resources.resource` module to understand the public API:

Let me check how the `wait_for_condition` and other resource methods work with the fake client:

Now I have enough information to write the documentation page. Let me check a couple more things:

Now I have all the information I need. Let me write the documentation page.

# Testing Without a Cluster Using the Fake Client

Write unit tests that create, read, update, and delete Kubernetes resources — including status checks, watch events, and custom CRDs — without connecting to a real cluster.

## Prerequisites

- `openshift-python-wrapper` installed (see [Installing and Creating Your First Resource](quickstart.html))
- `pytest` installed for running tests

## Quick Example

```python
import pytest
from ocp_resources.resource import get_client
from ocp_resources.namespace import Namespace

@pytest.fixture(scope="class")
def fake_client():
    return get_client(fake=True)

class TestMyFeature:
    def test_create_namespace(self, fake_client):
        ns = Namespace(client=fake_client, name="my-namespace")
        ns.deploy()

        assert ns.exists
        assert ns.kind == "Namespace"
        assert ns.status == Namespace.Status.ACTIVE

        ns.clean_up(wait=False)
        assert not ns.exists
```

Run it with `pytest` — no cluster, no kubeconfig, no network calls.

## How It Works

Calling `get_client(fake=True)` returns an in-memory client that simulates the Kubernetes API. It supports all standard resources (Pods, Deployments, Services, Namespaces, etc.) and OpenShift resources (Routes, Projects, etc.) out of the box.

All resource classes (`Pod`, `Deployment`, `Namespace`, etc.) accept this fake client through their `client` parameter, and all CRUD methods work identically to a real cluster.

## Step-by-Step: CRUD Operations

### 1. Set up a shared fixture

Create a `conftest.py` in your test directory:

```python
import pytest
from ocp_resources.resource import get_client

@pytest.fixture(scope="class")
def fake_client():
    """Provides a fake Kubernetes client for all tests."""
    return get_client(fake=True)
```

> **Tip:** Use `scope="class"` so resources created in one test method persist for subsequent methods in the same class.

### 2. Create a resource

```python
from ocp_resources.pod import Pod

def test_create_pod(fake_client):
    pod = Pod(
        client=fake_client,
        name="my-pod",
        namespace="default",
        containers=[{"name": "nginx", "image": "nginx:latest"}],
    )
    result = pod.deploy()

    assert result.name == "my-pod"
    assert pod.exists
```

The fake client automatically generates metadata (`uid`, `resourceVersion`, `creationTimestamp`) and a realistic `status` section for the resource.

### 3. Read and query resources

```python
from ocp_resources.namespace import Namespace

def test_list_resources(fake_client):
    # Create some namespaces first
    Namespace(client=fake_client, name="ns-a").deploy()
    Namespace(client=fake_client, name="ns-b").deploy()

    # Iterate over all Namespace resources
    for ns in Namespace.get(client=fake_client):
        assert ns.name
```

Access individual resource data via the `instance` property:

```python
def test_get_instance(fake_client):
    ns = Namespace(client=fake_client, name="test-ns")
    ns.deploy()

    assert ns.instance
    assert ns.kind == "Namespace"
```

### 4. Update a resource

Use `update()` to patch a resource or `update_replace()` to replace it entirely:

```python
def test_update_labels(fake_client):
    ns = Namespace(client=fake_client, name="labeled-ns")
    ns.deploy()

    # Patch: add a label
    resource_dict = ns.instance.to_dict()
    resource_dict["metadata"]["labels"]["env"] = "staging"
    ns.update(resource_dict=resource_dict)

    assert ns.labels["env"] == "staging"
```

```python
def test_replace_labels(fake_client):
    ns = Namespace(client=fake_client, name="replace-ns")
    ns.deploy()

    resource_dict = ns.instance.to_dict()
    resource_dict["metadata"]["labels"] = {"new-label": "value"}
    ns.update_replace(resource_dict=resource_dict)

    assert "new-label" in ns.labels.keys()
```

### 5. Delete a resource

```python
def test_delete_resource(fake_client):
    ns = Namespace(client=fake_client, name="temp-ns")
    ns.deploy()
    assert ns.exists

    ns.clean_up(wait=False)
    assert not ns.exists
```

### 6. Use context managers for automatic cleanup

Resources support Python's `with` statement. The resource deploys on entry and is cleaned up on exit:

```python
from ocp_resources.secret import Secret

def test_context_manager(fake_client):
    with Secret(name="my-secret", namespace="default", client=fake_client) as sec:
        assert sec.exists

    # Automatically cleaned up
    assert not sec.exists
```

## Testing Status and Conditions

The fake client generates realistic status objects so you can test code that reads resource status.

### Checking resource status

```python
def test_namespace_status(fake_client):
    ns = Namespace(client=fake_client, name="status-ns")
    ns.deploy()

    assert ns.status == Namespace.Status.ACTIVE
```

### Waiting for conditions

```python
from ocp_resources.pod import Pod

def test_wait_for_condition(fake_client):
    pod = Pod(
        client=fake_client,
        name="ready-pod",
        namespace="default",
        containers=[{"name": "app", "image": "nginx:latest"}],
    )
    pod.deploy()

    pod.wait_for_status(status=Pod.Status.RUNNING, timeout=5)
```

### Simulating "not ready" resources

Control a resource's readiness state with the `fake-client.io/ready` annotation:

```python
def test_not_ready_pod(fake_client):
    pod = Pod(
        client=fake_client,
        name="failing-pod",
        namespace="default",
        containers=[{"name": "app", "image": "nginx:latest"}],
        annotations={"fake-client.io/ready": "false"},
    )
    pod.deploy()

    # The pod will have Ready condition set to False
    pod.wait_for_condition(
        condition=pod.Condition.READY,
        status=pod.Condition.Status.FALSE,
        timeout=5,
    )

    message = pod.get_condition_message(
        condition_type=pod.Condition.READY,
        condition_status=pod.Condition.Status.FALSE,
    )
    assert message
```

Resource-specific status behavior when marked "not ready":

| Resource | Ready behavior | Not-ready behavior |
|---|---|---|
| Pod | `phase: Running`, containers ready | Containers in `waiting` state |
| Deployment | `readyReplicas` matches `spec.replicas` | `readyReplicas: 0`, `unavailableReplicas` set |
| Namespace | `phase: Active` | `phase: Terminating` |
| All others | `Ready` condition `True` | `Ready` condition `False` |

## Testing with Events

The fake client automatically generates Kubernetes events for create, update, and delete operations. You can query them through the resource's `events()` method:

```python
def test_resource_events(fake_client):
    pod = Pod(
        client=fake_client,
        name="event-pod",
        namespace="default",
        containers=[{"name": "nginx", "image": "nginx:latest"}],
    )
    pod.deploy()

    events = list(pod.events(timeout=1))
    assert events
```

## Testing with Deployments

Deployments get a full status template including replica counts, conditions, and observed generation:

```python
from ocp_resources.deployment import Deployment

def test_deployment(fake_client):
    dep = Deployment(
        client=fake_client,
        name="my-deploy",
        namespace="default",
        replicas=3,
        selector={"matchLabels": {"app": "web"}},
        template={
            "metadata": {"labels": {"app": "web"}},
            "spec": {"containers": [{"name": "web", "image": "nginx:latest"}]},
        },
    )
    dep.deploy()

    assert dep.exists
    assert dep.kind == "Deployment"

    dep.clean_up(wait=False)
    assert not dep.exists
```

## Testing with ResourceList

Test bulk resource creation with `ResourceList` and `NamespacedResourceList`:

```python
from ocp_resources.resource import ResourceList, NamespacedResourceList

def test_resource_list(fake_client):
    with ResourceList(
        client=fake_client,
        resource_class=Namespace,
        name="batch-ns",
        num_resources=3,
    ) as namespaces:
        assert len(namespaces) == 3
    # All three namespaces are automatically cleaned up
```

See [Managing Bulk Resources with ResourceList](managing-resource-lists.html) for more details.

## Advanced Usage

### Using the low-level API directly

You can bypass resource classes and work with the fake client's API directly — useful for testing custom logic:

```python
def test_low_level_api(fake_client):
    api = fake_client.resources.get(api_version="v1", kind="Pod")

    # Create
    pod = api.create(
        body={
            "apiVersion": "v1",
            "kind": "Pod",
            "metadata": {"name": "raw-pod", "namespace": "default"},
            "spec": {"containers": [{"name": "app", "image": "nginx:latest"}]},
        },
        namespace="default",
    )
    assert pod.metadata.name == "raw-pod"
    assert pod.status.phase == "Running"

    # List
    pods = api.get(namespace="default")
    assert len(pods.items) >= 1

    # Patch
    updated = api.patch(
        name="raw-pod",
        namespace="default",
        body={"metadata": {"labels": {"env": "test"}}},
    )
    assert updated.metadata.labels["env"] == "test"

    # Delete
    api.delete(name="raw-pod", namespace="default")
```

### Label and field selectors

The fake client supports filtering resources by labels and fields:

```python
def test_label_selector(fake_client):
    api = fake_client.resources.get(api_version="v1", kind="Pod")

    api.create(body={
        "apiVersion": "v1", "kind": "Pod",
        "metadata": {"name": "pod-a", "namespace": "default", "labels": {"app": "web"}},
        "spec": {"containers": [{"name": "c", "image": "nginx"}]},
    }, namespace="default")

    api.create(body={
        "apiVersion": "v1", "kind": "Pod",
        "metadata": {"name": "pod-b", "namespace": "default", "labels": {"app": "api"}},
        "spec": {"containers": [{"name": "c", "image": "nginx"}]},
    }, namespace="default")

    # Filter by label
    web_pods = api.get(namespace="default", label_selector="app=web")
    assert len(web_pods.items) == 1
    assert web_pods.items[0].metadata.name == "pod-a"
```

Field selectors support `metadata.name` and `metadata.namespace`:

```python
def test_field_selector(fake_client):
    api = fake_client.resources.get(api_version="v1", kind="Pod")
    result = api.get(namespace="default", field_selector="metadata.name=my-pod")
```

### Watch for resource changes

The `watch()` method yields existing resources as `ADDED` events:

```python
def test_watch(fake_client):
    api = fake_client.resources.get(api_version="v1", kind="Pod")

    api.create(body={
        "apiVersion": "v1", "kind": "Pod",
        "metadata": {"name": "watch-pod", "namespace": "default"},
        "spec": {"containers": [{"name": "c", "image": "nginx"}]},
    }, namespace="default")

    for event in api.watch(namespace="default"):
        assert event["type"] == "ADDED"
        assert event["object"].metadata.name
        break  # Stop after first event
```

### Registering Custom Resource Definitions

To test with CRDs not bundled in the default schema, register them on the client:

```python
def test_custom_resource(fake_client):
    # Register the CRD
    fake_client.register_resources({
        "kind": "MyApp",
        "api_version": "v1alpha1",
        "group": "apps.example.com",
        "namespaced": True,
    })

    # Use it through the low-level API
    api = fake_client.resources.get(
        api_version="apps.example.com/v1alpha1",
        kind="MyApp",
    )

    created = api.create(
        body={
            "apiVersion": "apps.example.com/v1alpha1",
            "kind": "MyApp",
            "metadata": {"name": "my-app", "namespace": "default"},
            "spec": {"replicas": 2},
        },
        namespace="default",
    )
    assert created.metadata.name == "my-app"
```

You can also register multiple CRDs at once:

```python
fake_client.register_resources([
    {"kind": "MyApp", "api_version": "v1", "group": "apps.example.com"},
    {"kind": "MyConfig", "api_version": "v1beta1", "group": "config.example.com", "namespaced": False},
])
```

### Testing incremental workflows

Use `@pytest.mark.incremental` to test a resource through its full lifecycle. Each test depends on the previous one succeeding:

```python
import pytest
from ocp_resources.pod import Pod

@pytest.mark.incremental
class TestPodLifecycle:
    @pytest.fixture(scope="class")
    def pod(self, fake_client):
        return Pod(
            client=fake_client,
            name="lifecycle-pod",
            namespace="default",
            containers=[{"name": "nginx", "image": "nginx:latest"}],
        )

    def test_01_create(self, pod):
        deployed = pod.deploy()
        assert deployed.name == "lifecycle-pod"
        assert pod.exists

    def test_02_read(self, pod):
        assert pod.instance
        assert pod.kind == "Pod"

    def test_03_update(self, pod):
        resource_dict = pod.instance.to_dict()
        resource_dict["metadata"]["labels"] = {"updated": "true"}
        pod.update(resource_dict=resource_dict)
        assert pod.labels["updated"] == "true"

    def test_04_delete(self, pod):
        pod.clean_up(wait=False)
        assert not pod.exists
```

> **Note:** The `@pytest.mark.incremental` marker requires custom `conftest.py` hooks. Add the following to your `conftest.py`:
> ```python
> def pytest_runtest_makereport(item, call):
>     if call.excinfo is not None and "incremental" in item.keywords:
>         parent = item.parent
>         parent._previousfailed = item
>
> def pytest_runtest_setup(item):
>     if "incremental" in item.keywords:
>         previousfailed = getattr(item.parent, "_previousfailed", None)
>         if previousfailed is not None:
>             pytest.xfail(f"previous test failed ({previousfailed.name})")
> ```

### Conflict detection

The fake client raises errors for duplicate creates and missing resources, just like a real cluster:

```python
from fake_kubernetes_client import NotFoundError, ConflictError

def test_conflict_on_duplicate_create(fake_client):
    api = fake_client.resources.get(api_version="v1", kind="Namespace")

    api.create(body={"metadata": {"name": "unique-ns"}})

    with pytest.raises(ConflictError):
        api.create(body={"metadata": {"name": "unique-ns"}})

def test_not_found_on_missing(fake_client):
    api = fake_client.resources.get(api_version="v1", kind="Pod")

    with pytest.raises(NotFoundError):
        api.get(name="does-not-exist", namespace="default")
```

### Accessing response data

The fake client returns response objects that support both attribute access and dictionary access:

```python
def test_response_access(fake_client):
    api = fake_client.resources.get(api_version="v1", kind="Pod")

    created = api.create(
        body={
            "apiVersion": "v1", "kind": "Pod",
            "metadata": {"name": "access-pod", "namespace": "default"},
            "spec": {"containers": [{"name": "c", "image": "nginx"}]},
        },
        namespace="default",
    )

    # Attribute access
    assert created.metadata.name == "access-pod"
    assert created.status.phase == "Running"

    # Dictionary-style access
    assert created["metadata"]["name"] == "access-pod"

    # Convert to a plain dict
    data = created.to_dict()
    assert isinstance(data, dict)
    assert data["metadata"]["name"] == "access-pod"
```

## Troubleshooting

**`ResourceNotFoundError` when creating a resource**

The fake client must recognize the resource's API group and kind. Standard Kubernetes and OpenShift resources are pre-registered. For CRDs, call `fake_client.register_resources(...)` before using them.

**Status fields are missing or empty**

Pods, Deployments, Services, and Namespaces have hardcoded status templates. Other resources get a generic status with a `Ready` condition. If the resource has a schema in the bundled OpenAPI definitions, a richer status is generated automatically.

**Tests interfere with each other**

Each call to `get_client(fake=True)` creates an independent in-memory store. Use separate fixtures or narrower fixture scopes to isolate tests that share a client.

## Related Pages

- [Connecting to Clusters](connecting-to-clusters.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Querying and Watching Resources](querying-resources.html)
- [Managing Bulk Resources with ResourceList](managing-resource-lists.html)
- [Common Resource Patterns](common-patterns.html)

---

Source: generating-resource-classes.md

Let me start by exploring the repository structure and the knowledge graph.

Now let me check the generated output example and the test generation flow:

Now let me look at a generated file with a spec section, and check one with duplicate kinds:

Let me also check the `--help` output structure and look at a few more details:

Now let me check a dry-run example and verify what shell completion setup looks like:

Now I have a thorough understanding of the codebase. Let me write the documentation:

# Generating New Resource Classes with class-generator

Scaffold Python wrapper classes for any Kubernetes or OpenShift custom resource definition (CRD) so you can manage those resources using the same patterns as built-in resources. The `class-generator` CLI introspects your cluster's OpenAPI schemas and produces ready-to-use Python files complete with typed constructors, docstrings, and `to_dict()` serialization.

## Prerequisites

- **openshift-python-wrapper** installed (see [Installing and Creating Your First Resource](quickstart.html))
- `oc` or `kubectl` CLI available on your `PATH`
- An active connection to a Kubernetes or OpenShift cluster with admin privileges (for `--kind` generation and schema updates)

## Quick Example

Generate a wrapper class for the `Deployment` resource:

```bash
class-generator -k Deployment
```

This creates a Python file in `ocp_resources/` with a fully typed class you can import and use immediately:

```python
from ocp_resources.deployment import Deployment

dep = Deployment(
    name="my-app",
    namespace="default",
    replicas=3,
    selector={"matchLabels": {"app": "my-app"}},
    template={...},
)
dep.deploy()
```

## Step-by-Step: Generating a Resource Class

### 1. Generate the class file

Pass the CRD's `Kind` (case-sensitive) to `--kind` / `-k`:

```bash
class-generator -k StorageCluster
```

The output file is automatically named using snake_case — `ocp_resources/storage_cluster.py`.

### 2. Preview without writing (dry run)

See exactly what would be generated without touching the filesystem:

```bash
class-generator -k StorageCluster --dry-run
```

The rendered Python code is printed to the console with syntax highlighting.

### 3. Review the generated file

Every generated file contains:

- A class inheriting from `Resource` (cluster-scoped) or `NamespacedResource` (namespace-scoped)
- Typed `__init__` parameters extracted from the OpenAPI spec and `status` fields
- A `to_dict()` method that serializes required and optional fields
- A `# End of generated code` marker — any code you add **below** this line is preserved on regeneration

Example generated output for `ConfigMap`:

```python
from typing import Any
from ocp_resources.resource import NamespacedResource

class ConfigMap(NamespacedResource):
    """
    ConfigMap holds configuration data for pods to consume.
    """

    api_version: str = NamespacedResource.ApiVersion.V1

    def __init__(
        self,
        binary_data: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
        immutable: bool | None = None,
        **kwargs: Any,
    ) -> None:
        ...
```

### 4. Generate multiple kinds at once

Pass a comma-separated list (no spaces):

```bash
class-generator -k Pod,Service,ConfigMap
```

Multiple kinds are generated in parallel for speed.

### 5. Overwrite an existing file

By default, existing files are not overwritten. Pass `--overwrite` to replace them:

```bash
class-generator -k Deployment --overwrite
```

To create a timestamped backup before overwriting:

```bash
class-generator -k Deployment --overwrite --backup
```

Backups are stored in `.backups/backup-YYYYMMDD-HHMMSS/` preserving the original directory structure.

### 6. Write to a custom output path

```bash
class-generator -k MyCustomResource -o my_package/custom_resource.py
```

## Handling Duplicate Kinds

Some Kubernetes kinds exist in multiple API groups. For example, `DNS` exists in both `config.openshift.io` and `operator.openshift.io`. When this happens, the generator automatically creates separate files with group-based suffixes:

| Kind | API Group | Generated File |
|------|-----------|----------------|
| `DNS` | `config.openshift.io` | `dns_config_openshift_io.py` |
| `DNS` | `operator.openshift.io` | `dns_operator_openshift_io.py` |

Both files define a class named `DNS` but with different `api_group` attributes. Import the one matching the API group you need.

## Adding User Code to Generated Files

Any code you add **below** the `# End of generated code` marker is preserved when you regenerate the file with `--overwrite`. This is how built-in resources like `ConfigMap` include custom properties:

```python
    # End of generated code

    @property
    def keys_to_hash(self):
        return ["data", "binaryData"]
```

Custom imports you add at the top of the file are also preserved during regeneration.

## Updating Schemas

The generator uses a local schema cache to resolve resource definitions. If a CRD is missing (e.g., you just installed a new operator), update the cache first.

### Full schema update

Fetch all schemas from the connected cluster:

```bash
class-generator --update-schema
```

> **Note:** If connected to an older cluster, existing schemas are preserved. Only new or missing resources are added.

### Single-resource schema update

Update just one resource without touching the rest:

```bash
class-generator --update-schema-for LlamaStackDistribution
```

Then generate the class:

```bash
class-generator -k LlamaStackDistribution --overwrite
```

> **Warning:** `--update-schema` and `--update-schema-for` are mutually exclusive. Use one or the other.

## Discovering Missing Resources

Find resources on your cluster that don't have wrapper classes yet:

```bash
class-generator --discover-missing
```

### Coverage report

Get a summary of how many resources have wrapper classes:

```bash
class-generator --coverage-report
```

Output in JSON for CI/CD pipelines:

```bash
class-generator --coverage-report --json
```

### Auto-generate all missing resources

Update schemas and generate classes for every resource that's missing:

```bash
class-generator --update-schema --generate-missing
```

> **Tip:** Use `--dry-run` with `--generate-missing` to preview what would be generated without writing any files.

## Advanced Usage

### Batch regeneration

Regenerate all existing generated resource classes to pick up schema changes:

```bash
class-generator --regenerate-all
```

With backups:

```bash
class-generator --regenerate-all --backup
```

Filter to a subset of resources using glob patterns:

```bash
class-generator --regenerate-all --filter "Pod*"
class-generator --regenerate-all --filter "*Service"
```

### Adding tests

Generate test files alongside the resource class:

```bash
class-generator -k Pod --add-tests
```

This creates test manifests and runs them automatically with `pytest`. Tests are placed under the test manifests directory and validate that the generated class can be parsed correctly.

### Shell completion

Add this to your `~/.bashrc` or `~/.zshrc` for tab completion:

```bash
# For zsh
if type class-generator > /dev/null; then eval "$(_CLASS_GENERATOR_COMPLETE=zsh_source class-generator)"; fi

# For bash
if type class-generator > /dev/null; then eval "$(_CLASS_GENERATOR_COMPLETE=bash_source class-generator)"; fi
```

### Verbose output

Enable debug logging to see exactly what the generator is doing:

```bash
class-generator -k Deployment -v
```

### When a kind is missing from the schema

If the kind you request is not in the local schema cache, the CLI prompts you:

```
deployment not found in schema mapping, Do you want to run --update-schema and retry? [Y/N]:
```

Answering `y` triggers a schema update and retries generation automatically.

> **Note:** The prompt only appears in interactive mode. When generating multiple kinds or running in batch mode, missing kinds are skipped with a warning. Run `--update-schema` separately beforehand in CI environments.

### API group and version warnings

If a generated resource uses an API group or version that hasn't been registered in the base `Resource` class, the generator logs a warning like:

```
Missing API Group in Resource
Please add `Resource.ApiGroup.MY_GROUP_IO = my-group.io` manually into
ocp_resources/resource.py under Resource class > ApiGroup class.
```

Follow the instructions to register the API group so the resource can connect to the correct API endpoint. See [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html) for details on how `ApiGroup` and `ApiVersion` are used.

## CLI Options Reference

| Option | Description |
|--------|-------------|
| `-k`, `--kind` | Kind(s) to generate (comma-separated) |
| `-o`, `--output-file` | Custom output file path |
| `--overwrite` | Overwrite existing files |
| `--dry-run` | Preview output without writing files |
| `--add-tests` | Generate test files for the kind |
| `--update-schema` | Update all schema files from cluster |
| `--update-schema-for` | Update schema for a single kind |
| `--discover-missing` | Find resources without wrapper classes |
| `--coverage-report` | Show coverage statistics |
| `--json` | Output reports in JSON format |
| `--generate-missing` | Generate classes for all missing resources |
| `--regenerate-all` | Regenerate all existing generated classes |
| `--backup` | Create backup before overwriting/regenerating |
| `--filter` | Glob pattern to filter `--regenerate-all` |
| `-v`, `--verbose` | Enable debug logging |

For the full CLI reference, see [class-generator CLI Reference](class-generator-cli.html).

## Troubleshooting

**"Neither 'oc' nor 'kubectl' binary found in PATH"**
Install `oc` from the [OpenShift mirror](https://mirror.openshift.com/pub/openshift-v4/x86_64/clients/ocp/stable/) or `kubectl` from the [Kubernetes docs](https://kubernetes.io/docs/tasks/tools/). Ensure the binary is on your `PATH`.

**"Resource kind 'X' not found"**
The kind is not in the local schema cache. Run `class-generator --update-schema` while connected to a cluster that has the CRD installed, then retry.

**Generated filename looks wrong (e.g., `c_d_i_config.py`)**
Single-letter segments in snake_case names indicate the camelCase-to-snake_case converter doesn't recognize an acronym. The generator raises an error with instructions to add the acronym to the converter.

**"Failed to fetch OpenAPI v3 index"**
Ensure you are logged in to the cluster with sufficient privileges. The generator needs to access the `/openapi/v3` endpoint, which typically requires cluster-admin access.

## Related Pages

- [class-generator CLI Reference](class-generator-cli.html)
- [Schema Validation and Code Generation Architecture](schema-validation-internals.html)
- [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html)
- [Resource and NamespacedResource API](resource-api.html)
- [Validating Resources Against OpenAPI Schemas](validating-resources.html)

---

Source: common-patterns.md

Let me start by reading the knowledge graph and the pages manifest, then explore the codebase.

Now let me look at more resource files for networking, RBAC, and DataVolume:

Now let me look at the secret, network policy, network attachment definition, and the duplicate API groups handling:

Now let me look at the duplicate API groups handling:

Now let me look at some files with duplicate kinds to understand the pattern:

Now let me look at how to exec in pods and also look at the user_defined_network and apply_yaml approach:

Now let me look at how the `yaml_file` parameter works and look at the event-related code:

Now let me check some gateway files to see the pattern for duplicate API groups:

Now let me look at the `user_defined_network.py` for newer networking resources:

Now I have enough information. Let me write the documentation page:

# Common Resource Patterns

Copy-paste recipes for the most common Kubernetes and OpenShift resource operations using openshift-python-wrapper.

> **Note:** All recipes assume you already have a connected client. See [Connecting to Clusters](connecting-to-clusters.html) for setup instructions.

```python
# Preamble used by all recipes below
from ocp_resources.resource import get_client
client = get_client()
```

---

## Create and Run a Pod

Create a pod with a single container and wait for it to start.

```python
from ocp_resources.pod import Pod

pod = Pod(
    client=client,
    name="my-nginx",
    namespace="default",
    containers=[{
        "name": "nginx",
        "image": "nginx:1.25",
        "ports": [{"containerPort": 80}],
    }],
    restart_policy="Never",
)
pod.deploy()
pod.wait_for_status(status=Pod.Status.RUNNING, timeout=120)
```

The `deploy()` method creates the pod on the cluster. `wait_for_status` polls until the pod reaches `Running`. Use `restart_policy="Always"` for long-running pods.

---

## Create a Pod as a Context Manager

Automatically clean up a pod when the block exits — ideal for tests.

```python
from ocp_resources.pod import Pod

with Pod(
    client=client,
    name="test-curl",
    namespace="default",
    containers=[{
        "name": "curl",
        "image": "curlimages/curl:8.5.0",
        "command": ["sleep", "3600"],
    }],
) as pod:
    pod.wait_for_status(status=Pod.Status.RUNNING, timeout=60)
    output = pod.execute(command=["curl", "-s", "http://httpbin.org/get"])
    print(output)
# Pod is automatically deleted here
```

The context manager calls `deploy()` on enter and `clean_up()` on exit. Set `teardown=False` to skip automatic deletion.

> **Tip:** See [Creating and Managing Resources](creating-and-managing-resources.html) for all lifecycle management patterns.

---

## Execute a Command Inside a Pod

Run a shell command inside an already-running pod and capture the output.

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-nginx", namespace="default", ensure_exists=True)
output = pod.execute(command=["cat", "/etc/nginx/nginx.conf"], timeout=30)
print(output)
```

`execute()` streams output via the Kubernetes exec API. Use `container="sidecar"` to target a specific container in multi-container pods. Set `ignore_rc=True` to suppress errors from non-zero exit codes.

---

## Get Pod Logs

Retrieve logs from a running or completed pod.

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-nginx", namespace="default", ensure_exists=True)

# Full logs
print(pod.log())

# Last 50 lines
print(pod.log(tail_lines=50))

# Logs from a specific container
print(pod.log(container="sidecar"))
```

The `log()` method passes kwargs through to the Kubernetes `read_namespaced_pod_log` API, so all standard parameters like `tail_lines`, `since_seconds`, and `container` are supported.

---

## Create a Deployment and Wait for Replicas

Deploy an application with multiple replicas and wait until they are all ready.

```python
from ocp_resources.deployment import Deployment

dep = Deployment(
    client=client,
    name="web-app",
    namespace="default",
    replicas=3,
    selector={"matchLabels": {"app": "web-app"}},
    template={
        "metadata": {"labels": {"app": "web-app"}},
        "spec": {
            "containers": [{
                "name": "app",
                "image": "nginx:1.25",
                "ports": [{"containerPort": 80}],
            }]
        },
    },
)
dep.deploy()
dep.wait_for_replicas(deployed=True, timeout=300)
```

`wait_for_replicas` polls until `availableReplicas == readyReplicas == updatedReplicas == spec.replicas`. Pass `deployed=False` to wait for all replicas to be scaled down.

---

## Scale a Deployment

Change the replica count of an existing deployment.

```python
from ocp_resources.deployment import Deployment

dep = Deployment(client=client, name="web-app", namespace="default", ensure_exists=True)

# Scale up
dep.scale_replicas(replica_count=5)
dep.wait_for_replicas(deployed=True, timeout=300)

# Scale down
dep.scale_replicas(replica_count=1)
dep.wait_for_replicas(deployed=True, timeout=300)
```

`scale_replicas` patches the deployment's `spec.replicas` field. Always follow with `wait_for_replicas` to confirm the rollout completes.

---

## Create a Service

Expose a deployment via a ClusterIP service.

```python
from ocp_resources.service import Service

svc = Service(
    client=client,
    name="web-app-svc",
    namespace="default",
    selector={"app": "web-app"},
    ports=[{
        "protocol": "TCP",
        "port": 80,
        "targetPort": 80,
    }],
    type="ClusterIP",
)
svc.deploy()
```

Change `type` to `"NodePort"` or `"LoadBalancer"` as needed. The `selector` must match labels on your target pods.

---

## Create an OpenShift Route

Expose a service externally via an OpenShift Route.

```python
from ocp_resources.route import Route

route = Route(
    client=client,
    name="web-app-route",
    namespace="default",
    service="web-app-svc",
)
route.deploy()

# Get the assigned hostname
print(route.host)
```

For TLS re-encrypt routes, pass `destination_ca_cert="<PEM certificate string>"`. Access the exposed service name with `route.exposed_service` and the TLS termination type with `route.termination`.

---

## Create a Namespace

Create a namespace (or use as a context manager for automatic cleanup).

```python
from ocp_resources.namespace import Namespace

# Simple creation
ns = Namespace(client=client, name="test-ns")
ns.deploy()

# As a context manager (deleted on exit)
with Namespace(client=client, name="ephemeral-ns") as ns:
    # ... do work in the namespace ...
    pass
```

`Namespace` extends `Resource` (cluster-scoped), so no `namespace` parameter is needed.

---

## Create a ConfigMap

Store configuration data for pods to consume.

```python
from ocp_resources.config_map import ConfigMap

cm = ConfigMap(
    client=client,
    name="app-config",
    namespace="default",
    data={
        "DATABASE_URL": "postgres://db:5432/myapp",
        "LOG_LEVEL": "info",
    },
)
cm.deploy()
```

Use `binary_data` for non-UTF-8 content. Set `immutable=True` to prevent changes after creation.

---

## Create a Secret

Store sensitive data such as credentials.

```python
from ocp_resources.secret import Secret

secret = Secret(
    client=client,
    name="db-credentials",
    namespace="default",
    string_data={
        "username": "admin",
        "password": "s3cur3-pa$$word",
    },
    type="Opaque",
)
secret.deploy()
```

Use `data_dict` instead of `string_data` if your values are already base64-encoded. Secret data is automatically hashed in log output for security.

---

## Create a Job

Run a one-off task to completion.

```python
from ocp_resources.job import Job

job = Job(
    client=client,
    name="db-migration",
    namespace="default",
    backoff_limit=3,
    restart_policy="Never",
    containers=[{
        "name": "migrate",
        "image": "my-app:latest",
        "command": ["python", "manage.py", "migrate"],
    }],
)
job.deploy()
job.wait_for_condition(
    condition=Job.Condition.COMPLETE,
    status=Job.Condition.Status.TRUE,
    timeout=300,
)
```

Set `background_propagation_policy="Background"` to delete leftover pods when the job is cleaned up. The `backoff_limit` controls how many times the job retries on failure.

---

## Create a PersistentVolumeClaim

Request persistent storage for your workloads.

```python
from ocp_resources.persistent_volume_claim import PersistentVolumeClaim

pvc = PersistentVolumeClaim(
    client=client,
    name="app-data",
    namespace="default",
    accessmodes=PersistentVolumeClaim.AccessMode.RWO,
    size="10Gi",
    storage_class="gp3-csi",
    volume_mode=PersistentVolumeClaim.VolumeMode.FILE,
)
pvc.deploy()
pvc.wait_for_status(status=PersistentVolumeClaim.Status.BOUND, timeout=120)
```

Use the `AccessMode` and `VolumeMode` constants instead of raw strings. The `storage_class` must match an available StorageClass on your cluster.

---

## Create a DataVolume (KubeVirt / CDI)

Import a VM disk image into a PVC using the Containerized Data Importer.

```python
from ocp_resources.datavolume import DataVolume

dv = DataVolume(
    client=client,
    name="fedora-disk",
    namespace="default",
    source_dict={"http": {"url": "https://download.fedoraproject.org/pub/fedora/linux/releases/40/Cloud/x86_64/images/Fedora-Cloud-Base-40-1.14.x86_64.qcow2"}},
    size="30Gi",
    storage_class="ocs-storagecluster-ceph-rbd-virtualization",
    access_modes=DataVolume.AccessMode.RWX,
    volume_mode=DataVolume.VolumeMode.BLOCK,
    api_name="storage",
)
dv.deploy()
dv.wait_for_dv_success(timeout=600)
```

Always pass `api_name="storage"` explicitly — the default will change in a future release. Use `source_dict={"blank": {}}` for empty disks, or `source_dict={"pvc": {"name": "source-pvc", "namespace": "default"}}` for cloning.

- **Clone a DataVolume:**
  ```python
  dv_clone = DataVolume(
      client=client,
      name="fedora-disk-clone",
      namespace="default",
      source_dict={"pvc": {"name": "fedora-disk", "namespace": "default"}},
      size="30Gi",
      storage_class="ocs-storagecluster-ceph-rbd-virtualization",
      access_modes=DataVolume.AccessMode.RWX,
      volume_mode=DataVolume.VolumeMode.BLOCK,
      api_name="storage",
  )
  ```

> **Tip:** See [Working with Virtual Machines (KubeVirt)](working-with-virtual-machines.html) for full VM lifecycle recipes.

---

## Create a NetworkPolicy

Restrict traffic to pods matching a label selector.

```python
from ocp_resources.network_policy import NetworkPolicy

netpol = NetworkPolicy(
    client=client,
    name="allow-http-only",
    namespace="default",
    pod_selector={"matchLabels": {"app": "web-app"}},
    policy_types=["Ingress"],
    ingress=[{
        "ports": [{"protocol": "TCP", "port": 80}],
        "from": [
            {"namespaceSelector": {"matchLabels": {"env": "production"}}},
        ],
    }],
)
netpol.deploy()
```

This allows TCP port 80 ingress only from namespaces labeled `env=production`. Omit `ingress` and set `policy_types=["Ingress"]` to deny all inbound traffic.

---

## Create a NetworkAttachmentDefinition

Attach pods to a secondary network using Multus.

```python
from ocp_resources.network_attachment_definition import LinuxBridgeNetworkAttachmentDefinition

nad = LinuxBridgeNetworkAttachmentDefinition(
    client=client,
    name="br-100",
    namespace="default",
    bridge_name="br-100",
    cni_type="cnv-bridge",
    vlan=100,
    mtu=1500,
)
nad.deploy()
```

Variant classes are available for different bridge types:
- `LinuxBridgeNetworkAttachmentDefinition` — Linux bridge (cnv-bridge)
- `OvsBridgeNetworkAttachmentDefinition` — OVS bridge
- `OVNOverlayNetworkAttachmentDefinition` — OVN overlay (layer 2/3)

---

## Create an RBAC Role and RoleBinding

Grant specific permissions to a service account within a namespace.

```python
from ocp_resources.role import Role
from ocp_resources.role_binding import RoleBinding
from ocp_resources.service_account import ServiceAccount

sa = ServiceAccount(
    client=client,
    name="app-sa",
    namespace="default",
)
sa.deploy()

role = Role(
    client=client,
    name="pod-reader",
    namespace="default",
    rules=[{
        "apiGroups": [""],
        "resources": ["pods", "pods/log"],
        "verbs": ["get", "list", "watch"],
    }],
)
role.deploy()

binding = RoleBinding(
    client=client,
    name="app-sa-pod-reader",
    namespace="default",
    subjects_kind="ServiceAccount",
    subjects_name="app-sa",
    subjects_namespace="default",
    role_ref_kind="Role",
    role_ref_name="pod-reader",
)
binding.deploy()
```

This grants the `app-sa` service account read-only access to pods in the `default` namespace.

---

## Create a ClusterRole and ClusterRoleBinding

Grant cluster-wide permissions to a service account.

```python
from ocp_resources.cluster_role import ClusterRole
from ocp_resources.cluster_role_binding import ClusterRoleBinding

cr = ClusterRole(
    client=client,
    name="node-viewer",
    rules=[{
        "apiGroups": [""],
        "resources": ["nodes"],
        "verbs": ["get", "list", "watch"],
    }],
)
cr.deploy()

crb = ClusterRoleBinding(
    client=client,
    name="app-sa-node-viewer",
    cluster_role="node-viewer",
    subjects=[{
        "kind": "ServiceAccount",
        "name": "app-sa",
        "namespace": "default",
    }],
)
crb.deploy()
```

`ClusterRole` and `ClusterRoleBinding` are cluster-scoped (`Resource` subclasses), so no `namespace` parameter is needed on the role or binding itself.

---

## List and Filter Resources

Query existing resources using label and field selectors.

```python
from ocp_resources.pod import Pod
from ocp_resources.namespace import Namespace

# List all pods in a namespace
for pod in Pod.get(client=client, namespace="default"):
    print(f"{pod.name} — {pod.status}")

# Filter by label
for pod in Pod.get(client=client, namespace="default", label_selector="app=web-app"):
    print(pod.name)

# Filter by field
for pod in Pod.get(client=client, namespace="default", field_selector="status.phase=Running"):
    print(pod.name)

# List cluster-scoped resources (no namespace)
for ns in Namespace.get(client=client, label_selector="env=staging"):
    print(ns.name)
```

The `get()` class method returns a generator of resource objects. Pass `raw=True` to get raw `ResourceInstance` objects instead.

> **Tip:** See [Querying and Watching Resources](querying-resources.html) for advanced querying patterns.

---

## Create a Resource from a YAML File

Load a resource definition from an existing YAML file.

```python
from ocp_resources.pod import Pod

pod = Pod(
    client=client,
    yaml_file="manifests/my-pod.yaml",
)
pod.deploy()
```

When using `yaml_file`, the `name` and `namespace` are read from the file automatically. You cannot combine `yaml_file` with `kind_dict`.

---

## Create a Resource from a Dictionary

Pass a raw Kubernetes resource dict directly.

```python
from ocp_resources.deployment import Deployment

dep = Deployment(
    client=client,
    kind_dict={
        "apiVersion": "apps/v1",
        "kind": "Deployment",
        "metadata": {
            "name": "from-dict-app",
            "namespace": "default",
        },
        "spec": {
            "replicas": 2,
            "selector": {"matchLabels": {"app": "from-dict"}},
            "template": {
                "metadata": {"labels": {"app": "from-dict"}},
                "spec": {
                    "containers": [{
                        "name": "app",
                        "image": "nginx:1.25",
                    }],
                },
            },
        },
    },
)
dep.deploy()
```

Using `kind_dict` bypasses all constructor logic in `to_dict()` — the dictionary is sent as-is to the API. This is useful when migrating existing manifests.

---

## Handle Resources with Duplicate API Groups

Some Kubernetes resource kinds (e.g., `DNS`, `Ingress`, `Gateway`) exist in multiple API groups. The wrapper provides separate modules for each variant.

```python
# DNS from config.openshift.io (cluster DNS config)
from ocp_resources.dns_config_openshift_io import DNS as DNSConfig

dns_config = DNSConfig(client=client, name="cluster", ensure_exists=True)
print(dns_config.instance.spec.baseDomain)

# DNS from operator.openshift.io (CoreDNS operator)
from ocp_resources.dns_operator_openshift_io import DNS as DNSOperator

dns_operator = DNSOperator(client=client, name="default", ensure_exists=True)
print(dns_operator.instance.spec.logLevel)
```

```python
# Ingress from config.openshift.io (cluster ingress config)
from ocp_resources.ingress_config_openshift_io import Ingress as IngressConfig

ingress = IngressConfig(client=client, name="cluster", ensure_exists=True)
print(ingress.instance.spec.domain)

# Ingress from networking.k8s.io (K8s Ingress rules)
from ocp_resources.ingress_networking_k8s_io import Ingress as K8sIngress

k8s_ingress = K8sIngress(
    client=client,
    name="my-app-ingress",
    rules=[{
        "host": "app.example.com",
        "http": {
            "paths": [{
                "path": "/",
                "pathType": "Prefix",
                "backend": {"service": {"name": "web-app-svc", "port": {"number": 80}}},
            }],
        },
    }],
)
```

The naming convention for duplicate-kind modules is `<kind>_<api_group_snake_case>.py`. Use Python import aliases (`as`) to distinguish them in your code.

**Common duplicate kinds and their modules:**

| Kind | Module | API Group |
|------|--------|-----------|
| `DNS` | `dns_config_openshift_io` | `config.openshift.io` |
| `DNS` | `dns_operator_openshift_io` | `operator.openshift.io` |
| `Ingress` | `ingress_config_openshift_io` | `config.openshift.io` |
| `Ingress` | `ingress_networking_k8s_io` | `networking.k8s.io` |
| `Gateway` | `gateway` | `networking.istio.io` |
| `Gateway` | `gateway_gateway_networking_k8s_io` | `gateway.networking.k8s.io` |
| `Gateway` | `gateway_networking_istio_io` | `networking.istio.io` |

> **Warning:** Importing the wrong module will target the wrong API group, causing `NotFoundError` or unexpected behavior. Always verify the `api_group` attribute on your resource class.

---

## Manage Multiple Similar Resources with ResourceList

Create N copies of a resource with auto-numbered names.

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import ResourceList

with ResourceList(
    resource_class=Namespace,
    num_resources=3,
    client=client,
    name="perf-test-ns",
) as namespaces:
    # Creates: perf-test-ns-1, perf-test-ns-2, perf-test-ns-3
    for ns in namespaces:
        print(ns.name)
# All 3 namespaces are deleted on exit
```

> **Tip:** See [Managing Bulk Resources with ResourceList](managing-resource-lists.html) for `NamespacedResourceList` and other bulk patterns.

---

## Deploy One Resource Per Namespace with NamespacedResourceList

Create an identical namespaced resource in each of several namespaces.

```python
from ocp_resources.namespace import Namespace
from ocp_resources.config_map import ConfigMap
from ocp_resources.resource import ResourceList, NamespacedResourceList

with ResourceList(
    resource_class=Namespace,
    num_resources=3,
    client=client,
    name="team-ns",
) as namespaces:
    with NamespacedResourceList(
        resource_class=ConfigMap,
        namespaces=namespaces,
        client=client,
        name="shared-config",
        data={"REGION": "us-east-1"},
    ) as configmaps:
        # One ConfigMap "shared-config" in each of team-ns-1, team-ns-2, team-ns-3
        for cm in configmaps:
            print(f"{cm.namespace}/{cm.name}")
```

`NamespacedResourceList` requires a `ResourceList` of `Namespace` objects. All resources are cleaned up in reverse order on exit.

---

## Temporarily Edit a Resource with ResourceEditor

Apply temporary patches during a test and automatically restore original values.

```python
from ocp_resources.resource import ResourceEditor
from ocp_resources.config_map import ConfigMap

cm = ConfigMap(client=client, name="app-config", namespace="default", ensure_exists=True)

with ResourceEditor(
    patches={cm: {"data": {"LOG_LEVEL": "debug"}}},
    action="update",
):
    # LOG_LEVEL is now "debug"
    print(cm.instance.data.LOG_LEVEL)
# Original LOG_LEVEL is restored here
```

`ResourceEditor` backs up original values on enter and restores them on exit. Use `action="replace"` when you need to remove fields entirely rather than patch them.

> **Tip:** See [Editing Resources Temporarily with ResourceEditor](editing-resources-temporarily.html) for complete details.

---

## Validate a Resource Before Creating It

Catch schema errors before submitting to the API server.

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ValidationError

pod = Pod(
    client=client,
    name="validated-pod",
    namespace="default",
    containers=[{
        "name": "app",
        "image": "nginx:1.25",
    }],
    schema_validation_enabled=True,  # Auto-validate on create()
)

# Manual validation
try:
    pod.validate()
    print("Resource is valid")
except ValidationError as e:
    print(f"Validation failed: {e}")

# Or validate a raw dict without instantiation
try:
    Pod.validate_dict({
        "apiVersion": "v1",
        "kind": "Pod",
        "metadata": {"name": "test"},
        "spec": {"containers": [{"name": "app", "image": "nginx"}]},
    })
except ValidationError as e:
    print(f"Dict validation failed: {e}")
```

When `schema_validation_enabled=True`, `create()` and `update_replace()` automatically validate before sending to the API. Validation requires OpenAPI schemas to be available.

> **Tip:** See [Validating Resources Against OpenAPI Schemas](validating-resources.html) for schema setup and troubleshooting.

---

## Check If a Resource Exists

Verify a resource exists before operating on it.

```python
from ocp_resources.deployment import Deployment

dep = Deployment(client=client, name="web-app", namespace="default")

if dep.exists:
    print(f"Deployment exists, status: {dep.instance.status.availableReplicas} replicas available")
else:
    print("Deployment not found")
```

`exists` returns the resource instance if found or `None` if not. For fail-fast behavior, use `ensure_exists=True` in the constructor — it raises `ResourceNotFoundError` immediately if the resource is missing.

---

## Watch Resource Events

Stream events related to a specific resource.

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-nginx", namespace="default", ensure_exists=True)

# Watch events for this pod (10 second window)
for event in pod.events(timeout=10):
    ev = event["object"]
    print(f"[{ev.type}] {ev.reason}: {ev.message}")
```

Use `field_selector` to narrow results further, for example: `field_selector="type==Warning"`. The `events()` method automatically filters by `involvedObject.name`.

---

## List Recent Events in a Namespace

Get existing events (not a watch stream) from the last N seconds.

```python
from ocp_resources.event import Event

# Warning events from the last 5 minutes
events = Event.list(
    client=client,
    namespace="default",
    field_selector="type==Warning",
    since_seconds=300,
)
for ev in events:
    print(f"[{ev.reason}] {ev.message}")
```

`Event.list()` returns a sorted list (most recent first), unlike `Event.get()` which is a live watch stream.

---

## Wait for a Resource Condition

Block until a resource reaches a specific condition.

```python
from ocp_resources.deployment import Deployment

dep = Deployment(client=client, name="web-app", namespace="default", ensure_exists=True)

dep.wait_for_condition(
    condition="Available",
    status="True",
    timeout=300,
)
```

Use `stop_condition` to fail fast if an unrecoverable condition is detected:

```python
dep.wait_for_condition(
    condition="Available",
    status="True",
    timeout=300,
    stop_condition="ReplicaFailure",
    stop_status="True",
)
```

> **Tip:** See [Waiting for Resource Conditions and Status](waiting-for-conditions.html) for advanced waiting patterns.

---

## Get a Resource's YAML Representation

Dump the intended resource dict as YAML (useful for debugging).

```python
from ocp_resources.pod import Pod

pod = Pod(
    client=client,
    name="debug-pod",
    namespace="default",
    containers=[{"name": "app", "image": "busybox", "command": ["sleep", "3600"]}],
)
print(pod.to_yaml())
```

`to_yaml()` calls `to_dict()` internally and returns the YAML string. This shows what would be sent to the API, not the live state.

---

## Get the Pod's Node and IP

Access commonly needed runtime properties of a pod.

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-nginx", namespace="default", ensure_exists=True)

# Node where the pod is running
print(f"Node: {pod.node.name}")

# Pod IP address
print(f"IP: {pod.ip}")
```

The `node` property returns a `Node` resource object. The `ip` property reads from `status.podIP`.

## Related Pages

- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Querying and Watching Resources](querying-resources.html)
- [Working with Virtual Machines (KubeVirt)](working-with-virtual-machines.html)
- [Executing Commands in Pods and Retrieving Logs](pod-execution-and-logs.html)
- [Working with Kubernetes Events](working-with-events.html)

---

Source: mcp-server-integration.md

# Integrating with AI Assistants via MCP Server

The openshift-python-wrapper ships with a built-in [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) server that lets AI assistants like Cursor, Claude Desktop, and other MCP-compatible tools directly manage your OpenShift/Kubernetes cluster resources.

## Configure the MCP Server for Cursor

Connect your Cursor IDE to your cluster so the AI assistant can list, create, update, and delete resources on your behalf.

```json
// ~/.cursor/mcp.json
{
  "mcpServers": {
    "openshift-python-wrapper": {
      "command": "openshift-mcp-server"
    }
  }
}
```

After saving, restart Cursor and reference the server with `@openshift-python-wrapper` in your prompts. The server uses your active kubeconfig context automatically.

> **Note:** The `openshift-mcp-server` command is installed as a script entry point when you install the package. See [Installing and Creating Your First Resource](quickstart.html) for installation instructions.

## Configure the MCP Server for Claude Desktop

Set up Claude Desktop to interact with your cluster through natural language.

```json
// macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
// Windows: %APPDATA%\Claude\claude_desktop_config.json
// Linux: ~/.config/Claude/claude_desktop_config.json
{
  "mcpServers": {
    "openshift-python-wrapper": {
      "command": "openshift-mcp-server"
    }
  }
}
```

Claude Desktop will connect to the server on startup. You can then ask it to list pods, create resources, check logs, and more using natural language.

## Configure a Custom Kubeconfig Path

Point the MCP server at a specific kubeconfig when your default location doesn't apply.

```json
{
  "mcpServers": {
    "openshift-python-wrapper": {
      "command": "openshift-mcp-server",
      "env": {
        "KUBECONFIG": "/home/user/.kube/my-cluster-config"
      }
    }
  }
}
```

The `KUBECONFIG` environment variable is passed to the server process at startup. This is useful when managing multiple clusters or when your kubeconfig is in a non-standard location.

> **Tip:** See [Connecting to Clusters](connecting-to-clusters.html) for all authentication methods supported by openshift-python-wrapper, including tokens, basic auth, and in-cluster config.

## Run the MCP Server from Source (Development)

Run the server directly from a cloned repository for development or testing.

```bash
git clone https://github.com/RedHatQE/openshift-python-wrapper.git
cd openshift-python-wrapper
uv run mcp_server/server.py
```

For MCP client configuration when running from source:

```json
{
  "mcpServers": {
    "openshift-python-wrapper": {
      "command": "uv",
      "args": ["run", "mcp_server/server.py"],
      "cwd": "/path/to/openshift-python-wrapper"
    }
  }
}
```

This bypasses the installed entry point and runs the server module directly, which is useful when testing local changes.

## List Cluster Resources via AI Chat

Ask your AI assistant to list resources — the MCP server exposes a `list_resources` tool that supports type filtering, namespaces, label selectors, and limits.

```
Prompt: "List all pods in the openshift-monitoring namespace with label app=prometheus"
```

The AI will call the `list_resources` tool, which maps to:

```python
list_resources(
    resource_type="pod",
    namespace="openshift-monitoring",
    label_selector="app=prometheus",
    limit=20
)
```

Returned data includes name, namespace, UID, creation timestamp, labels, phase, and conditions for each matched resource.

> **Tip:** Use `get_resource_types` to discover all available resource types. The server dynamically scans every module in `ocp_resources/` at startup, so any resource supported by the library — including OpenShift-specific types like `route`, `virtualmachine`, or `clusterserviceversion` — is available.

## Get Detailed Resource Information

Retrieve a single resource with full metadata, status, and optionally the raw YAML or JSON.

```
Prompt: "Show me the deployment 'api-server' in namespace 'production' as YAML"
```

The AI calls:

```python
get_resource(
    resource_type="deployment",
    name="api-server",
    namespace="production",
    output_format="yaml"    # also accepts "json" or "info"
)
```

The `info` format (default) returns structured metadata plus resource-specific details: for pods you get container statuses and node placement; for deployments you get replica counts; for services you get ports and cluster IP.

## Create Resources Through Natural Language

Ask the AI to create resources — it can use either YAML content or structured spec parameters.

```
Prompt: "Create a ConfigMap called 'app-settings' in the 'staging' namespace with keys
database_host=db.staging.svc and log_level=debug"
```

The AI calls:

```python
create_resource(
    resource_type="configmap",
    name="app-settings",
    namespace="staging",
    spec={"data": {"database_host": "db.staging.svc", "log_level": "debug"}},
    labels={"managed-by": "mcp-server"}
)
```

You can also provide full YAML:

```python
create_resource(
    resource_type="pod",
    name="nginx",
    yaml_content="""
apiVersion: v1
kind: Pod
metadata:
  name: nginx
  namespace: default
spec:
  containers:
  - name: nginx
    image: nginx:1.27
    ports:
    - containerPort: 80
"""
)
```

> **Warning:** The MCP server creates real resources on your cluster. Make sure your AI assistant is targeting the correct cluster and namespace before confirming creation operations.

## Apply Multi-Document YAML Manifests

Deploy a complete application stack from a single YAML block containing multiple resources.

```
Prompt: "Apply this manifest to create a namespace, deployment, and service for my app"
```

The AI calls:

```python
apply_yaml(yaml_content="""
---
apiVersion: v1
kind: Namespace
metadata:
  name: my-app
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: backend
  namespace: my-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: backend
  template:
    metadata:
      labels:
        app: backend
    spec:
      containers:
      - name: api
        image: myapp/backend:v1.0
        ports:
        - containerPort: 8080
---
apiVersion: v1
kind: Service
metadata:
  name: backend
  namespace: my-app
spec:
  selector:
    app: backend
  ports:
  - port: 80
    targetPort: 8080
""")
```

The response includes a summary with `total_resources`, `successful`, and `failed` counts, plus per-resource results. Each YAML document is parsed and deployed independently.

## Update Resources with Patches

Modify existing resources using merge or strategic-merge patches.

```
Prompt: "Scale the 'web-frontend' deployment in 'production' to 5 replicas"
```

The AI calls:

```python
update_resource(
    resource_type="deployment",
    name="web-frontend",
    namespace="production",
    patch={"spec": {"replicas": 5}},
    patch_type="merge"       # or "strategic"
)
```

The `patch_type` parameter controls the content type: `"merge"` uses `application/merge-patch+json`, while `"strategic"` uses `application/strategic-merge-patch+json`.

## Delete Resources

Remove resources with optional wait-for-deletion behavior.

```
Prompt: "Delete the pod 'debug-pod' in namespace 'default' and wait for it to be gone"
```

The AI calls:

```python
delete_resource(
    resource_type="pod",
    name="debug-pod",
    namespace="default",
    wait=True,
    timeout=60
)
```

If the resource doesn't exist, the server returns a success response with a warning rather than an error, making the operation idempotent.

## Retrieve Pod Logs

Fetch logs from running or crashed pods with filtering options.

```
Prompt: "Show me the last 200 lines of logs from pod 'api-server-7f8b9c' container 'app'
in namespace 'production'"
```

The AI calls:

```python
get_pod_logs(
    name="api-server-7f8b9c",
    namespace="production",
    container="app",
    tail_lines=200
)
```

- Use `since_seconds=3600` to get logs from the last hour
- Use `previous=True` to get logs from a crashed container's previous instance
- Omit `container` for single-container pods

## Execute Commands in Pods

Run diagnostic commands inside a running pod container.

```
Prompt: "Check the nginx config in pod 'web-server' namespace 'production'"
```

The AI calls:

```python
exec_in_pod(
    name="web-server",
    namespace="production",
    command=["nginx", "-t"],
    container="nginx"
)
```

The response includes `stdout`, `stderr`, and `returncode`. If the command fails, the exit code and error output are captured rather than raising an exception.

> **Warning:** Pod exec gives shell-level access to your containers. Ensure your RBAC policies restrict which service accounts and users can perform `exec` operations.

## Get Resource Events

Retrieve Kubernetes events related to a specific resource for troubleshooting.

```
Prompt: "Show me the events for pod 'crashloop-pod' in 'default' namespace"
```

The AI calls:

```python
get_resource_events(
    resource_type="pod",
    name="crashloop-pod",
    namespace="default",
    limit=10
)
```

Events are filtered using field selectors on `involvedObject.name`, `involvedObject.namespace`, and `involvedObject.kind`. Each event includes type (Normal/Warning), reason, message, count, timestamps, and source component.

## Troubleshoot a Failing Pod (End-to-End Workflow)

Combine multiple MCP tools in sequence to diagnose pod issues — the AI assistant will chain these calls automatically.

```
Prompt: "Pod 'payment-svc-6d4f8' in namespace 'checkout' keeps crashing. Help me figure out why."
```

The AI will typically execute this sequence:

1. **Check status:** `get_resource(resource_type="pod", name="payment-svc-6d4f8", namespace="checkout")` — gets phase, container statuses, restart count
2. **Review events:** `get_resource_events(resource_type="pod", name="payment-svc-6d4f8", namespace="checkout")` — shows scheduling, pulling, crash events
3. **Read current logs:** `get_pod_logs(name="payment-svc-6d4f8", namespace="checkout", tail_lines=100)` — application output before crash
4. **Read previous crash logs:** `get_pod_logs(name="payment-svc-6d4f8", namespace="checkout", previous=True)` — logs from the last terminated container
5. **Inspect config:** `exec_in_pod(name="payment-svc-6d4f8", namespace="checkout", command=["cat", "/etc/app/config.yaml"])` — verify mounted configuration

The AI then synthesizes all the data into a diagnosis with suggested fixes.

## Write a Programmatic MCP Client

Connect to the MCP server programmatically using `FastMCPClient` for automation scripts.

```python
import asyncio
from fastmcp import FastMCPClient


async def main():
    async with FastMCPClient() as client:
        await client.connect_stdio(cmd=["openshift-mcp-server"])

        # Discover available resource types
        types = await client.call_tool(
            name="get_resource_types",
            arguments={"random_string": "x"},
        )
        print(f"Available types: {types['total_count']}")

        # List pods in a namespace
        pods = await client.call_tool(
            name="list_resources",
            arguments={"resource_type": "pod", "namespace": "default", "limit": 5},
        )
        for pod in pods["resources"]:
            print(f"  {pod['name']} — {pod.get('phase', 'Unknown')}")


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

This uses the same stdio transport that AI assistants use. The `FastMCPClient` is provided by the `fastmcp` library, which is already a dependency of openshift-python-wrapper.

## Debug MCP Server Issues

Enable debug logging and inspect the server log file when tools aren't behaving as expected.

```bash
# Check the MCP server log file
tail -f /tmp/mcp_server_debug.log
```

```bash
# Run with debug-level logging from the simple-logger
SIMPLE_LOGGER_LEVEL=DEBUG openshift-mcp-server
```

The server writes debug logs to `/tmp/mcp_server_debug.log` by default, including every resource scan at startup, client creation events, and detailed error tracebacks.

- Verify cluster connectivity independently: `kubectl cluster-info`
- Check your permissions: `kubectl auth can-i --list`
- Confirm the entry point is installed: `which openshift-mcp-server`

> **Tip:** See [Environment Variables and Configuration](environment-variables.html) for other environment variables that affect library behavior.

## Available MCP Tools Reference

The MCP server exposes ten tools. All resource type names are case-insensitive (e.g., `"Pod"`, `"pod"`, and `"POD"` all work).

| Tool | Purpose | Required Parameters |
|------|---------|-------------------|
| `list_resources` | List resources by type with optional filters | `resource_type` |
| `get_resource` | Get a single resource by name | `resource_type`, `name` |
| `create_resource` | Create a resource from spec or YAML | `resource_type`, `name`, plus `spec` or `yaml_content` |
| `update_resource` | Patch an existing resource | `resource_type`, `name`, `patch` |
| `delete_resource` | Delete a resource | `resource_type`, `name` |
| `get_pod_logs` | Retrieve pod container logs | `name`, `namespace` |
| `exec_in_pod` | Execute a command in a pod | `name`, `namespace`, `command` |
| `get_resource_events` | Get events for a resource | `resource_type`, `name` |
| `apply_yaml` | Apply one or more YAML documents | `yaml_content` |
| `get_resource_types` | List all discovered resource types | `random_string` (any value) |

> **Note:** Namespaced resources (pods, deployments, configmaps, etc.) require a `namespace` parameter. Cluster-scoped resources (nodes, namespaces, cluster roles) do not. The server validates this automatically and returns a clear error if a namespace is missing.

## Related Pages

- [Connecting to Clusters](connecting-to-clusters.html)
- [Installing and Creating Your First Resource](quickstart.html)
- [Environment Variables and Configuration](environment-variables.html)
- [Executing Commands in Pods and Retrieving Logs](pod-execution-and-logs.html)
- [Querying and Watching Resources](querying-resources.html)

---

Source: resource-api.md

Let me start by reading the knowledge graph and the pages manifest, then explore the source code.

Now let me examine the constants and exceptions:

Now let me check the `apply_yaml` function and the `get_client` function signature, plus see what `ResourceConstants` inherits:

Let me look for key public methods in resource.py that I might have missed (the file was truncated before):

Let me also check for import paths:

Now let me look at the `NamespacedResource._base_body` more carefully and check for anything more:

Now let me check what the `Event` class looks like for the events method reference:

Let me also check for `generate_yaml_from_template`:

That's a test utility. Let me check for the `skip_existing_resource_creation_teardown` function:

Now I have all the information needed. Let me write the comprehensive API reference page.

# Resource and NamespacedResource API

Complete API reference for the `Resource` and `NamespacedResource` base classes, the `get_client` factory function, and supporting classes (`ResourceEditor`, `ResourceList`, `NamespacedResourceList`, `KubeAPIVersion`).

**Import path:**

```python
from ocp_resources.resource import (
    Resource,
    NamespacedResource,
    ResourceEditor,
    ResourceList,
    NamespacedResourceList,
    get_client,
)
```

---

## `get_client`

Factory function to obtain a Kubernetes `DynamicClient`.

```python
from ocp_resources.resource import get_client

client = get_client(config_file="~/.kube/config")
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `config_file` | `str \| None` | `None` | Path to a kubeconfig file. Falls back to `KUBECONFIG` env var or `~/.kube/config`. |
| `config_dict` | `dict[str, Any] \| None` | `None` | Dict with kubeconfig configuration. Mutually exclusive with `config_file`. |
| `context` | `str \| None` | `None` | Name of the kubeconfig context to use. |
| `client_configuration` | `kubernetes.client.Configuration \| None` | `None` | Pre-built Kubernetes client configuration object. |
| `persist_config` | `bool` | `True` | Whether to persist config file changes. |
| `temp_file_path` | `str \| None` | `None` | Path to a temporary kubeconfig file. |
| `try_refresh_token` | `bool` | `True` | Try to refresh the authentication token. |
| `username` | `str \| None` | `None` | Username for basic auth. Requires `password` and `host`. |
| `password` | `str \| None` | `None` | Password for basic auth. Requires `username` and `host`. |
| `host` | `str \| None` | `None` | Cluster host URL. |
| `verify_ssl` | `bool \| None` | `None` | Whether to verify SSL certificates. |
| `token` | `str \| None` | `None` | Bearer token for authentication. Requires `host`. |
| `fake` | `bool` | `False` | Return a `FakeDynamicClient` for testing without a cluster. |
| `generate_kubeconfig` | `bool` | `False` | Save the resolved kubeconfig to a temp file and attach the path to the client. |

**Returns:** `DynamicClient | FakeDynamicClient`

```python
# Default kubeconfig
client = get_client()

# With explicit config file and context
client = get_client(config_file="/path/to/kubeconfig", context="my-cluster")

# Token-based auth
client = get_client(host="https://api.cluster.example.com:6443", token="sha256~abc123")

# Basic auth
client = get_client(host="https://api.cluster.example.com:6443", username="admin", password="secret")

# Fake client for unit testing
client = get_client(fake=True)
```

> **Note:** If neither `config_file` nor `config_dict` is provided, the client falls back to the `KUBECONFIG` environment variable, then `~/.kube/config`, and finally in-cluster configuration.

See [Connecting to Clusters](connecting-to-clusters.html) for connection patterns. See [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html) for `fake=True` usage.

---

## `Resource`

```python
from ocp_resources.resource import Resource
```

Base class for **cluster-scoped** Kubernetes/OpenShift resources (e.g., `Namespace`, `Node`, `ClusterRole`). Inherits from `ResourceConstants`. All concrete resource classes inherit from either `Resource` or `NamespacedResource`.

See [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html) for the inheritance model.

### Class Attributes

| Attribute | Type | Default | Description |
|---|---|---|---|
| `api_group` | `str` | `""` | API group for the resource (e.g., `"apps"`, `"batch"`). |
| `api_version` | `str` | `""` | Full API version string (e.g., `"v1"`, `"apps/v1"`). Resolved automatically if `api_group` is set. |
| `singular_name` | `str` | `""` | Singular resource name for API calls. Used to disambiguate when multiple resources match the same kind. |
| `timeout_seconds` | `int` | `60` | Default timeout for API list/watch operations. |

### Constructor

```python
Resource(
    client=client,
    name="my-resource",
)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `client` | `DynamicClient \| None` | `None` | Kubernetes dynamic client. Will be mandatory in the next major release. |
| `name` | `str \| None` | `None` | Resource name. Required unless `yaml_file` or `kind_dict` is provided. |
| `teardown` | `bool` | `True` | Whether this resource should be deleted when used as a context manager. |
| `yaml_file` | `str \| None` | `None` | Path to a YAML file defining the resource. Mutually exclusive with `kind_dict`. |
| `delete_timeout` | `int` | `240` | Timeout in seconds for delete operations. |
| `dry_run` | `bool` | `False` | If `True`, create operations use dry-run mode. |
| `node_selector` | `dict[str, Any] \| None` | `None` | Node selector for scheduling. |
| `node_selector_labels` | `dict[str, str] \| None` | `None` | Node selector labels for scheduling. |
| `config_file` | `str \| None` | `None` | Path to kubeconfig. Deprecated; pass `client` instead. |
| `config_dict` | `dict[str, Any] \| None` | `None` | Kubeconfig dict. |
| `context` | `str \| None` | `None` | Kubeconfig context name. Deprecated; pass `client` instead. |
| `label` | `dict[str, str] \| None` | `None` | Labels to set on the resource. |
| `annotations` | `dict[str, str] \| None` | `None` | Annotations to set on the resource. |
| `api_group` | `str` | `""` | Override the class-level `api_group`. |
| `hash_log_data` | `bool` | `True` | Hash sensitive fields (defined by `keys_to_hash`) in log output. |
| `ensure_exists` | `bool` | `False` | Check that the resource already exists on the cluster at init time. Raises `ResourceNotFoundError` if not. |
| `kind_dict` | `dict[Any, Any] \| None` | `None` | Complete resource dict. Mutually exclusive with `yaml_file`. Bypasses `to_dict()` logic. |
| `wait_for_resource` | `bool` | `False` | When used as a context manager, wait for the resource to exist after creation. |
| `schema_validation_enabled` | `bool` | `False` | Enable automatic OpenAPI schema validation on `create()` and `update_replace()`. |

**Raises:**

| Exception | Condition |
|---|---|
| `ValueError` | Both `yaml_file` and `kind_dict` are provided. |
| `NotImplementedError` | Neither `api_group` nor `api_version` is defined on the class. |
| `MissingRequiredArgumentError` | None of `name`, `yaml_file`, or `kind_dict` is provided. |
| `ResourceNotFoundError` | `ensure_exists=True` and the resource does not exist. |

```python
from ocp_resources.namespace import Namespace

# Basic creation
ns = Namespace(client=client, name="my-namespace")

# From a YAML file
ns = Namespace(client=client, yaml_file="namespace.yaml")

# From a dict
ns = Namespace(client=client, kind_dict={"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "test"}})

# With labels and annotations
ns = Namespace(client=client, name="my-ns", label={"env": "test"}, annotations={"owner": "team-a"})

# Verify it already exists
ns = Namespace(client=client, name="default", ensure_exists=True)
```

---

### Context Manager Protocol

`Resource` supports the Python context manager protocol for automatic creation and cleanup.

```python
from ocp_resources.namespace import Namespace

with Namespace(client=client, name="temp-ns") as ns:
    # Resource is created on __enter__
    print(ns.name)
# Resource is deleted on __exit__ (if teardown=True)
```

| Method | Description |
|---|---|
| `__enter__()` | Calls `deploy(wait=self.wait_for_resource)`. Registers a `SIGINT` handler on the main thread to ensure cleanup on Ctrl+C. Returns `self`. |
| `__exit__(...)` | Calls `clean_up()` if `teardown=True`. Raises `ResourceTeardownError` if cleanup fails. |

> **Tip:** Set `teardown=False` to prevent automatic deletion on context exit.

---

### CRUD Methods

#### `create`

```python
def create(
    self,
    wait: bool = False,
    exceptions_dict: dict[type[Exception], list[str]] = ...,
) -> ResourceInstance | None
```

Create the resource on the cluster.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `wait` | `bool` | `False` | Wait for the resource to exist after creation. |
| `exceptions_dict` | `dict[type[Exception], list[str]]` | `DEFAULT_CLUSTER_RETRY_EXCEPTIONS \| PROTOCOL_ERROR_EXCEPTION_DICT` | Exceptions to retry on. |

**Returns:** `ResourceInstance | None`

**Raises:** `ValidationError` if `schema_validation_enabled=True` and the resource dict is invalid.

```python
ns = Namespace(client=client, name="my-ns")
ns.create(wait=True)
```

#### `delete`

```python
def delete(
    self,
    wait: bool = False,
    timeout: int = 240,
    body: dict[str, Any] | None = None,
) -> bool
```

Delete the resource from the cluster.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `wait` | `bool` | `False` | Wait for the resource to be fully deleted. |
| `timeout` | `int` | `240` | Timeout in seconds when `wait=True`. |
| `body` | `dict[str, Any] \| None` | `None` | Optional delete options body. |

**Returns:** `bool` — `True` if deleted or not found; `False` only if wait timed out.

```python
ns.delete(wait=True, timeout=120)
```

#### `update`

```python
def update(self, resource_dict: dict[str, Any]) -> None
```

Patch the resource with a merge-patch (`application/merge-patch+json`). Only updates the fields present in `resource_dict`.

| Parameter | Type | Description |
|---|---|---|
| `resource_dict` | `dict[str, Any]` | Partial resource dictionary with fields to update. |

> **Note:** Schema validation is **not** applied on `update()` because patches are partial and would fail full-schema validation.

```python
ns.update(resource_dict={"metadata": {"labels": {"env": "staging"}}})
```

#### `update_replace`

```python
def update_replace(self, resource_dict: dict[str, Any]) -> None
```

Replace the full resource. Use this to **remove** existing fields (unlike `update()`, which only adds/modifies).

| Parameter | Type | Description |
|---|---|---|
| `resource_dict` | `dict[str, Any]` | Complete resource dictionary to replace with. |

**Raises:** `ValidationError` if `schema_validation_enabled=True` and the dict is invalid.

```python
full_dict = ns.instance.to_dict()
full_dict["metadata"]["labels"] = {"new-label": "only"}
ns.update_replace(resource_dict=full_dict)
```

---

### Deploy / Clean Up

#### `deploy`

```python
def deploy(self, wait: bool = False) -> Self
```

Create the resource (respects `REUSE_IF_RESOURCE_EXISTS` environment variable).

| Parameter | Type | Default | Description |
|---|---|---|---|
| `wait` | `bool` | `False` | Wait for the resource after creation. |

**Returns:** `Self`

See [Environment Variables and Configuration](environment-variables.html) for `REUSE_IF_RESOURCE_EXISTS` details.

#### `clean_up`

```python
def clean_up(self, wait: bool = True, timeout: int | None = None) -> bool
```

Delete the resource (respects `SKIP_RESOURCE_TEARDOWN` environment variable).

| Parameter | Type | Default | Description |
|---|---|---|---|
| `wait` | `bool` | `True` | Wait for deletion to complete. |
| `timeout` | `int \| None` | `None` | Timeout in seconds. Defaults to `delete_timeout`. |

**Returns:** `bool` — `True` if deleted successfully.

See [Environment Variables and Configuration](environment-variables.html) for `SKIP_RESOURCE_TEARDOWN` details.

---

### Query Methods and Properties

#### `exists`

```python
@property
def exists(self) -> ResourceInstance | None
```

Check if the resource exists on the cluster.

**Returns:** `ResourceInstance` if found, `None` if not.

```python
if ns.exists:
    print("Namespace exists")
```

#### `instance`

```python
@property
def instance(self) -> ResourceInstance
```

Get the live resource instance from the cluster. Retries on transient cluster errors.

**Returns:** `ResourceInstance`

**Raises:** `NotFoundError` if the resource does not exist.

```python
resource_version = ns.instance.metadata.resourceVersion
```

#### `status`

```python
@property
def status(self) -> str
```

Get `status.phase` from the resource instance.

**Returns:** `str` — The status phase string (e.g., `"Running"`, `"Active"`, `"Pending"`).

```python
print(ns.status)  # "Active"
```

#### `labels`

```python
@property
def labels(self) -> ResourceField
```

Get resource labels from `metadata.labels`.

**Returns:** `ResourceField`

```python
for key, value in ns.labels.items():
    print(f"{key}={value}")
```

#### `kind`

```python
@ClassProperty
def kind(cls) -> str | None
```

Get the resource kind name derived from the class hierarchy. This is a **class-level property** — accessible on both the class and instances.

**Returns:** `str | None`

```python
from ocp_resources.namespace import Namespace
print(Namespace.kind)  # "Namespace"
```

#### `api`

```python
@property
def api(self) -> ResourceInstance
```

Get the resource API object (shortcut for `full_api()` with no extra kwargs).

**Returns:** `ResourceInstance`

#### `full_api`

```python
def full_api(self, **kwargs: Any) -> ResourceInstance
```

Get the resource API object with optional filtering kwargs.

| Keyword Argument | Description |
|---|---|
| `pretty` | Pretty-print output. |
| `_continue` | Continuation token for list pagination. |
| `field_selector` | Filter by field. |
| `label_selector` | Filter by label. |
| `limit` | Maximum number of results. |
| `resource_version` | Filter by resource version. |
| `timeout_seconds` | Request timeout. |
| `watch` | Enable watch mode. |

**Returns:** `ResourceInstance`

---

### Class Method: `get`

```python
@classmethod
def get(
    cls,
    client: DynamicClient | None = None,
    dyn_client: DynamicClient | None = None,
    config_file: str = "",
    singular_name: str = "",
    exceptions_dict: dict[type[Exception], list[str]] = ...,
    raw: bool = False,
    context: str | None = None,
    *args: Any,
    **kwargs: Any,
) -> Generator[Any, None, None]
```

List resources of this kind from the cluster.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `client` | `DynamicClient \| None` | `None` | Kubernetes client. |
| `dyn_client` | `DynamicClient \| None` | `None` | Deprecated alias for `client`. |
| `config_file` | `str` | `""` | Path to kubeconfig. Deprecated; pass `client`. |
| `singular_name` | `str` | `""` | Singular resource name for disambiguation. |
| `exceptions_dict` | `dict` | `DEFAULT_CLUSTER_RETRY_EXCEPTIONS` | Exceptions to retry. |
| `raw` | `bool` | `False` | If `True`, yield raw `ResourceInstance` objects instead of wrapper instances. |
| `context` | `str \| None` | `None` | Kubeconfig context. Deprecated; pass `client`. |
| `**kwargs` | | | Passed through to the API (e.g., `label_selector`, `field_selector`). |

**Returns:** `Generator` of resource instances.

> **Note:** For `Resource` (cluster-scoped), yielded objects are constructed with `name` only. For `NamespacedResource`, yielded objects include both `name` and `namespace`.

```python
from ocp_resources.namespace import Namespace

for ns in Namespace.get(client=client):
    print(ns.name)

# With label selector
for ns in Namespace.get(client=client, label_selector="env=production"):
    print(ns.name)

# Raw mode
for raw_ns in Namespace.get(client=client, raw=True):
    print(raw_ns.metadata.name, raw_ns.status.phase)
```

See [Querying and Watching Resources](querying-resources.html) for advanced list/filter patterns.

---

### Static Method: `get_all_cluster_resources`

```python
@staticmethod
def get_all_cluster_resources(
    client: DynamicClient | None = None,
    config_file: str = "",
    context: str | None = None,
    config_dict: dict[str, Any] | None = None,
    *args: Any,
    **kwargs: Any,
) -> Generator[ResourceField, None, None]
```

Yield all resources across **all** API groups in the cluster.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `client` | `DynamicClient \| None` | `None` | Kubernetes client. |
| `**kwargs` | | | Passed to the API (e.g., `label_selector`). |

**Yields:** `ResourceField`

```python
for resource in Resource.get_all_cluster_resources(client=client, label_selector="app=myapp"):
    print(f"{resource.kind}: {resource.metadata.name}")
```

---

### Wait Methods

#### `wait`

```python
def wait(self, timeout: int = 240, sleep: int = 1) -> None
```

Wait until the resource exists on the cluster.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `timeout` | `int` | `240` | Maximum wait time in seconds. |
| `sleep` | `int` | `1` | Sleep interval between retries. |

**Raises:** `TimeoutExpiredError` if the resource does not exist within the timeout.

#### `wait_deleted`

```python
def wait_deleted(self, timeout: int = 240) -> bool
```

Wait until the resource no longer exists.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `timeout` | `int` | `240` | Maximum wait time in seconds. |

**Returns:** `bool` — `True` if deleted, `False` if timed out.

#### `wait_for_status`

```python
def wait_for_status(
    self,
    status: str,
    timeout: int = 240,
    stop_status: str | None = None,
    sleep: int = 1,
    exceptions_dict: dict[type[Exception], list[str]] = ...,
) -> None
```

Wait for `status.phase` to reach the expected value.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `status` | `str` | — | Expected status phase string. |
| `timeout` | `int` | `240` | Maximum wait time in seconds. |
| `stop_status` | `str \| None` | `None` | Stop and fail immediately if this status is reached. Defaults to `Status.FAILED`. |
| `sleep` | `int` | `1` | Sleep interval between retries. |
| `exceptions_dict` | `dict` | `PROTOCOL_ERROR_EXCEPTION_DICT \| DEFAULT_CLUSTER_RETRY_EXCEPTIONS` | Exceptions to retry. |

**Raises:** `TimeoutExpiredError` if the desired status is not reached.

```python
from ocp_resources.pod import Pod

pod.wait_for_status(status=Pod.Status.RUNNING, timeout=120)
```

#### `wait_for_condition`

```python
def wait_for_condition(
    self,
    condition: str,
    status: str,
    timeout: int = 300,
    sleep_time: int = 1,
    reason: str | None = None,
    message: str = "",
    stop_condition: str | None = None,
    stop_status: str = "True",
) -> None
```

Wait for a specific condition to reach the desired state.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `condition` | `str` | — | Condition type name (e.g., `"Ready"`, `"Available"`). |
| `status` | `str` | — | Expected condition status (e.g., `"True"`, `"False"`). |
| `timeout` | `int` | `300` | Maximum wait time in seconds. |
| `sleep_time` | `int` | `1` | Interval between retries. |
| `reason` | `str \| None` | `None` | Expected condition reason. If set, must match exactly. |
| `message` | `str` | `""` | Expected substring within the condition message. |
| `stop_condition` | `str \| None` | `None` | Condition type that, if matched, stops the wait and fails immediately. |
| `stop_status` | `str` | `"True"` | Status for `stop_condition` matching. |

**Raises:**

| Exception | Condition |
|---|---|
| `TimeoutExpiredError` | Condition not met within timeout. |
| `ConditionError` | `stop_condition` is detected with matching `stop_status`. |

```python
ns.wait_for_condition(
    condition="Ready",
    status="True",
    timeout=60,
)
```

See [Waiting for Resource Conditions and Status](waiting-for-conditions.html) for more patterns.

#### `wait_for_conditions`

```python
def wait_for_conditions(self) -> None
```

Wait for the resource to have any conditions populated in its status. Uses a 30-second timeout.

---

### Serialization

#### `to_dict`

```python
def to_dict(self) -> None
```

Populate `self.res` with the intended dict representation of the resource. Called automatically before `create()`. Override this in subclasses to add resource-specific fields.

> **Note:** If `kind_dict` or `yaml_file` was provided, `to_dict()` uses those directly instead of building from individual parameters.

#### `to_yaml`

```python
def to_yaml(self) -> str
```

**Returns:** `str` — YAML string representation of the resource dict.

```python
print(ns.to_yaml())
```

---

### Watch / Events

#### `watcher`

```python
def watcher(
    self,
    timeout: int,
    resource_version: str = "",
) -> Generator[dict[str, Any], None, None]
```

Watch for changes to this specific resource.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `timeout` | `int` | — | Watch duration in seconds. |
| `resource_version` | `str` | `""` | Only return events after this version. Defaults to the version at resource creation time. |

**Yields:** Event dicts with keys `type` (`"ADDED"`, `"MODIFIED"`, `"DELETED"`), `raw_object`, and `object`.

```python
for event in ns.watcher(timeout=30):
    print(event["type"], event["object"].metadata.name)
```

#### `events`

```python
def events(
    self,
    name: str = "",
    label_selector: str = "",
    field_selector: str = "",
    resource_version: str = "",
    timeout: int = 240,
) -> Generator[Any, Any, None]
```

Get Kubernetes events related to this resource.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `name` | `str` | `""` | Filter by event name. |
| `label_selector` | `str` | `""` | Comma-separated `key=value` label filters. |
| `field_selector` | `str` | `""` | Additional field selectors (auto-prefixed with `involvedObject.name==<resource_name>`). |
| `resource_version` | `str` | `""` | Filter events by resource version. |
| `timeout` | `int` | `240` | Timeout in seconds. |

**Yields:** `Event` objects.

```python
for event in pod.events(field_selector="type==Warning", timeout=10):
    print(event.object)
```

---

### Validation

#### `validate`

```python
def validate(self) -> None
```

Validate `self.res` against the OpenAPI schema for this resource kind. Called automatically during `create()` and `update_replace()` when `schema_validation_enabled=True`.

**Raises:** `ValidationError` if the resource dict is invalid.

```python
pod = Pod(client=client, name="test", namespace="default")
pod.to_dict()
pod.validate()  # Raises ValidationError on invalid spec
```

#### `validate_dict` (classmethod)

```python
@classmethod
def validate_dict(cls, resource_dict: dict[str, Any]) -> None
```

Validate an arbitrary resource dictionary against the schema without creating a resource instance.

| Parameter | Type | Description |
|---|---|---|
| `resource_dict` | `dict[str, Any]` | Complete resource dictionary. |

**Raises:** `ValidationError` if validation fails.

```python
from ocp_resources.pod import Pod

Pod.validate_dict({
    "apiVersion": "v1",
    "kind": "Pod",
    "metadata": {"name": "test"},
    "spec": {"containers": [{"name": "web", "image": "nginx"}]},
})
```

See [Validating Resources Against OpenAPI Schemas](validating-resources.html) for full validation guide.

---

### API Request

#### `api_request`

```python
def api_request(
    self,
    method: str,
    action: str,
    url: str,
    retry_params: dict[str, int] | None = None,
    **params: Any,
) -> dict[str, Any]
```

Send a raw HTTP request to the resource's API endpoint. Used internally for subresource actions (e.g., VirtualMachine start/stop).

| Parameter | Type | Default | Description |
|---|---|---|---|
| `method` | `str` | — | HTTP method (`"GET"`, `"PUT"`, `"POST"`, etc.). |
| `action` | `str` | — | Subresource action to append to the URL (e.g., `"start"`, `"stop"`). |
| `url` | `str` | — | Base URL of the resource. |
| `retry_params` | `dict[str, int] \| None` | `None` | Dict with `timeout` and `sleep_time` keys for retry behavior. |
| `**params` | | | Additional params passed to the HTTP request. |

**Returns:** `dict[str, Any]` — Parsed JSON response, or raw response data if not valid JSON.

---

### Utility Methods

#### `retry_cluster_exceptions` (static)

```python
@staticmethod
def retry_cluster_exceptions(
    func: Callable,
    exceptions_dict: dict[type[Exception], list[str]] = DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
    timeout: int = 10,
    sleep_time: int = 1,
    **kwargs: Any,
) -> Any
```

Retry a callable on transient cluster errors.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `func` | `Callable` | — | Function to call. |
| `exceptions_dict` | `dict` | `DEFAULT_CLUSTER_RETRY_EXCEPTIONS` | Map of exception types to message substrings to match. |
| `timeout` | `int` | `10` | Total retry timeout in seconds. |
| `sleep_time` | `int` | `1` | Sleep between retries. |
| `**kwargs` | | | Passed to `func`. |

**Returns:** The return value of `func`.

**Raises:** The last exception if timeout is reached.

#### `get_condition_message`

```python
def get_condition_message(
    self,
    condition_type: str,
    condition_status: str = "",
) -> str
```

Get the message for a specific condition.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `condition_type` | `str` | — | Condition type name. |
| `condition_status` | `str` | `""` | If set, only return the message when the condition status matches. |

**Returns:** `str` — The condition message, or `""` if not found or status doesn't match.

#### `hash_resource_dict`

```python
def hash_resource_dict(self, resource_dict: dict[Any, Any]) -> dict[Any, Any]
```

Replace sensitive fields (defined by `keys_to_hash`) with `"*******"` for logging.

| Parameter | Type | Description |
|---|---|---|
| `resource_dict` | `dict` | The resource dict to hash. |

**Returns:** `dict` — A copy with sensitive values replaced.

> **Tip:** Override the `keys_to_hash` property in subclasses to define which fields to mask.

#### `keys_to_hash`

```python
@property
def keys_to_hash(self) -> list[str]
```

List of key paths to mask in logs. Uses `>` as separator, `[]` for list elements.

**Returns:** `list[str]` — Default: `[]` (no hashing).

```python
# Example override in a Secret subclass:
@property
def keys_to_hash(self):
    return ["data", "stringData"]
```

---

### Inner Classes

#### `Resource.Status`

Pre-defined status phase constants (inherited from `ResourceConstants`).

| Constant | Value |
|---|---|
| `Status.SUCCEEDED` | `"Succeeded"` |
| `Status.FAILED` | `"Failed"` |
| `Status.DELETING` | `"Deleting"` |
| `Status.DEPLOYED` | `"Deployed"` |
| `Status.PENDING` | `"Pending"` |
| `Status.COMPLETED` | `"Completed"` |
| `Status.RUNNING` | `"Running"` |
| `Status.READY` | `"Ready"` |
| `Status.TERMINATING` | `"Terminating"` |
| `Status.ERROR` | `"Error"` |
| `Status.ACTIVE` | `"Active"` |
| `Status.SCHEDULING` | `"Scheduling"` |
| `Status.CRASH_LOOPBACK_OFF` | `"CrashLoopBackOff"` |
| `Status.IMAGE_PULL_BACK_OFF` | `"ImagePullBackOff"` |

```python
pod.wait_for_status(status=Pod.Status.RUNNING)
```

#### `Resource.Condition`

Pre-defined condition type and status constants.

| Constant | Value |
|---|---|
| `Condition.READY` | `"Ready"` |
| `Condition.AVAILABLE` | `"Available"` |
| `Condition.DEGRADED` | `"Degraded"` |
| `Condition.PROGRESSING` | `"Progressing"` |
| `Condition.UPGRADEABLE` | `"Upgradeable"` |
| `Condition.Status.TRUE` | `"True"` |
| `Condition.Status.FALSE` | `"False"` |
| `Condition.Status.UNKNOWN` | `"Unknown"` |

```python
ns.wait_for_condition(
    condition=Resource.Condition.READY,
    status=Resource.Condition.Status.TRUE,
)
```

#### `Resource.ApiGroup`

Pre-defined API group string constants. A selection of commonly used values:

| Constant | Value |
|---|---|
| `ApiGroup.APPS` | `"apps"` |
| `ApiGroup.BATCH` | `"batch"` |
| `ApiGroup.NETWORKING_K8S_IO` | `"networking.k8s.io"` |
| `ApiGroup.RBAC_AUTHORIZATION_K8S_IO` | `"rbac.authorization.k8s.io"` |
| `ApiGroup.STORAGE_K8S_IO` | `"storage.k8s.io"` |
| `ApiGroup.CONFIG_OPENSHIFT_IO` | `"config.openshift.io"` |
| `ApiGroup.KUBEVIRT_IO` | `"kubevirt.io"` |
| `ApiGroup.CDI_KUBEVIRT_IO` | `"cdi.kubevirt.io"` |
| `ApiGroup.ROUTE_OPENSHIFT_IO` | `"route.openshift.io"` |
| `ApiGroup.MACHINE_OPENSHIFT_IO` | `"machine.openshift.io"` |

> **Note:** Over 100 API group constants are available. Use IDE auto-complete to discover all values.

#### `Resource.ApiVersion`

| Constant | Value |
|---|---|
| `ApiVersion.V1` | `"v1"` |
| `ApiVersion.V1BETA1` | `"v1beta1"` |
| `ApiVersion.V1ALPHA1` | `"v1alpha1"` |
| `ApiVersion.V1ALPHA3` | `"v1alpha3"` |

---

## `NamespacedResource`

```python
from ocp_resources.resource import NamespacedResource
```

Base class for **namespace-scoped** resources (e.g., `Pod`, `Deployment`, `Service`, `ConfigMap`). Extends `Resource` with namespace awareness.

### Constructor

```python
NamespacedResource(
    client=client,
    name="my-pod",
    namespace="my-namespace",
)
```

All parameters from [`Resource`](#constructor) are accepted, plus:

| Parameter | Type | Default | Description |
|---|---|---|---|
| `namespace` | `str \| None` | `None` | Kubernetes namespace. Required unless `yaml_file` or `kind_dict` is provided. |

**Raises:** `MissingRequiredArgumentError` if neither (`name` + `namespace`) nor `yaml_file` / `kind_dict` is provided.

```python
from ocp_resources.pod import Pod

pod = Pod(
    client=client,
    name="nginx",
    namespace="default",
    label={"app": "web"},
)
```

### Overridden Methods

#### `instance`

```python
@property
def instance(self) -> ResourceInstance
```

Get the live resource instance, scoped to `self.namespace`.

**Returns:** `ResourceInstance`

#### `to_dict`

```python
def to_dict(self) -> None
```

Populates `self.res` and sets `metadata.namespace`. If using `yaml_file` or `kind_dict`, reads the namespace from the YAML/dict.

**Raises:** `MissingRequiredArgumentError` if namespace is still not set after processing.

#### `get` (classmethod)

Behaves like `Resource.get()`, but yields instances constructed with both `name` and `namespace`.

```python
from ocp_resources.pod import Pod

for pod in Pod.get(client=client, namespace="default"):
    print(f"{pod.namespace}/{pod.name}")

# With label selector
for pod in Pod.get(client=client, namespace="default", label_selector="app=web"):
    print(pod.name)
```

---

## `ResourceEditor`

```python
from ocp_resources.resource import ResourceEditor
```

Temporarily patch resources and automatically restore original values. Designed for test scenarios.

### Constructor

```python
ResourceEditor(
    patches: dict[Any, Any],
    action: str = "update",
    user_backups: dict[Any, Any] | None = None,
)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `patches` | `dict` | — | Map of `{Resource: patch_dict}`. |
| `action` | `str` | `"update"` | `"update"` for merge-patch, `"replace"` for full replacement. |
| `user_backups` | `dict \| None` | `None` | Pre-computed backup dicts. If provided, skips automatic backup creation. |

### Context Manager Usage

```python
from ocp_resources.resource import ResourceEditor
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="my-ns", ensure_exists=True)

with ResourceEditor(
    patches={ns: {"metadata": {"labels": {"temporary": "true"}}}}
) as editor:
    # Labels are patched
    assert ns.instance.metadata.labels.temporary == "true"
# Labels are restored to original values
```

### Methods

| Method | Signature | Description |
|---|---|---|
| `update` | `update(backup_resources: bool = False) -> None` | Apply patches. If `backup_resources=True`, back up original values first. |
| `restore` | `restore() -> None` | Restore all backed-up values. |

### Properties

| Property | Type | Description |
|---|---|---|
| `backups` | `dict[Any, Any]` | Backed-up original values for each patched resource. |
| `patches` | `dict[Any, Any]` | The patches dict from the constructor. |

> **Warning:** The `DynamicClient` used to obtain the resource objects must have sufficient privileges to patch and replace resources.

See [Editing Resources Temporarily with ResourceEditor](editing-resources-temporarily.html) for detailed patterns.

---

## `ResourceList`

```python
from ocp_resources.resource import ResourceList
```

Create and manage N copies of a cluster-scoped resource with indexed names.

### Constructor

```python
ResourceList(
    resource_class: type[Resource],
    num_resources: int,
    client: DynamicClient,
    **kwargs: Any,
)
```

| Parameter | Type | Description |
|---|---|---|
| `resource_class` | `type[Resource]` | The resource class to instantiate. |
| `num_resources` | `int` | Number of resource copies to create. |
| `client` | `DynamicClient` | Kubernetes client. |
| `**kwargs` | | Passed to each resource constructor. Must include `name` (used as base name). |

Resources are named `{name}-1`, `{name}-2`, ..., `{name}-N`.

```python
from ocp_resources.namespace import Namespace
from ocp_resources.resource import ResourceList

namespaces = ResourceList(
    resource_class=Namespace,
    num_resources=3,
    client=client,
    name="test-ns",
)

with namespaces:
    # Creates test-ns-1, test-ns-2, test-ns-3
    for ns in namespaces:
        print(ns.name)
# All three namespaces are deleted
```

### Methods (inherited from `BaseResourceList`)

| Method | Signature | Returns | Description |
|---|---|---|---|
| `deploy` | `deploy(wait=False)` | `list[Resource]` | Deploy all resources. |
| `clean_up` | `clean_up(wait=True)` | `bool` | Delete all resources in reverse order. |
| `__len__` | `__len__()` | `int` | Number of resources. |
| `__getitem__` | `__getitem__(index)` | `Resource` | Access by index. |
| `__iter__` | `__iter__()` | `Generator` | Iterate over resources. |

---

## `NamespacedResourceList`

```python
from ocp_resources.resource import NamespacedResourceList
```

Create one instance of a namespaced resource in each namespace from a `ResourceList`.

### Constructor

```python
NamespacedResourceList(
    resource_class: type[NamespacedResource],
    namespaces: ResourceList,
    client: DynamicClient,
    **kwargs: Any,
)
```

| Parameter | Type | Description |
|---|---|---|
| `resource_class` | `type[NamespacedResource]` | The namespaced resource class (e.g., `Pod`). |
| `namespaces` | `ResourceList` | A `ResourceList` of `Namespace` resources. |
| `client` | `DynamicClient` | Kubernetes client. |
| `**kwargs` | | Passed to each resource constructor. Must include `name`. |

**Raises:** `TypeError` if any resource in `namespaces` is not a `Namespace`.

```python
from ocp_resources.namespace import Namespace
from ocp_resources.pod import Pod
from ocp_resources.resource import ResourceList, NamespacedResourceList

namespaces = ResourceList(
    resource_class=Namespace, num_resources=2, client=client, name="test-ns"
)

pods = NamespacedResourceList(
    resource_class=Pod,
    namespaces=namespaces,
    client=client,
    name="nginx",
)

with namespaces:
    with pods:
        # Creates nginx in test-ns-1 and test-ns-2
        for pod in pods:
            print(f"{pod.namespace}/{pod.name}")
```

See [Managing Bulk Resources with ResourceList](managing-resource-lists.html) for full usage patterns.

---

## `KubeAPIVersion`

```python
from ocp_resources.resource import KubeAPIVersion
```

Implements [Kubernetes API versioning](https://kubernetes.io/docs/concepts/overview/kubernetes-api/#api-versioning) with comparison operators. Extends `packaging.version.Version`.

### Constructor

```python
KubeAPIVersion(vstring: str)
```

| Parameter | Type | Description |
|---|---|---|
| `vstring` | `str` | Version string (e.g., `"v1"`, `"v1beta1"`, `"v1alpha2"`). |

**Raises:** `ValueError` if the version string does not conform to Kubernetes versioning.

### Ordering

Versions are ordered: `v1alpha1 < v1alpha2 < v1beta1 < v1beta2 < v1 < v2`.

```python
assert KubeAPIVersion("v1") > KubeAPIVersion("v1beta1")
assert KubeAPIVersion("v1beta1") > KubeAPIVersion("v1alpha1")
assert KubeAPIVersion("v1") == KubeAPIVersion("v1")
```

---

## Exceptions

All exceptions are importable from `ocp_resources.exceptions`.

```python
from ocp_resources.exceptions import (
    MissingRequiredArgumentError,
    ResourceTeardownError,
    ValidationError,
    ConditionError,
)
```

| Exception | Raised By | Description |
|---|---|---|
| `MissingRequiredArgumentError` | `Resource.__init__`, `NamespacedResource.__init__`, `NamespacedResource._base_body` | Required parameters (`name`, `namespace`) not provided. |
| `ResourceTeardownError` | `Resource.__exit__` | `clean_up()` returned `False` during context manager exit. |
| `ValidationError` | `validate()`, `validate_dict()`, `create()`, `update_replace()` | Resource dict fails OpenAPI schema validation. Has `message`, `path`, and `schema_error` attributes. |
| `ConditionError` | `wait_for_condition()` | A `stop_condition` was detected during condition waiting. |
| `MissingResourceResError` | `Resource._base_body` | Deprecated. `self.res` is empty after `_base_body()`. |

---

## Timeout Constants

Available from `ocp_resources.utils.constants`:

```python
from ocp_resources.utils.constants import (
    TIMEOUT_1SEC,      # 1
    TIMEOUT_5SEC,      # 5
    TIMEOUT_10SEC,     # 10
    TIMEOUT_30SEC,     # 30
    TIMEOUT_1MINUTE,   # 60
    TIMEOUT_2MINUTES,  # 120
    TIMEOUT_4MINUTES,  # 240
    TIMEOUT_10MINUTES, # 600
)
```

---

## Module-Level Functions

### `replace_key_with_hashed_value`

```python
from ocp_resources.resource import replace_key_with_hashed_value

replace_key_with_hashed_value(resource_dict: dict, key_name: str) -> dict
```

Recursively replace the value at a key path with `"*******"`. Supports dict paths (`a>b>c`) and list paths (`a>b[]>c`).

| Parameter | Type | Description |
|---|---|---|
| `resource_dict` | `dict` | The nested dictionary. |
| `key_name` | `str` | Key path using `>` as separator, `[]` for list elements. |

**Returns:** `dict` — A deep copy with the value replaced.

```python
data = {"spec": {"secret": {"token": "sensitive"}}}
hashed = replace_key_with_hashed_value(data, "spec>secret>token")
# {"spec": {"secret": {"token": "*******"}}}

# With list path
data = {"spec": {"containers": [{"env": [{"value": "secret"}]}]}}
hashed = replace_key_with_hashed_value(data, "spec>containers[]>env[]>value")
```

## Related Pages

- [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Querying and Watching Resources](querying-resources.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Environment Variables and Configuration](environment-variables.html)

---

Source: class-generator-cli.md

I'll start by reading the knowledge graph and page manifest, then explore the class-generator source code in detail.

Now let me check the entry point configuration and any additional schema-related functions:

Now let me check the test files for CLI tests and the entry point:

Now let me check the parsers and any additional CLI-relevant components:

Now I have all the information I need. Let me compile the comprehensive CLI reference page.

# class-generator CLI Reference

The `class-generator` command generates Python wrapper classes for Kubernetes and OpenShift resources. It is installed as a console script entry point from the `openshift-python-wrapper` package.

```
class-generator = "class_generator.cli:main"
```

```bash
pip install openshift-python-wrapper
# or
uv tool install openshift-python-wrapper
```

> **Tip:** Enable shell completion by adding to `~/.bashrc` or `~/.zshrc`:
> ```bash
> if type class-generator > /dev/null; then eval "$(_CLASS_GENERATOR_COMPLETE=zsh_source class-generator)"; fi
> ```

## Synopsis

```
class-generator [OPTIONS]
```

At least one action option must be specified: `--kind`, `--update-schema`, `--update-schema-for`, `--discover-missing`, `--coverage-report`, `--generate-missing`, or `--regenerate-all`.

---

## Options Reference

### Kind Generation

#### `-k`, `--kind`

| Property | Value |
|----------|-------|
| **Type** | `STRING` |
| **Default** | `None` |
| **Required** | No (but at least one action must be specified) |
| **Requires** | Connected cluster with admin privileges |

Generate Python wrapper classes for the specified Kubernetes resource Kind(s). Multiple kinds can be comma-separated (no spaces).

When a kind is not found in the local schema mapping file, the CLI interactively prompts to run `--update-schema` (only in interactive CLI mode).

```bash
# Single kind
class-generator -k Pod

# Multiple kinds (processed in parallel)
class-generator -k Deployment,Pod,ConfigMap
```

When multiple kinds share the same Kind name but belong to different API groups (e.g., `DNS` from `config.openshift.io` and `operator.openshift.io`), separate files are generated with API group suffixes:

```
dns_config_openshift_io.py
dns_operator_openshift_io.py
```

---

#### `-o`, `--output-file`

| Property | Value |
|----------|-------|
| **Type** | `PATH` |
| **Default** | `ocp_resources/<snake_case_kind>.py` |
| **Required** | No |

Override the output file path for the generated Python module. If not provided, the filename is derived from the Kind using `convert_camel_case_to_snake_case`.

```bash
class-generator -k Pod -o my_custom_pod.py
```

> **Note:** When generating multiple comma-separated kinds, the `--output-file` value applies to all kinds. For independent output paths, run separate commands.

---

#### `--overwrite`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Overwrite an existing output file. Without this flag, if the target file already exists, a `_TEMP.py` suffixed file is created instead.

When overwriting, any user-added code blocks (code after the `# End of generated code` marker) and user imports are preserved in the regenerated file.

```bash
class-generator -k Pod --overwrite
```

---

#### `--dry-run`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Preview the generated output without writing any files. The generated Python code is printed to the console with syntax highlighting and line numbers using Rich.

```bash
class-generator -k Pod --dry-run
```

---

#### `--backup`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |
| **Requires** | `--regenerate-all` or `--overwrite` |

Create a timestamped backup of existing files before overwriting or regenerating. Backups are stored in `.backups/backup-YYYYMMDD-HHMMSS/` preserving the original directory structure.

```bash
class-generator -k Pod --overwrite --backup
```

> **Note:** Using `--backup` without either `--regenerate-all` or `--overwrite` results in a constraint error.

---

### Test Generation

#### `--add-tests`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |
| **Requires** | `-k`/`--kind` |

Generate test files for the specified Kind and run them. This performs two actions:

1. Generates a test manifest in `class_generator/tests/manifests/<Kind>/` and regenerates `class_generator/tests/test_class_generator.py` from the Jinja2 template.
2. Runs the generated test file using `uv run --group tests pytest class_generator/tests/test_class_generator.py`.

```bash
class-generator -k Pod --add-tests
```

> **Warning:** `--add-tests` cannot be used without `-k`/`--kind`. Running `class-generator --add-tests` alone exits with a non-zero status.

---

### Schema Management

#### `--update-schema`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |
| **Requires** | Connected cluster; `oc` or `kubectl` in PATH |
| **Mutually exclusive with** | `--update-schema-for` |

Fetch all resource schemas from the connected cluster's OpenAPI v3 endpoints and update the local schema files:

- `class_generator/schema/__resources-mappings.json` (compressed as `.json.gz`)
- `class_generator/schema/_definitions.json`

The update strategy depends on the cluster version:

| Cluster Version | Behavior |
|-----------------|----------|
| Same or newer than last update | Full update — fetches all schemas, updates existing resources |
| Older than last update | Incremental — only adds missing resources, preserves existing schemas |

When used alone, exits after updating. When combined with `--generate-missing`, continues to resource generation after the update.

```bash
# Update schema only
class-generator --update-schema

# Update schema then generate missing resources
class-generator --update-schema --generate-missing
```

> **Note:** When `--update-schema` is used without `--generate-missing`, it cannot be combined with `-k`, `--discover-missing`, `--coverage-report`, `--dry-run`, `--overwrite`, `-o`, `--add-tests`, or `--regenerate-all`.

---

#### `--update-schema-for`

| Property | Value |
|----------|-------|
| **Type** | `STRING` |
| **Default** | `None` |
| **Requires** | Connected cluster; `oc` or `kubectl` in PATH |
| **Mutually exclusive with** | `--update-schema` |

Update the schema for a single resource Kind without affecting other resources. The Kind name is **case-sensitive**.

This fetches only the API paths relevant to the specified Kind, updates (or adds) its schema in the mapping file, and exits.

```bash
class-generator --update-schema-for LlamaStackDistribution
```

Use cases:
- Connected to an older cluster but need to update a specific CRD
- A new operator was installed and you need its resource schema
- Refreshing just one resource without a full schema update

After updating, regenerate the class:

```bash
class-generator --update-schema-for LlamaStackDistribution
class-generator -k LlamaStackDistribution --overwrite
```

**Errors:**

| Error | Cause |
|-------|-------|
| `ResourceNotFoundError` | The Kind is not found on the cluster (CRD not installed or name misspelled) |
| `RuntimeError` | API paths not found or schema extraction failed |

> **Note:** Cannot be combined with `-k`, `--discover-missing`, `--coverage-report`, `--generate-missing`, or `--regenerate-all`.

---

### Coverage and Discovery

#### `--discover-missing`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Analyze resource coverage by comparing schema-mapped resources against implemented wrapper classes in `ocp_resources/`. Generates a console report showing coverage statistics and missing resources.

```bash
class-generator --discover-missing
```

---

#### `--coverage-report`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Generate a detailed coverage report showing:

| Metric | Description |
|--------|-------------|
| Total Resources in Schema | Number of resource Kinds in the schema mapping |
| Auto-Generated Resources | Wrapper classes with the generated marker |
| Coverage | Percentage of mapped resources with generated classes |
| Missing (Not Generated) | Resources in schema but without generated classes |
| Manual Implementations | Resource classes without the generated marker |

```bash
# Console output (default)
class-generator --coverage-report

# JSON output
class-generator --coverage-report --json
```

---

#### `--json`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Output reports in JSON format instead of Rich console tables. Applies to `--coverage-report`, `--discover-missing`, and `--generate-missing`.

The JSON output structure:

```json
{
  "generated_resources": ["ConfigMap", "Deployment", "Pod"],
  "manual_resources": ["VirtualMachine"],
  "missing_resources": [{"kind": "Binding"}, {"kind": "ComponentStatus"}],
  "coverage_stats": {
    "total_in_mapping": 400,
    "total_generated": 197,
    "total_manual": 25,
    "coverage_percentage": 49.25,
    "missing_count": 203
  }
}
```

```bash
class-generator --coverage-report --json
```

---

#### `--generate-missing`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Generate wrapper classes for all resources found in the schema mapping that do not yet have generated files. Each missing resource Kind is passed to `class_generator()` individually.

Can be combined with `--update-schema` to first refresh the schema, then generate all missing classes.

```bash
# Generate missing resources
class-generator --generate-missing

# Update schema first, then generate missing
class-generator --update-schema --generate-missing

# Dry run to preview
class-generator --generate-missing --dry-run
```

---

### Batch Regeneration

#### `--regenerate-all`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Regenerate all existing generated resource classes using the latest schemas. Only files containing the `# Generated using` marker in `ocp_resources/` are processed.

Regeneration runs in parallel (up to 10 workers). User-added code (below `# End of generated code`) is preserved during regeneration.

```bash
# Regenerate all resources
class-generator --regenerate-all

# With backup
class-generator --regenerate-all --backup

# Dry run
class-generator --regenerate-all --dry-run

# Filter to specific resources
class-generator --regenerate-all --filter "Pod*"
```

Output summary:

```
Regeneration complete: 195 succeeded, 2 failed
Backup files stored in: .backups/backup-20260705-143022
```

---

#### `--filter`

| Property | Value |
|----------|-------|
| **Type** | `STRING` |
| **Default** | `None` |
| **Requires** | `--regenerate-all` |

Filter which resources to regenerate using a glob pattern matched against the resource Kind name. Uses `fnmatch` semantics.

```bash
# Regenerate only Pod-related resources
class-generator --regenerate-all --filter "Pod*"

# Regenerate resources ending in "Service"
class-generator --regenerate-all --filter "*Service"

# Regenerate a specific resource
class-generator --regenerate-all --filter "VirtualMachine"
```

---

### Logging

#### `-v`, `--verbose`

| Property | Value |
|----------|-------|
| **Type** | Flag |
| **Default** | `False` |

Enable verbose output with debug-level logs. Sets `DEBUG` level on the following loggers:

- `class_generator.core.schema`
- `class_generator.core.generator`
- `class_generator.core.coverage`
- `class_generator.core.discovery`
- `class_generator.cli`
- `class_generator.utils`
- `ocp_resources`

```bash
class-generator -k Pod -v
```

---

## Constraint Rules

The CLI enforces the following option constraints:

| Constraint | Rule |
|------------|------|
| `--update-schema` ↔ `--update-schema-for` | Mutually exclusive |
| `--update-schema` (alone) | Cannot combine with `-k`, `--discover-missing`, `--coverage-report`, `--dry-run`, `--overwrite`, `-o`, `--add-tests`, `--regenerate-all` |
| `--update-schema-for` | Cannot combine with `-k`, `--discover-missing`, `--coverage-report`, `--generate-missing`, `--regenerate-all` |
| `--backup` | Requires `--regenerate-all` or `--overwrite` |
| `--filter` | Requires `--regenerate-all` |
| No options | Exits with error — at least one action required |

---

## Execution Order

When multiple compatible options are specified together, the CLI processes them in this fixed order:

1. `--update-schema-for` (exits after completion)
2. `--update-schema` (exits unless `--generate-missing` is also set)
3. `--coverage-report` / `--discover-missing` / `--generate-missing` (coverage analysis and reporting)
4. `--generate-missing` (generates classes for missing resources)
5. `--regenerate-all` (batch regeneration, exits after completion)
6. `-k`/`--kind` (normal kind generation)
7. `--add-tests` (test generation and execution)

---

## Exit Codes

| Code | Meaning |
|------|---------|
| `0` | Success |
| `1` | Generation failure, schema update failure, resource not found, or any kind in a batch failed |
| `2` | Invalid CLI arguments or constraint violation |

---

## Programmatic API

The generation logic can be invoked directly from Python. See [Generating New Resource Classes with class-generator](generating-resource-classes.html) for usage examples.

### `class_generator.core.generator.class_generator()`

```python
from class_generator.core.generator import class_generator

generated_files: list[str] = class_generator(
    kind="Pod",
    overwrite=False,
    dry_run=False,
    output_file="",
    output_dir="",
    add_tests=False,
    called_from_cli=True,
    update_schema_executed=False,
)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `kind` | `str` | *(required)* | Kubernetes resource Kind |
| `overwrite` | `bool` | `False` | Overwrite existing files |
| `dry_run` | `bool` | `False` | Preview output without writing files |
| `output_file` | `str` | `""` | Specific output file path |
| `output_dir` | `str` | `""` | Output directory (defaults to `ocp_resources`) |
| `add_tests` | `bool` | `False` | Generate test files |
| `called_from_cli` | `bool` | `True` | Enables interactive prompts when `True` |
| `update_schema_executed` | `bool` | `False` | Whether schema update was already performed |

**Returns:** `list[str]` — List of generated file paths. Empty list if the kind is not found in the schema mapping (when `called_from_cli=False`).

**Raises:**
- `RuntimeError` — Kind not found after schema update, or user declined schema update
- `ValueError` — Generated filename contains invalid patterns (single-letter segments)

---

### `class_generator.core.generator.generate_resource_file_from_dict()`

```python
from class_generator.core.generator import generate_resource_file_from_dict

orig_filename, generated_filename = generate_resource_file_from_dict(
    resource_dict={"kind": "Pod", ...},
    overwrite=False,
    dry_run=False,
    output_file="",
    add_tests=False,
    output_file_suffix="",
    output_dir="",
)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `resource_dict` | `dict[str, Any]` | *(required)* | Dictionary containing parsed resource information |
| `overwrite` | `bool` | `False` | Overwrite existing files |
| `dry_run` | `bool` | `False` | Preview without writing |
| `output_file` | `str` | `""` | Specific output file path |
| `add_tests` | `bool` | `False` | Generate test files under `class_generator/tests/manifests/` |
| `output_file_suffix` | `str` | `""` | Suffix appended to filename (for API group disambiguation) |
| `output_dir` | `str` | `""` | Output directory (defaults to `ocp_resources`) |

**Returns:** `tuple[str, str]` — `(original_filename, generated_filename)`. These differ when a `_TEMP.py` file is created.

---

### `class_generator.core.schema.update_kind_schema()`

```python
from class_generator.core.schema import update_kind_schema

update_kind_schema(client=None)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `client` | `str \| None` | `None` | Path to `oc`/`kubectl` binary. Auto-detected if `None`. |

**Raises:**
- `ClusterVersionError` — Cannot determine cluster version
- `RuntimeError` — Failed to fetch OpenAPI v3 index
- `OSError` — Failed to write schema files

---

### `class_generator.core.schema.update_single_resource_schema()`

```python
from class_generator.core.schema import update_single_resource_schema

update_single_resource_schema(kind="LlamaStackDistribution", client=None)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `kind` | `str` | *(required)* | Resource Kind name (case-sensitive) |
| `client` | `str \| None` | `None` | Path to `oc`/`kubectl` binary. Auto-detected if `None`. |

**Raises:**
- `ResourceNotFoundError` — Kind not found on the cluster
- `RuntimeError` — Schema extraction or save failed

---

### `class_generator.core.coverage.analyze_coverage()`

```python
from class_generator.core.coverage import analyze_coverage

result: dict[str, Any] = analyze_coverage(resources_dir="ocp_resources")
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `resources_dir` | `str` | `"ocp_resources"` | Directory to scan for wrapper classes |

**Returns:** `dict[str, Any]` with keys:

| Key | Type | Description |
|-----|------|-------------|
| `generated_resources` | `list[str]` | Sorted list of auto-generated resource class names |
| `manual_resources` | `list[str]` | Sorted list of manually written resource class names |
| `missing_resources` | `list[dict]` | List of `{"kind": "..."}` for resources in schema but not generated |
| `coverage_stats` | `dict` | Statistics including `total_in_mapping`, `total_generated`, `total_manual`, `coverage_percentage`, `missing_count` |

---

### `class_generator.core.discovery.discover_generated_resources()`

```python
from class_generator.core.discovery import discover_generated_resources

resources: list[dict[str, Any]] = discover_generated_resources()
```

**Returns:** `list[dict[str, Any]]` — Each dict contains:

| Key | Type | Description |
|-----|------|-------------|
| `path` | `str` | Full path to the resource file |
| `kind` | `str` | Resource class name |
| `filename` | `str` | File name without extension |
| `has_user_code` | `bool` | Whether file contains user modifications below `# End of generated code` |

---

### `class_generator.core.discovery.discover_cluster_resources()`

```python
from class_generator.core.discovery import discover_cluster_resources

resources: dict[str, list[dict[str, Any]]] = discover_cluster_resources(
    client=None,
    api_group_filter=None,
)
```

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `client` | `DynamicClient \| None` | `None` | Kubernetes dynamic client. Creates one if `None`. |
| `api_group_filter` | `str \| None` | `None` | Filter by API group name |

**Returns:** `dict[str, list[dict[str, Any]]]` — Mapping of API version to list of resource dicts (`name`, `kind`, `namespaced`).

**Raises:** `ValueError` — If client is not a `DynamicClient` instance.

---

## Exceptions

### `class_generator.exceptions.ResourceNotFoundError`

```python
from class_generator.exceptions import ResourceNotFoundError
```

Raised when a resource Kind is not found in the schema mapping or on the cluster.

| Attribute | Type | Description |
|-----------|------|-------------|
| `kind` | `str` | The Kind that was not found |

### `class_generator.core.schema.ClusterVersionError`

```python
from class_generator.core.schema import ClusterVersionError
```

Raised when the cluster version cannot be determined (client binary missing, cluster unreachable, or authentication failure).

---

## Schema Files

The CLI manages two schema files under `class_generator/schema/`:

| File | Purpose |
|------|---------|
| `__resources-mappings.json.gz` | Compressed JSON mapping of lowercase Kind → list of schemas with GVK metadata and namespaced status |
| `_definitions.json` | JSON Schema definitions for `$ref` resolution during validation |
| `__cluster_version__.txt` | Last cluster version used for schema generation |

See [Schema Validation and Code Generation Architecture](schema-validation-internals.html) for details on how these files are structured and consumed.

---

## Common Workflows

### Generate a new resource class

```bash
class-generator -k MyCustomResource
```

### Update schema and regenerate all classes

```bash
class-generator --update-schema
class-generator --regenerate-all --backup
```

### Add a new CRD to an older cluster

```bash
class-generator --update-schema-for MyNewCRD
class-generator -k MyNewCRD
```

### CI/CD coverage check

```bash
class-generator --coverage-report --json > coverage.json
```

### Preview changes before writing

```bash
class-generator -k Pod --dry-run
class-generator --regenerate-all --dry-run
```

---

## Related Pages

- [Generating New Resource Classes with class-generator](generating-resource-classes.html) — Step-by-step guide for scaffolding resource classes
- [Resource and NamespacedResource API](resource-api.html) — API reference for the base classes that generated code extends
- [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html) — How generated subclasses fit into the class hierarchy
- [Schema Validation and Code Generation Architecture](schema-validation-internals.html) — Internals of schema fetching, caching, and code generation
- [Environment Variables and Configuration](environment-variables.html) — Environment variables that affect runtime behavior

## Related Pages

- [Generating New Resource Classes with class-generator](generating-resource-classes.html)
- [Schema Validation and Code Generation Architecture](schema-validation-internals.html)
- [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html)
- [Resource and NamespacedResource API](resource-api.html)
- [Environment Variables and Configuration](environment-variables.html)

---

Source: environment-variables.md

Let me start by reading the knowledge graph and page manifest, then explore the codebase for environment variables.

Now let me look at the `keys_to_hash` and hash-related functionality more closely:

Now let me look at the `get_logger` import and how log levels work:

Now let me check for any CI/CD or config-related environment variables:

Let me also look for any validation-related environment variables and enable_validation_by_default:

Now let me look at the `get_logger` usage and the `simple_logger` format:

Now let me check the deploy() and clean_up() methods more carefully, and also the hash_resource_dict method:

Let me also check how `replace_key_with_hashed_value` is defined:

Now let me check the MCP server for environment variables:

Now let me check if there are any other environment variable usages in the class_generator:

Let me verify there are no other env vars in tests:

Let me also check the full `get_client` function and the `save_kubeconfig` function:

Now let me look for any additional environment variables in fake_kubernetes_client:

Let me look at how the deploy and clean_up methods use the skip logic:

Now I have all the information needed. Let me write the documentation:

# Environment Variables and Configuration

This page documents all environment variables recognized by `openshift-python-wrapper` at runtime. These variables control logging behavior, resource lifecycle operations, proxy configuration, cluster connectivity, and sensitive data hashing.

## Quick Reference

| Environment Variable | Purpose | Default |
|---|---|---|
| `KUBECONFIG` | Path to kubeconfig file | `~/.kube/config` |
| `HTTPS_PROXY` | HTTPS proxy URL for cluster connections | _(unset)_ |
| `HTTP_PROXY` | HTTP proxy URL for cluster connections | _(unset)_ |
| `OPENSHIFT_PYTHON_WRAPPER_LOG_LEVEL` | Logging verbosity level | `INFO` |
| `OPENSHIFT_PYTHON_WRAPPER_LOG_FILE` | Path to log output file | `""` (stdout) |
| `OPENSHIFT_PYTHON_WRAPPER_HASH_LOG_DATA` | Enable/disable hashing sensitive data in logs | `true` |
| `REUSE_IF_RESOURCE_EXISTS` | Skip resource creation if already exists | _(unset)_ |
| `SKIP_RESOURCE_TEARDOWN` | Skip resource deletion during teardown | _(unset)_ |

---

## Cluster Connection

### `KUBECONFIG`

| Property | Value |
|---|---|
| **Type** | `str` |
| **Default** | `~/.kube/config` |
| **Used by** | `get_client()` |
| **Source** | `ocp_resources/resource.py` |

Path to the kubeconfig file used when creating a Kubernetes `DynamicClient`. Read when no explicit `config_file`, `config_dict`, `host`/`token`, or `username`/`password` arguments are passed to `get_client()`.

> **Note:** The standard `kubernetes` Python client reads `KUBECONFIG` at import time. If you set this variable in code (after import), `openshift-python-wrapper` handles it by explicitly passing the value to the client constructor. See [Connecting to Clusters](connecting-to-clusters.html) for all connection options.

```bash
export KUBECONFIG=/path/to/my/kubeconfig
```

```python
from ocp_resources.resource import get_client

# Automatically uses $KUBECONFIG, or falls back to ~/.kube/config
client = get_client()
```

---

### `HTTPS_PROXY` / `HTTP_PROXY`

| Property | Value |
|---|---|
| **Type** | `str` |
| **Default** | _(unset — no proxy)_ |
| **Used by** | `get_client()` |
| **Source** | `ocp_resources/resource.py` |

Sets the proxy on the Kubernetes client configuration when no proxy is already configured on the `client_configuration` object. `HTTPS_PROXY` takes precedence over `HTTP_PROXY` when both are set.

```bash
export HTTPS_PROXY=http://proxy.example.com:8080
```

```python
from ocp_resources.resource import get_client

# Proxy is automatically applied from environment
client = get_client()
```

> **Tip:** If you pass a `client_configuration` object that already has a `.proxy` set, the environment variables are ignored.

**Precedence order:**

1. Explicit `client_configuration.proxy` (if already set)
2. `HTTPS_PROXY` environment variable
3. `HTTP_PROXY` environment variable

---

## Logging

### `OPENSHIFT_PYTHON_WRAPPER_LOG_LEVEL`

| Property | Value |
|---|---|
| **Type** | `str` |
| **Default** | `INFO` |
| **Valid values** | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` |
| **Used by** | `Resource._set_logger()` |
| **Source** | `ocp_resources/resource.py` |

Controls the log level for each resource instance's logger. The logger is created per-resource using the `simple_logger` library.

```bash
export OPENSHIFT_PYTHON_WRAPPER_LOG_LEVEL=DEBUG
```

```python
import os
os.environ["OPENSHIFT_PYTHON_WRAPPER_LOG_LEVEL"] = "DEBUG"

from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()
pod = Pod(name="my-pod", namespace="default", client=client)
# Logger on this Pod instance now uses DEBUG level
```

---

### `OPENSHIFT_PYTHON_WRAPPER_LOG_FILE`

| Property | Value |
|---|---|
| **Type** | `str` |
| **Default** | `""` (empty string — logs to stdout) |
| **Used by** | `Resource._set_logger()` |
| **Source** | `ocp_resources/resource.py` |

Redirects resource log output to a file. When unset or empty, logs are written to stdout.

```bash
export OPENSHIFT_PYTHON_WRAPPER_LOG_FILE=/var/log/ocp-wrapper.log
```

```python
import os
os.environ["OPENSHIFT_PYTHON_WRAPPER_LOG_FILE"] = "/tmp/ocp-wrapper.log"

from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()
ns = Namespace(name="test-ns", client=client)
# All log output for this resource is written to /tmp/ocp-wrapper.log
```

---

## Sensitive Data Hashing

### `OPENSHIFT_PYTHON_WRAPPER_HASH_LOG_DATA`

| Property | Value |
|---|---|
| **Type** | `str` |
| **Default** | `"true"` |
| **Valid values** | `"true"`, `"false"` |
| **Used by** | `Resource.hash_resource_dict()` |
| **Source** | `ocp_resources/resource.py` |

Controls whether sensitive fields in resource dictionaries are replaced with `"*******"` when logged. When set to `"false"`, sensitive data is logged in cleartext.

Hashing is applied to fields declared in each resource class's `keys_to_hash` property. The following built-in resources define sensitive keys:

| Resource Class | Hashed Fields |
|---|---|
| `Secret` | `data`, `stringData` |
| `ConfigMap` | `data`, `binaryData` |
| `SealedSecret` | `spec>data`, `spec>encryptedData` |
| `VirtualMachine` | `spec>template>spec>volumes[]>cloudInitNoCloud>userData` |

> **Warning:** Setting this to `"false"` causes sensitive data (e.g., secrets, tokens) to appear in log output. Use only for debugging in secure environments.

```bash
# Disable hashing to see raw values in logs (debug only)
export OPENSHIFT_PYTHON_WRAPPER_HASH_LOG_DATA=false
```

```python
import os
os.environ["OPENSHIFT_PYTHON_WRAPPER_HASH_LOG_DATA"] = "false"

from ocp_resources.secret import Secret
from ocp_resources.resource import get_client

client = get_client()
secret = Secret(name="my-secret", namespace="default", client=client)
# Logs will now show raw secret data instead of "*******"
```

> **Note:** Hashing also depends on the `hash_log_data` constructor parameter on each resource instance. Both must be enabled for hashing to occur. The `hash_log_data` parameter defaults to `True`. See [Resource and NamespacedResource API](resource-api.html) for constructor details.

**Interaction with `hash_log_data` parameter:**

| `OPENSHIFT_PYTHON_WRAPPER_HASH_LOG_DATA` | `hash_log_data` param | Result |
|---|---|---|
| `"true"` (default) | `True` (default) | Sensitive fields hashed |
| `"true"` | `False` | Sensitive fields **not** hashed |
| `"false"` | `True` | Sensitive fields **not** hashed |
| `"false"` | `False` | Sensitive fields **not** hashed |

---

## Resource Reuse (Skip Creation)

### `REUSE_IF_RESOURCE_EXISTS`

| Property | Value |
|---|---|
| **Type** | `str` (YAML dict syntax) |
| **Default** | _(unset — no resources skipped)_ |
| **Used by** | `Resource.deploy()` |
| **Source** | `ocp_resources/resource.py` |

When set, `deploy()` checks whether the target resource already exists on the cluster. If a match is found, the existing resource is returned without creating a new one. This is intended for debugging and iterative development workflows.

> **Warning:** Spaces are significant in the value syntax. The value is parsed as YAML.

**Value format:**

```
{<Kind>: {<name>: <namespace>}}
```

**Matching rules:**

| Pattern | Behavior |
|---|---|
| `{Pod: {}}` | Skip creation for **all** Pods (match by kind only) |
| `{Pod: {my-pod:}}` | Skip creation for Pod named `my-pod` in **any** namespace |
| `{Pod: {my-pod: my-ns}}` | Skip creation for Pod named `my-pod` in namespace `my-ns` only |
| `{Kind1: {}, Kind2: {name: ns}}` | Multiple resource patterns combined |

**Examples:**

```bash
# Skip all Pod creation if the pod already exists
export REUSE_IF_RESOURCE_EXISTS="{Pod: {}}"

# Skip specific pod in specific namespace
export REUSE_IF_RESOURCE_EXISTS="{Pod: {my-pod: my-namespace}}"

# Skip namespace and pod creation
export REUSE_IF_RESOURCE_EXISTS="{Namespace: {test-ns:}, Pod: {my-pod: test-ns}}"
```

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

# If REUSE_IF_RESOURCE_EXISTS is set and a matching Pod exists,
# deploy() returns the existing resource without calling create()
pod = Pod(
    name="my-pod",
    namespace="my-namespace",
    client=client,
    containers=[{"name": "test", "image": "nginx"}],
)
pod.deploy()  # Skips creation if resource matches the env var pattern
```

> **Note:** The resource must actually exist on the cluster for the skip to take effect. If the resource does not exist, `deploy()` proceeds with normal creation.

---

## Skip Teardown

### `SKIP_RESOURCE_TEARDOWN`

| Property | Value |
|---|---|
| **Type** | `str` (YAML dict syntax) |
| **Default** | _(unset — no resources skipped)_ |
| **Used by** | `Resource.clean_up()` |
| **Source** | `ocp_resources/resource.py` |

When set, `clean_up()` skips deletion for matching resources and returns `True` without calling `delete()`. This is intended for debugging — preserving resources on the cluster after tests finish.

> **Warning:** Spaces are significant in the value syntax. The value is parsed as YAML.

**Value format:**

Uses the same YAML dict syntax as [`REUSE_IF_RESOURCE_EXISTS`](#reuse_if_resource_exists).

**Matching rules:**

| Pattern | Behavior |
|---|---|
| `{Pod: {}}` | Skip teardown for **all** Pods |
| `{Pod: {my-pod:}}` | Skip teardown for Pod named `my-pod` in **any** namespace |
| `{Pod: {my-pod: my-ns}}` | Skip teardown for Pod named `my-pod` in namespace `my-ns` only |
| `{Kind1: {}, Kind2: {name: ns}}` | Multiple resource patterns combined |

**Examples:**

```bash
# Keep all Namespaces after test run
export SKIP_RESOURCE_TEARDOWN="{Namespace: {}}"

# Keep specific resources
export SKIP_RESOURCE_TEARDOWN="{Namespace: {test-ns:}, Pod: {debug-pod: test-ns}}"
```

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()
pod = Pod(
    name="debug-pod",
    namespace="test-ns",
    client=client,
    containers=[{"name": "test", "image": "nginx"}],
)
pod.deploy()

# If SKIP_RESOURCE_TEARDOWN is set with a matching pattern,
# clean_up() returns True without deleting the resource
pod.clean_up()  # Resource remains on the cluster
```

> **Tip:** Use `REUSE_IF_RESOURCE_EXISTS` and `SKIP_RESOURCE_TEARDOWN` together for a fast debug loop: skip creation if a resource already exists, and skip teardown so it persists between runs.

```bash
export REUSE_IF_RESOURCE_EXISTS="{Pod: {debug-pod: test-ns}}"
export SKIP_RESOURCE_TEARDOWN="{Pod: {debug-pod: test-ns}}"
```

---

## Combined Usage Patterns

### Debug iteration loop

Set both skip variables to avoid re-creating and tearing down resources between test runs:

```bash
export REUSE_IF_RESOURCE_EXISTS="{Namespace: {test-ns:}, Pod: {my-pod: test-ns}}"
export SKIP_RESOURCE_TEARDOWN="{Namespace: {test-ns:}, Pod: {my-pod: test-ns}}"
```

### Full debug logging with unhashed data

```bash
export OPENSHIFT_PYTHON_WRAPPER_LOG_LEVEL=DEBUG
export OPENSHIFT_PYTHON_WRAPPER_LOG_FILE=/tmp/debug.log
export OPENSHIFT_PYTHON_WRAPPER_HASH_LOG_DATA=false
```

> **Warning:** This combination logs sensitive data in plaintext. Do not use in production or shared environments.

### Custom kubeconfig with proxy

```bash
export KUBECONFIG=/path/to/custom/kubeconfig
export HTTPS_PROXY=http://corporate-proxy.example.com:3128
```

---

## Programmatic Configuration Alternatives

Several behaviors controlled by environment variables can also be set programmatically via constructor parameters or function arguments. See [Resource and NamespacedResource API](resource-api.html) for full constructor parameter documentation.

| Environment Variable | Programmatic Alternative |
|---|---|
| `KUBECONFIG` | `get_client(config_file="/path/to/kubeconfig")` |
| `HTTPS_PROXY` / `HTTP_PROXY` | `client_configuration.proxy = "http://proxy:port"` |
| `OPENSHIFT_PYTHON_WRAPPER_HASH_LOG_DATA` | `Resource(hash_log_data=False, ...)` constructor parameter |

> **Note:** `OPENSHIFT_PYTHON_WRAPPER_LOG_LEVEL`, `OPENSHIFT_PYTHON_WRAPPER_LOG_FILE`, `REUSE_IF_RESOURCE_EXISTS`, and `SKIP_RESOURCE_TEARDOWN` can **only** be configured via environment variables.

## Related Pages

- [Connecting to Clusters](connecting-to-clusters.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Resource and NamespacedResource API](resource-api.html)
- [Common Resource Patterns](common-patterns.html)
- [Error Handling and Exception Reference](error-handling.html)

---

Source: resource-class-hierarchy.md

Let me start by reading the exclusions file and pages manifest, then explore the codebase thoroughly.

Now let me look at some generated resource subclasses to understand how they extend the base classes:

Now let me look at some more examples - a cluster-scoped resource, the `ResourceConstants`, the class generator, and the `virtual_machine.py` for a more complex subclass:

Now let me look at the `kind` property and how it's derived, plus some other resources for variety:

Now let me look at how the `kind` property works (the `sub_resource_level` function) and the `conversion.py` for the kind name derivation:

Now let me look at how resources discover their API version dynamically:

Let me now look at one more important aspect - how the `get` method works differently in the two base classes, and how `_base_body` differs:

Now let me check the `__init__.py` to see what's exported:

Let me also check for examples:

Now I have enough context to write the documentation. Let me draft it:

# Understanding the Resource Class Hierarchy

Every Kubernetes and OpenShift object you interact with through openshift-python-wrapper — whether it's a Pod, a Namespace, or a VirtualMachine — is a Python class that inherits from a small set of base classes. Understanding how these classes relate to each other helps you know **what capabilities any resource has**, **why some resources require a `namespace` argument and others don't**, and **how new resource types get added to the library**.

This page explains the architecture so you can use existing resources effectively and extend the library when you need to work with a new custom resource.

---

## The Big Picture

The class hierarchy has three conceptual layers:

| Layer | Classes | Purpose |
|---|---|---|
| **Constants** | `ResourceConstants` | Provides shared status strings, condition names, and type constants used across all resources |
| **Base classes** | `Resource`, `NamespacedResource` | Define all CRUD operations, lifecycle management, waiting, validation, and API discovery logic |
| **Concrete resource classes** | `Pod`, `Namespace`, `Deployment`, `VirtualMachine`, and 200+ others | Represent specific Kubernetes/OpenShift resource kinds, adding kind-specific parameters and behavior |

The inheritance flows like this:

```
ResourceConstants
  └── Resource                  ← cluster-scoped resources (Node, Namespace, StorageClass, ClusterRole, …)
        └── NamespacedResource  ← namespace-scoped resources (Pod, Deployment, ConfigMap, Secret, Route, …)
```

Every concrete resource class inherits from either `Resource` (for cluster-scoped resources) or `NamespacedResource` (for namespace-scoped resources). This single design decision controls whether the class requires a `namespace` argument and how it builds API requests.

---

## Key Concepts

### ResourceConstants: Shared Vocabulary

At the root of the hierarchy sits `ResourceConstants`, which defines inner classes for common values:

- **`Status`** — Strings like `RUNNING`, `SUCCEEDED`, `FAILED`, `PENDING`, `ACTIVE`
- **`Condition`** — Condition types like `READY`, `AVAILABLE`, `DEGRADED` and their status values (`TRUE`, `FALSE`, `UNKNOWN`)
- **`Type`** — Service types like `ClusterIP`, `NodePort`, `LoadBalancer`

Because `Resource` inherits from `ResourceConstants`, every resource class in the library can reference these constants:

```python
from ocp_resources.namespace import Namespace

ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
```

Concrete classes can extend these constants with kind-specific values. For example, `VirtualMachine` adds statuses like `MIGRATING`, `STOPPED`, and `PROVISIONING` to the base `Status` class.

### Resource: The Foundation for Cluster-Scoped Resources

`Resource` is the main base class. It provides everything needed to manage a Kubernetes resource:

| Capability | Methods / Properties |
|---|---|
| **CRUD operations** | `create()`, `delete()`, `update()`, `update_replace()` |
| **Lifecycle management** | `deploy()`, `clean_up()`, context manager (`with` statement) |
| **Querying** | `exists`, `instance`, `status`, `labels` |
| **Waiting** | `wait()`, `wait_deleted()`, `wait_for_status()`, `wait_for_condition()` |
| **Listing** | `get()` class method — yields resource objects matching filters |
| **Validation** | `validate()`, `validate_dict()` |
| **Serialization** | `to_dict()`, `to_yaml()` |
| **API discovery** | Automatic `api_version` resolution from the cluster when only `api_group` is set |

Resources that exist at the cluster level — not inside any namespace — inherit directly from `Resource`. Examples include `Namespace`, `Node`, `StorageClass`, `ClusterRole`, and `CustomResourceDefinition`.

```python
from ocp_resources.namespace import Namespace

# No namespace argument needed — Namespace is cluster-scoped
ns = Namespace(client=client, name="my-namespace")
```

#### Automatic Kind Detection

You never set the `kind` field manually. The `kind` property is a class-level property that automatically derives the Kubernetes kind name from the **class name** using Python's method resolution order (MRO). When you define a class named `StorageClass`, its `kind` is automatically `"StorageClass"`.

#### API Version Discovery

Resources can specify their API identity in two ways:

1. **`api_version`** — A fixed version string (e.g., `"v1"`), used for core Kubernetes resources
2. **`api_group`** — An API group string (e.g., `"apps"`, `"kubevirt.io"`), where the full `apiVersion` is discovered dynamically from the cluster

When only `api_group` is set, the library queries the cluster at runtime to find the latest supported version for that resource kind. This means resource classes automatically work across cluster versions without code changes.

```python
class Namespace(Resource):
    # Core resource — fixed version, no group
    api_version: str = Resource.ApiVersion.V1

class ClusterRole(Resource):
    # Grouped resource — version discovered from cluster
    api_group = Resource.ApiGroup.RBAC_AUTHORIZATION_K8S_IO
```

> **Note:** The `Resource.ApiGroup` and `Resource.ApiVersion` inner classes provide predefined constants for all known API groups and versions. Using these constants avoids typos and makes your code self-documenting.

### NamespacedResource: Adding Namespace Awareness

`NamespacedResource` extends `Resource` with one critical addition: **namespace handling**. It requires a `namespace` argument during construction and injects the namespace into all API calls.

The differences from `Resource` are focused but important:

| Behavior | `Resource` | `NamespacedResource` |
|---|---|---|
| `namespace` required? | No | Yes (unless using `yaml_file` or `kind_dict`) |
| `to_dict()` output | No namespace in metadata | Adds `metadata.namespace` |
| `get()` yields | Objects with `name` only | Objects with both `name` and `namespace` |
| `instance` property | Fetches by name only | Fetches by name and namespace |

```python
from ocp_resources.pod import Pod

# Namespace is required for namespaced resources
pod = Pod(client=client, name="my-pod", namespace="default",
          containers=[{"name": "app", "image": "nginx"}])
```

### Concrete Resource Classes: Where Specifics Live

Concrete classes add three things on top of the base classes:

1. **API identity** — Setting `api_group` or `api_version` to identify which Kubernetes API to call
2. **Constructor parameters** — Typed arguments for the resource's spec fields (like `containers` for Pod, `replicas` for Deployment)
3. **Custom `to_dict()` method** — Builds the Kubernetes resource dictionary from constructor arguments

Here is how a typical generated class is structured:

```python
class Deployment(NamespacedResource):
    # 1. API identity
    api_group: str = NamespacedResource.ApiGroup.APPS

    def __init__(self, replicas=None, selector=None, template=None, **kwargs):
        # 2. Pass common args to base class, store kind-specific args
        super().__init__(**kwargs)
        self.replicas = replicas
        self.selector = selector
        self.template = template

    def to_dict(self):
        # 3. Build the resource dictionary
        super().to_dict()
        if not self.kind_dict and not self.yaml_file:
            self.res["spec"] = {}
            _spec = self.res["spec"]
            _spec["selector"] = self.selector
            _spec["template"] = self.template
            if self.replicas is not None:
                _spec["replicas"] = self.replicas
```

> **Tip:** You can always bypass the typed constructor entirely by passing `yaml_file` or `kind_dict` to any resource class. When you do, the `to_dict()` logic is skipped and the resource is created from your raw definition instead.

#### Adding Kind-Specific Behavior

Many concrete classes go beyond what the generator produces by adding custom methods and properties. For example:

- **Pod** adds `execute()` for running commands, `log()` for reading logs, and a `node` property
- **Deployment** adds `scale_replicas()` and `wait_for_replicas()`
- **Secret** overrides `keys_to_hash` to ensure sensitive data is masked in logs

These additions are preserved across regeneration because the class generator recognizes the `# End of generated code` marker and keeps any code written below it.

### How Generated Classes Are Created

Most concrete resource classes in the library are **code-generated** from the cluster's OpenAPI schema using the `class-generator` tool. The generator:

1. Reads the resource definition from the cluster's OpenAPI schema
2. Determines whether the resource is namespaced (→ `NamespacedResource`) or cluster-scoped (→ `Resource`)
3. Extracts spec fields with their types and descriptions
4. Renders a Python class from a Jinja2 template
5. Preserves any hand-written code below the `# End of generated code` marker

This means the hierarchy is not just an architectural choice — it is **enforced by the code generation pipeline**. Every generated class correctly inherits from the appropriate base class based on the resource's actual scope in Kubernetes.

See [Generating New Resource Classes with class-generator](generating-resource-classes.html) for details on generating classes for new resource types.

---

## How It Affects You

Understanding the hierarchy helps you in several practical ways:

### Knowing What Methods Are Available

Every resource — regardless of kind — inherits the full set of CRUD, waiting, and lifecycle methods from `Resource`. You don't need to check whether a particular resource supports `wait_for_condition()` or context managers; they all do.

```python
# Works for any resource type
with SomeResource(client=client, name="example", **specific_args) as res:
    res.wait_for_condition(condition="Ready", status="True")
```

See [Resource and NamespacedResource API](resource-api.html) for the complete method reference.

### Understanding Constructor Requirements

The base class determines what arguments are mandatory:

| If you're using... | Required arguments |
|---|---|
| `Resource` subclass | `client`, `name` |
| `NamespacedResource` subclass | `client`, `name`, `namespace` |
| Any class with `yaml_file` | `client`, `yaml_file` |
| Any class with `kind_dict` | `client`, `kind_dict` |

See [Creating and Managing Resources](creating-and-managing-resources.html) for full examples of all creation methods.

### Using Status and Condition Constants

The `Status` and `Condition` constants inherited from `ResourceConstants` are available on every resource class. Concrete classes may extend them:

```python
from ocp_resources.virtual_machine import VirtualMachine

# Base constants work on all resources
vm.wait_for_status(status=VirtualMachine.Status.RUNNING)

# Kind-specific constants are added by the concrete class
vm.wait_for_status(status=VirtualMachine.Status.STOPPED)
```

### Extending the Library

If you need to work with a CRD that isn't included in the library, you have two options:

1. **Use the class generator** to scaffold a new class automatically — see [Generating New Resource Classes with class-generator](generating-resource-classes.html)
2. **Write a class manually** by inheriting from `Resource` or `NamespacedResource` and setting `api_group` or `api_version`

A minimal hand-written resource class looks like this:

```python
from ocp_resources.resource import NamespacedResource

class MyCustomResource(NamespacedResource):
    api_group = "example.com"
    # That's it — you get full CRUD, waiting, and lifecycle support
```

> **Warning:** Your class name must match the Kubernetes `kind` exactly (in PascalCase). The `kind` property is derived from the class name automatically — a class named `MyCustomResource` will have `kind = "MyCustomResource"`.

---

## Related Pages

- [Resource and NamespacedResource API](resource-api.html) — Complete method and property reference for the base classes
- [Creating and Managing Resources](creating-and-managing-resources.html) — Practical guide to creating, updating, and deleting resources
- [Generating New Resource Classes with class-generator](generating-resource-classes.html) — Generate Python classes for any CRD
- [Querying and Watching Resources](querying-resources.html) — Use the `get()` class method and watchers to list and observe resources
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html) — Use the waiting methods inherited from the base classes
- [Validating Resources Against OpenAPI Schemas](validating-resources.html) — Validate resources using the built-in schema validation
- [Environment Variables and Configuration](environment-variables.html) — Configure logging, resource reuse, and teardown behavior

## Related Pages

- [Resource and NamespacedResource API](resource-api.html)
- [Generating New Resource Classes with class-generator](generating-resource-classes.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Schema Validation and Code Generation Architecture](schema-validation-internals.html)
- [Validating Resources Against OpenAPI Schemas](validating-resources.html)

---

Source: schema-validation-internals.md

Let me start by reading the knowledge graph and the pages manifest, then explore the relevant source files.

Now let me look at some more key files - the explain parser, resource base class validation, archive utils, and some test files:

Now let me look at the exceptions module for validation:

Now let me check the CI/CD configuration to understand how schemas are updated:

Let me look at a few generated resource files to understand the output of the code generation:

Let me also check the parsers for status schema generation:

Now let me check what data is actually in the schema file to understand the structure:

Now I have enough information. Let me write the documentation page.

# Schema Validation and Code Generation Architecture

The openshift-python-wrapper uses OpenAPI schemas as the single source of truth for two critical capabilities: **validating resource definitions** before they reach the cluster, and **generating Python wrapper classes** for any Kubernetes or OpenShift resource type. Understanding this architecture helps you diagnose validation errors, generate classes for custom resources, and keep your schemas up to date as clusters evolve.

## The Big Picture

OpenAPI schemas flow through the system in two directions: they are fetched from a live cluster and stored locally, then consumed by the validation engine at runtime and by the code generator at development time.

| Component | Location | Purpose |
|---|---|---|
| **Schema Fetcher** | `class_generator/core/schema.py` | Connects to a cluster, downloads OpenAPI v3 schemas, and writes them to disk |
| **Resource Mappings** | `class_generator/schema/__resources-mappings.json.gz` | Compressed archive mapping each resource kind (lowercase) to its schema(s) |
| **Definitions File** | `class_generator/schema/_definitions.json` | JSON file containing detailed schema definitions with `$ref` targets for nested types |
| **Schema Validator** | `ocp_resources/utils/schema_validator.py` | Runtime engine that loads schemas, resolves `$ref` references, and validates resource dicts |
| **Code Generator** | `class_generator/core/generator.py` | Reads schemas and generates Python class files from a Jinja2 template |
| **Explain Parser** | `class_generator/parsers/explain_parser.py` | Parses the resource mapping to extract fields, types, and group-version-kind metadata |
| **Cluster Version Tracker** | `class_generator/schema/__cluster_version__.txt` | Tracks the cluster version that last produced the schema, controlling update strategy |

### Data Flow: Schema Fetch and Storage

1. **Detect client binary** — The system finds `oc` or falls back to `kubectl` via `get_client_binary()`.
2. **Check cluster version** — `check_and_update_cluster_version()` compares the connected cluster's version against the stored version in `__cluster_version__.txt`.
3. **Determine update strategy** — If the cluster is the same version or newer, all schemas are fetched and existing entries are updated. If older, only missing resources are fetched and existing schemas are preserved.
4. **Fetch OpenAPI v3 index** — `GET /openapi/v3` returns an index of all API group paths (e.g., `api/v1`, `apis/apps/v1`).
5. **Fetch schemas in parallel** — `fetch_all_api_schemas()` downloads schemas from each API path using a thread pool (up to 10 workers).
6. **Build namespacing dictionary** — `build_namespacing_dict()` queries `api-resources` to determine which kinds are namespaced vs. cluster-scoped.
7. **Process definitions** — `process_schema_definitions()` extracts `x-kubernetes-group-version-kind` metadata, builds schema entries, and merges them into the resource mappings. Existing resources are **never deleted** — only added or updated.
8. **Supplement with `oc explain`** — Missing field descriptions, required-field markers, and `$ref` definitions are filled in by running `oc explain` commands in parallel.
9. **Write to disk** — The resource mappings are saved as a gzip-compressed JSON archive (`__resources-mappings.json.gz`), and definitions are written to `_definitions.json`.

### Data Flow: Validation at Runtime

1. **Load on first use** — `SchemaValidator.load_mappings_data()` decompresses and loads the mappings archive and definitions file into class-level caches.
2. **Look up by kind** — `SchemaValidator.load_schema(kind, api_group)` finds the schema for a resource kind (case-insensitive). When multiple API groups define the same kind (e.g., `Ingress` in `networking.k8s.io` and `config.openshift.io`), the `api_group` parameter disambiguates.
3. **Resolve `$ref` references** — The validator recursively resolves all `$ref` pointers against the definitions data, producing a self-contained schema.
4. **Cache resolved schemas** — Resolved schemas are stored in `_schema_cache` keyed by `api_group:kind`, so subsequent validations are fast.
5. **Validate with jsonschema** — `jsonschema.validate()` checks the resource dictionary against the resolved schema.
6. **Format errors** — If validation fails, `format_validation_error()` produces a human-readable message including the field path, error details, and schema context.

### Data Flow: Code Generation

1. **Read resource mappings** — `parse_explain()` loads the mapping file and finds all schema entries for the requested kind.
2. **Select latest API version** — When multiple versions exist within an API group, the latest version is selected using a priority ranking (`v2 > v1 > v1beta2 > v1beta1 > v1alpha2 > v1alpha1`).
3. **Extract fields and types** — The `type_parser` module converts OpenAPI types to Python type annotations and builds parameter dictionaries for the class constructor.
4. **Render Jinja2 template** — `render_jinja_template()` processes `class_generator_template.j2` with the extracted data, producing a complete Python class.
5. **Preserve user code** — If the target file already exists, `parse_user_code_from_file()` extracts any user-added code below the `# End of generated code` marker and re-inserts it after regeneration.
6. **Format and write** — `write_and_format_rendered()` writes the file, then runs `prek` (pre-commit hooks) or falls back to `ruff` for formatting.

## Key Concepts

### The Resource Mappings File

The resource mappings file (`__resources-mappings.json.gz`) is the central schema store. It maps lowercase kind names to arrays of schema objects:

```json
{
  "pod": [
    {
      "description": "Pod is a collection of containers...",
      "properties": { ... },
      "required": [],
      "type": "object",
      "x-kubernetes-group-version-kind": [
        { "group": "", "kind": "Pod", "version": "v1" }
      ],
      "namespaced": true
    }
  ],
  "ingress": [
    { "x-kubernetes-group-version-kind": [{ "group": "networking.k8s.io", ... }], ... },
    { "x-kubernetes-group-version-kind": [{ "group": "config.openshift.io", ... }], ... }
  ]
}
```

Resources with the same kind but different API groups (like `Ingress` above) are stored as separate entries in the array. This structure allows the validator and generator to handle API group disambiguation correctly.

> **Note:** The mappings file is compressed with gzip to reduce repository size. The `archive_utils` module handles transparent compression and decompression via `save_json_archive()` and `load_json_archive()`.

### The Definitions File

The definitions file (`_definitions.json`) contains detailed schema definitions keyed by `group/version/Kind` paths (e.g., `apps/v1/Deployment`, `v1/Pod`). It also stores referenced sub-schemas like `io.k8s.api.core.v1.PodSpec` that are targets of `$ref` pointers. During validation, the `SchemaValidator` uses this file as the reference store for resolving nested types.

### SchemaValidator

`SchemaValidator` is a class with only class-level methods and caches — you never need to instantiate it. It serves as the shared validation engine used by both `Resource.validate()` and `Resource.validate_dict()`.

```python
from ocp_resources.utils.schema_validator import SchemaValidator
```

| Method | Signature | Description |
|---|---|---|
| `load_mappings_data` | `(skip_cache: bool = False) -> bool` | Loads the mappings archive and definitions file. Returns `True` on success. |
| `get_mappings_data` | `(skip_cache: bool = False) -> dict[str, Any] \| None` | Returns loaded mappings data, loading first if needed. |
| `get_definitions_data` | `() -> dict[str, Any] \| None` | Returns loaded definitions data, loading first if needed. |
| `load_schema` | `(kind: str, api_group: str \| None = None) -> dict[str, Any] \| None` | Loads and resolves a complete schema for a kind. Handles API group disambiguation and caching. |
| `validate` | `(resource_dict: dict[str, Any], kind: str, api_group: str \| None = None) -> None` | Validates a resource dict. Raises `jsonschema.ValidationError` on failure. |
| `format_validation_error` | `(error, kind, name, api_group=None) -> str` | Formats a validation error into a user-friendly message with field path and context. |
| `clear_cache` | `() -> None` | Clears the resolved schema cache (not the raw mappings/definitions data). |

> **Tip:** The `load_schema` method resolves `$ref` references recursively. For core Kubernetes types like `ObjectMeta` or `TypeMeta` that may be missing from definitions, it falls back to a permissive `{"type": "object", "additionalProperties": true}` schema rather than failing outright.

### Resource Validation Methods

The `Resource` base class exposes two validation methods that delegate to `SchemaValidator`:

```python
from ocp_resources.resource import Resource

# Instance-level validation — validates self.res
resource.validate()

# Class-level validation — validates any dict against the class's schema
MyResource.validate_dict(resource_dict)
```

Both raise `ocp_resources.exceptions.ValidationError` (not the raw `jsonschema.ValidationError`) with formatted error messages that include the resource identifier, field path, and details.

**Auto-validation** can be enabled per-instance by passing `schema_validation_enabled=True` to the constructor:

```python
from ocp_resources.pod import Pod

pod = Pod(
    name="my-pod",
    namespace="default",
    containers=[{"name": "app", "image": "nginx"}],
    client=client,
    schema_validation_enabled=True,
)
# validation runs automatically before create() and update_replace()
pod.deploy()
```

When auto-validation is enabled:
- `create()` calls `self.validate()` after `to_dict()` builds the resource dictionary
- `update_replace()` calls `validate_dict()` on the replacement resource dictionary
- `update()` (patch operations) does **not** trigger validation because patches are partial documents that may not pass full schema validation

### Schema Update Strategies

The `update_kind_schema()` function employs a version-aware update strategy:

| Cluster Version | Behavior |
|---|---|
| **Same or newer** than last recorded | Fetches **all** API schemas, updates existing resources, supplements with `oc explain` data |
| **Older** than last recorded | Identifies only **missing** resources via `identify_missing_resources()`, fetches only relevant API paths, does **not** modify existing schemas |

This strategy prevents an older cluster from overwriting schemas that contain richer data from a newer cluster.

> **Warning:** If you connect to an older cluster and need to update a specific CRD's schema (for example, after installing a new operator), use `update_single_resource_schema(kind)` or the CLI flag `--update-schema-for <Kind>`. This bypasses the version guard for that one resource.

### Single-Resource Schema Updates

The `update_single_resource_schema()` function fetches the schema for exactly one resource kind without affecting other resources in the mapping:

```python
from class_generator.core.schema import update_single_resource_schema

update_single_resource_schema(kind="LlamaStackDistribution")
```

| Parameter | Type | Description |
|---|---|---|
| `kind` | `str` | The resource Kind (case-sensitive, e.g., `"LlamaStackDistribution"`) |
| `client` | `str \| None` | Client binary path. Auto-detected if `None`. |

**Raises:**
- `ResourceNotFoundError` — if the kind is not found on the cluster
- `RuntimeError` — if schema fetching fails

### Code Generation Pipeline

The `class_generator()` function orchestrates the full generation pipeline:

```python
from class_generator.core.generator import class_generator

generated_files = class_generator(kind="Deployment", overwrite=True)
```

| Parameter | Type | Default | Description |
|---|---|---|---|
| `kind` | `str` | (required) | Kubernetes resource kind |
| `overwrite` | `bool` | `False` | Overwrite existing files |
| `dry_run` | `bool` | `False` | Print output without writing |
| `output_file` | `str` | `""` | Specific output file path |
| `output_dir` | `str` | `""` | Output directory (defaults to `ocp_resources`) |
| `add_tests` | `bool` | `False` | Generate test files in `class_generator/tests/manifests/` |

**Returns:** `list[str]` — paths of generated files.

**Raises:**
- `RuntimeError` — if the kind is not found in schema mappings
- `ValueError` — if the generated filename contains invalid patterns

When a kind exists in multiple API groups (e.g., `Ingress` in both `networking.k8s.io` and `config.openshift.io`), the generator produces **separate files** with a group-based suffix (e.g., `ingress_networking_k8s_io.py`, `ingress_config_openshift_io.py`).

### Generated Class Structure

Every generated file follows a consistent structure produced by the Jinja2 template:

```python
# Generated using https://github.com/RedHatQE/openshift-python-wrapper/blob/main/class_generator/README.md

from typing import Any
from ocp_resources.resource import NamespacedResource

class Pod(NamespacedResource):
    """Pod is a collection of containers that can run on a host."""

    api_version: str = NamespacedResource.ApiVersion.V1

    def __init__(self, containers: list[Any] | None = None, ..., **kwargs: Any) -> None:
        """Args:
            containers (list[Any]): List of containers belonging to the pod.
            ...
        """
        super().__init__(**kwargs)
        self.containers = containers
        ...

    def to_dict(self) -> None:
        super().to_dict()
        if not self.kind_dict and not self.yaml_file:
            # Required fields raise MissingRequiredArgumentError
            # Optional fields added only if not None
            ...
    # End of generated code
```

User-added code placed **after** the `# End of generated code` marker is preserved across regenerations.

### `$ref` Resolution

OpenAPI schemas extensively use `$ref` pointers to reference shared type definitions (e.g., `"$ref": "#/components/schemas/io.k8s.apimachinery.pkg.apis.meta.v1.ObjectMeta"`). The `SchemaValidator._resolve_refs()` method handles this by:

1. Parsing the reference path to extract the definition name
2. Looking up the definition in `_definitions_data` using multiple key formats (full dotted path, short name, or `version/Kind` format)
3. Recursively resolving any `$ref` pointers within the resolved definition
4. Falling back to a permissive object schema for well-known core types that may be missing

### Schema Supplementation via `oc explain`

The raw OpenAPI v3 schemas from the cluster sometimes lack field descriptions and required-field markers. The system supplements these gaps by:

- Running `oc explain <resource>.<field>` commands **in parallel** for critical resource types (Pod, Deployment, Service, etc.)
- Parsing the explain output to extract field descriptions, types, and required-field markers
- Merging descriptions into existing schema properties where they are missing
- Detecting missing `$ref` targets and fetching their definitions via `oc explain --recursive`

This supplementation only runs during a **full update** (same or newer cluster version), not when only fetching missing resources.

## How It Affects You

| What you do | What happens under the hood |
|---|---|
| Call `resource.validate()` | `SchemaValidator` loads schemas from the compressed archive, resolves `$ref` references, validates with jsonschema, and raises `ValidationError` with a formatted message |
| Pass `schema_validation_enabled=True` | `create()` and `update_replace()` automatically validate before sending to the API server |
| Run `class-generator --update-schema` | Full schema fetch from cluster, version check, parallel API downloads, `oc explain` supplementation, and compressed archive write |
| Run `class-generator -k Pod` | Reads the mapping for `pod`, parses fields/types, renders Jinja2 template, preserves user code, writes and formats the Python file |
| Run `class-generator --update-schema-for MyResource` | Single-resource fetch that bypasses the cluster version guard |
| Run `class-generator --regenerate-all` | Discovers all generated files (by marker comment), regenerates each in parallel, preserving user code |
| Validation reports "No schema found" | The resource kind is not in the mappings archive — run `--update-schema` to fetch it |
| Multiple files generated for one kind | The kind exists in multiple API groups — each group gets its own file with a suffix |

## Related Pages

- See [Validating Resources Against OpenAPI Schemas](validating-resources.html) for a hands-on guide to enabling and using validation.
- See [Generating New Resource Classes with class-generator](generating-resource-classes.html) for step-by-step class generation instructions.
- See [class-generator CLI Reference](class-generator-cli.html) for all CLI options including `--update-schema`, `--update-schema-for`, `--regenerate-all`, and `--coverage-report`.
- See [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html) to learn how generated classes fit into the `Resource` / `NamespacedResource` inheritance tree.
- See [Resource and NamespacedResource API](resource-api.html) for complete documentation of the `validate()` and `validate_dict()` methods.
- See [Testing Without a Cluster Using the Fake Client](testing-without-cluster.html) for how the fake client uses schema-based status templates.

## Related Pages

- [Validating Resources Against OpenAPI Schemas](validating-resources.html)
- [Generating New Resource Classes with class-generator](generating-resource-classes.html)
- [class-generator CLI Reference](class-generator-cli.html)
- [Understanding the Resource Class Hierarchy](resource-class-hierarchy.html)
- [Resource and NamespacedResource API](resource-api.html)

---

Source: working-with-events.md

Let me start by exploring the repository structure and the knowledge graph.

Now let me explore the Event-related source code:

Now let me look at the `_build_event_field_selector` function and related event test files:

Now let me check the `events` test files in the tests directory:

Now let me check the resource.py import and the `events` method more carefully, and also look at how TIMEOUT_4MINUTES is defined:

Let me check where TIMEOUT_4MINUTES is defined in resource.py:

Now let me check the constants value for TIMEOUT_4MINUTES:

Now let me check how the Event class is used in the fake client:

Now let me look at the `_extract_event_info` and the `_process_watch_events` functions to understand the event data structure better:

Now let me look at the event object structure from watch events:

Now let me check how to properly import and use Event:

Let me also look at the test file for events in more detail:

Now let me check the watch method's return structure for events:

Now I have a thorough understanding. Let me write the documentation.

# Working with Kubernetes Events

Monitor, retrieve, and clean up Kubernetes events to debug resource behavior and automate event-driven workflows using the `Event` class and the `resource.events()` method.

## Prerequisites

- A connected Kubernetes/OpenShift client — see [Connecting to Clusters](connecting-to-clusters.html)
- The `ocp_resources` package installed — see [Installing and Creating Your First Resource](quickstart.html)

## Quick Example

Stream events from a namespace in real-time:

```python
from ocp_resources.event import Event

for event in Event.get(client=client, namespace="my-namespace", timeout=30):
    print(f"{event['type']}: {event.object.reason} — {event.object.message}")
```

## Streaming Events in Real-Time with `Event.get()`

`Event.get()` opens a watch connection and yields events as they occur. It's a generator that blocks until the timeout expires or you break out of the loop.

```python
from ocp_resources.event import Event

for event in Event.get(
    client=client,
    namespace="my-namespace",
    timeout=60,
):
    obj = event.object
    print(f"[{obj.type}] {obj.reason}: {obj.message}")
```

Each yielded event is a watch event dictionary with these keys:

| Key            | Description                                                        |
|----------------|--------------------------------------------------------------------|
| `type`         | Watch event type: `ADDED`, `MODIFIED`, or `DELETED`                |
| `object`       | The Event resource object (access `.type`, `.reason`, `.message`, `.involvedObject`, etc.) |
| `raw_object`   | The raw dictionary representation of the event                     |

### `Event.get()` Parameters

| Parameter          | Type                    | Default  | Description                                               |
|--------------------|-------------------------|----------|-----------------------------------------------------------|
| `client`           | `DynamicClient`         | `None`   | Kubernetes dynamic client (required)                      |
| `namespace`        | `str \| None`           | `None`   | Filter events to a specific namespace                     |
| `name`             | `str \| None`           | `None`   | Filter by event name                                      |
| `label_selector`   | `str \| None`           | `None`   | Filter by labels (e.g. `"app=nginx"`)                     |
| `field_selector`   | `str \| None`           | `None`   | Filter by fields (e.g. `"type==Warning"`)                 |
| `resource_version` | `str \| None`           | `None`   | Start watching from a specific resource version           |
| `timeout`          | `int \| None`           | `None`   | Timeout in seconds; `None` watches indefinitely           |

**Returns:** `Generator` — yields watch event dictionaries.

### Filtering with Field Selectors

Field selectors let you narrow down events to exactly what you care about. Combine multiple selectors with commas:

```python
# Only Warning events for ClusterServiceVersion resources
for event in Event.get(
    client=client,
    namespace="my-namespace",
    field_selector="involvedObject.kind==ClusterServiceVersion,type==Warning,reason==AnEventReason",
    timeout=10,
):
    print(event.object.message)
```

Common field selector fields:

| Field                        | Example                                        |
|------------------------------|-------------------------------------------------|
| `type`                       | `type==Warning` or `type==Normal`               |
| `reason`                     | `reason==FailedScheduling`                      |
| `involvedObject.kind`        | `involvedObject.kind==Pod`                      |
| `involvedObject.name`        | `involvedObject.name==my-pod`                   |
| `involvedObject.namespace`   | `involvedObject.namespace==default`             |

## Listing Existing Events with `Event.list()`

Unlike `Event.get()`, which streams events in real-time, `Event.list()` returns existing events immediately as a list. By default it only returns events from the last 5 minutes.

```python
from ocp_resources.event import Event

# List all Warning events from the last 5 minutes
events = Event.list(
    client=client,
    namespace="my-namespace",
    field_selector="type==Warning",
)

for event in events:
    print(f"{event.reason}: {event.message}")
```

Results are sorted by `lastTimestamp` descending (most recent first).

### `Event.list()` Parameters

| Parameter        | Type                | Default | Description                                               |
|------------------|---------------------|---------|-----------------------------------------------------------|
| `client`         | `DynamicClient`     | —       | Kubernetes dynamic client (required)                      |
| `namespace`      | `str \| None`       | `None`  | Filter events to a specific namespace                     |
| `field_selector` | `str \| None`       | `None`  | Filter by fields (e.g. `"type==Warning"`)                 |
| `label_selector` | `str \| None`       | `None`  | Filter by labels                                          |
| `since_seconds`  | `int`               | `300`   | Only return events from the last N seconds                |

**Returns:** `list` — event resource objects sorted by timestamp (most recent first).

**Raises:** `ValueError` — if `since_seconds` is negative.

```python
# Events from the last 30 minutes
events = Event.list(client=client, since_seconds=1800)

# All events (no time filter — uses a very large window)
events = Event.list(client=client, since_seconds=999999)
```

### When to Use `Event.get()` vs `Event.list()`

| Use case                                         | Method         |
|--------------------------------------------------|----------------|
| Watch for new events as they happen               | `Event.get()`  |
| Fetch events that already occurred                | `Event.list()` |
| Collect events during a test run                  | `Event.get()`  |
| Check recent events after a failure               | `Event.list()` |
| Stream events with a timeout                      | `Event.get()`  |
| Get a snapshot sorted by time                     | `Event.list()` |

## Getting Events for a Specific Resource

Every resource instance has an `.events()` method that automatically filters events to that specific resource by setting `involvedObject.name` in the field selector.

```python
from ocp_resources.pod import Pod

pod = Pod(client=client, name="my-pod", namespace="default")

for event in pod.events(timeout=10):
    print(f"{event.object.reason}: {event.object.message}")
```

You can add extra filters on top of the automatic resource filter:

```python
# Only Warning events for this specific pod
for event in pod.events(
    field_selector="type==Warning",
    timeout=10,
):
    print(event.object.message)
```

### `resource.events()` Parameters

| Parameter          | Type    | Default | Description                                               |
|--------------------|---------|---------|-----------------------------------------------------------|
| `name`             | `str`   | `""`    | Filter by event name                                      |
| `label_selector`   | `str`   | `""`    | Filter by labels                                          |
| `field_selector`   | `str`   | `""`    | Additional field selectors (combined with `involvedObject.name` automatically) |
| `resource_version` | `str`   | `""`    | Start watching from a specific resource version           |
| `timeout`          | `int`   | `240`   | Timeout in seconds (default: 4 minutes)                   |

**Returns:** `Generator` — yields watch event dictionaries, same format as `Event.get()`.

> **Note:** The `field_selector` you provide is appended to the automatic `involvedObject.name==<resource-name>` filter. You don't need to specify the resource name yourself.

## Deleting Events with `Event.delete_events()`

Clean up events before a test run to avoid false positives from stale events:

```python
from ocp_resources.event import Event

# Delete all events in a namespace
Event.delete_events(client=client, namespace="my-namespace")

# Delete events matching a specific reason
Event.delete_events(
    client=client,
    namespace="my-namespace",
    field_selector="reason==AnEventReason",
)
```

### `Event.delete_events()` Parameters

| Parameter          | Type                    | Default  | Description                                               |
|--------------------|-------------------------|----------|-----------------------------------------------------------|
| `client`           | `DynamicClient`         | `None`   | Kubernetes dynamic client (required)                      |
| `namespace`        | `str \| None`           | `None`   | Target namespace                                          |
| `name`             | `str \| None`           | `None`   | Specific event name to delete                             |
| `label_selector`   | `str \| None`           | `None`   | Filter by labels                                          |
| `field_selector`   | `str \| None`           | `None`   | Filter by fields                                          |
| `resource_version` | `str \| None`           | `None`   | Filter by resource version                                |
| `timeout`          | `int \| None`           | `None`   | Timeout in seconds                                        |

**Returns:** `None`

## Advanced Usage

### Test Setup: Clean Events Before Each Test

A common pattern is deleting events before a test so that only events generated during the test are captured:

```python
from ocp_resources.event import Event


def test_pod_scheduling(client, namespace):
    # Clean slate — remove old events
    Event.delete_events(client=client, namespace=namespace)

    # ... create resources and trigger the behavior under test ...

    # Verify expected events occurred
    events = Event.list(
        client=client,
        namespace=namespace,
        field_selector="reason==Scheduled",
        since_seconds=60,
    )
    assert len(events) > 0, "Pod was not scheduled"
```

### Collecting Events During an Operation

Use `Event.get()` with a timeout to capture all events that occur during an operation:

```python
from ocp_resources.event import Event

events = []
for event in Event.get(
    client=client,
    namespace="my-namespace",
    field_selector="involvedObject.kind==Deployment",
    timeout=30,
):
    events.append(event.object)
    if event.object.reason == "ScalingReplicaSet":
        break  # Found what we were looking for

print(f"Captured {len(events)} events")
```

### Watching Cluster-Wide Events

Omit the `namespace` parameter to watch events across all namespaces:

```python
from ocp_resources.event import Event

# Watch all Warning events cluster-wide
for event in Event.get(
    client=client,
    field_selector="type==Warning",
    timeout=60,
):
    ns = event.object.involvedObject.get("namespace", "cluster-scoped")
    print(f"[{ns}] {event.object.reason}: {event.object.message}")
```

### Accessing Event Object Properties

The `event.object` yielded by `Event.get()` and `resource.events()` is a Kubernetes `ResourceInstance` with these commonly used attributes:

| Attribute                     | Description                                  |
|-------------------------------|----------------------------------------------|
| `event.object.type`           | `"Normal"` or `"Warning"`                    |
| `event.object.reason`         | Short reason string (e.g. `"Scheduled"`)     |
| `event.object.message`        | Human-readable event message                 |
| `event.object.count`          | Number of times this event occurred          |
| `event.object.firstTimestamp`  | When the event first occurred                |
| `event.object.lastTimestamp`   | When the event most recently occurred        |
| `event.object.source`         | Dict with `component` and `host`             |
| `event.object.involvedObject` | Dict with `kind`, `name`, `namespace`, etc.  |

## Troubleshooting

**`Event.get()` hangs indefinitely**
You likely omitted the `timeout` parameter. Always set a `timeout` value to avoid blocking forever:

```python
# Bad — hangs if no events arrive
for event in Event.get(client=client, namespace="ns"):
    ...

# Good — stops after 30 seconds
for event in Event.get(client=client, namespace="ns", timeout=30):
    ...
```

**`Event.list()` returns an empty list**
- Check the `since_seconds` value. The default is 300 (5 minutes). If the events are older, increase the value.
- Verify the `namespace` is correct.
- Ensure the `field_selector` syntax uses `==` (double equals), not `=`.

**`resource.events()` returns events for other resources too**
This method filters only by `involvedObject.name`. If multiple resources share the same name across different kinds, add a `field_selector` to narrow it down:

```python
for event in pod.events(
    field_selector="involvedObject.kind==Pod",
    timeout=10,
):
    print(event.object.message)
```

> **Warning:** The `dyn_client` parameter on `Event.get()` and `Event.delete_events()` is deprecated and will be removed in the next major release. Use `client` instead.

## Related Pages

- [Querying and Watching Resources](querying-resources.html)
- [Common Resource Patterns](common-patterns.html)
- [Executing Commands in Pods and Retrieving Logs](pod-execution-and-logs.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)

---

Source: pod-execution-and-logs.md

Let me explore the repository structure and understand the codebase first.

Now let me check the `read_namespaced_pod_log` kwargs available, and look at how the MCP server uses log parameters:

Now let me check the import paths and verify the `get_client` function:

Now let me check test files for pod execution tests:

Now let me also verify whether the Node class has a `name` property:

Now I have all the information I need to write the documentation. Let me compose it:

# Executing Commands in Pods and Retrieving Logs

Run commands inside running pods and retrieve their logs to debug and interact with workloads on your OpenShift or Kubernetes cluster.

## Prerequisites

- A connected cluster client (see [Connecting to Clusters](connecting-to-clusters.html))
- An existing, running pod you want to interact with

## Quick Example

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()
pod = Pod(client=client, name="my-app", namespace="default")

# Run a command inside the pod
output = pod.execute(command=["echo", "hello"])
print(output)  # "hello\n"

# Get pod logs
logs = pod.log()
print(logs)
```

## Executing Commands with `Pod.execute()`

`Pod.execute()` runs a command inside a pod container using the Kubernetes exec API and returns the standard output as a string.

### Method Signature

```python
Pod.execute(
    command: list[str],
    timeout: int = 60,
    container: str = "",
    ignore_rc: bool = False,
) -> str
```

| Parameter   | Type         | Default | Description                                                                                 |
|-------------|--------------|---------|---------------------------------------------------------------------------------------------|
| `command`   | `list[str]`  | —       | The command to run, as a list of strings (e.g., `["ls", "-la", "/tmp"]`)                    |
| `timeout`   | `int`        | `60`    | Maximum seconds to wait for the command to complete                                         |
| `container` | `str`        | `""`    | Container name to execute in. If empty, uses the first container in the pod spec            |
| `ignore_rc` | `bool`       | `False` | When `True`, return stdout even if the command exits with a non-zero return code            |

**Returns:** `str` — the standard output of the command.

**Raises:** `ExecOnPodError` — when the command fails (non-zero exit code) and `ignore_rc` is `False`.

### Step-by-Step: Running a Simple Command

1. Get a reference to the pod:

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()
pod = Pod(client=client, name="my-app", namespace="my-namespace")
```

2. Execute a command:

```python
output = pod.execute(command=["cat", "/etc/hostname"])
print(output)
```

3. Use the result for further logic:

```python
files = pod.execute(command=["ls", "/app/data"])
for filename in files.strip().split("\n"):
    print(f"Found: {filename}")
```

### Selecting a Container

For multi-container pods, specify which container to run the command in. If you omit `container`, the first container defined in the pod spec is used.

```python
# Execute in a specific container
output = pod.execute(
    command=["cat", "/var/log/app.log"],
    container="sidecar",
)
```

### Setting a Timeout

Long-running commands may need a longer timeout. The default is 60 seconds.

```python
# Allow up to 5 minutes for a heavy operation
output = pod.execute(
    command=["pg_dump", "mydb"],
    timeout=300,
)
```

When the timeout is exceeded, an `ExecOnPodError` is raised with the error message `"stream resp is closed"`.

### Ignoring Non-Zero Exit Codes

Some commands return a non-zero exit code as part of normal operation (e.g., `grep` returns 1 when no match is found). Use `ignore_rc=True` to get the output regardless:

```python
# grep returns exit code 1 when there's no match — don't treat that as an error
output = pod.execute(
    command=["grep", "ERROR", "/var/log/app.log"],
    ignore_rc=True,
)
if output:
    print("Errors found:", output)
else:
    print("No errors in logs")
```

### Handling Errors with `ExecOnPodError`

When a command fails, `ExecOnPodError` provides structured access to the failure details.

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ExecOnPodError

try:
    pod.execute(command=["ls", "/nonexistent"])
except ExecOnPodError as e:
    print(f"Command: {e.cmd}")      # ['ls', '/nonexistent']
    print(f"Return code: {e.rc}")   # Non-zero exit code (or -1 for stream errors)
    print(f"Stdout: {e.out}")       # Standard output captured before failure
    print(f"Stderr: {e.err}")       # Standard error or error channel details
```

`ExecOnPodError` attributes:

| Attribute | Type        | Description                                                          |
|-----------|-------------|----------------------------------------------------------------------|
| `cmd`     | `list[str]` | The command that was executed                                        |
| `rc`      | `int`       | Return code (`-1` for stream/timeout errors, otherwise the exit code)|
| `out`     | `str`       | Standard output captured from the command                            |
| `err`     | `str` or `dict` | Standard error output, or the Kubernetes error channel response  |

> **Tip:** For a complete list of all custom exceptions, see [Error Handling and Exception Reference](error-handling.html).

## Retrieving Logs with `Pod.log()`

`Pod.log()` returns the logs from a pod container as a string. It passes keyword arguments directly to the Kubernetes `read_namespaced_pod_log` API.

### Basic Usage

```python
logs = pod.log()
print(logs)
```

### Common Keyword Arguments

Pass any parameter supported by the Kubernetes `read_namespaced_pod_log` API:

| Parameter       | Type   | Description                                               |
|-----------------|--------|-----------------------------------------------------------|
| `container`     | `str`  | Container name to get logs from (required for multi-container pods) |
| `previous`      | `bool` | Return logs from a previous terminated container instance  |
| `tail_lines`    | `int`  | Number of lines from the end of the logs to return         |
| `since_seconds` | `int`  | Only return logs newer than this many seconds              |

### Examples

```python
# Get logs from a specific container
logs = pod.log(container="nginx")

# Get the last 50 lines
logs = pod.log(tail_lines=50)

# Get logs from the last 5 minutes
logs = pod.log(since_seconds=300)

# Get logs from a crashed container's previous instance
logs = pod.log(container="app", previous=True)
```

## Accessing Pod Properties

Beyond executing commands and reading logs, the `Pod` class provides properties for inspecting the pod's runtime state.

### Getting the Node

The `node` property returns a `Node` object for the node where the pod is scheduled:

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

for pod in Pod.get(client=client, label_selector="app=my-app"):
    node = pod.node
    print(f"Pod {pod.name} is running on node {node.name}")
```

> **Note:** The `node` property raises an `AssertionError` if the pod has not yet been scheduled to a node.

### Getting the Pod IP Address

The `ip` property returns the pod's IP address from its status:

```python
pod_ip = pod.ip
print(f"Pod IP: {pod_ip}")
```

### Getting the Pod Status

The `status` property (inherited from the base resource class) returns the pod's phase:

```python
print(f"Pod status: {pod.status}")  # e.g., "Running", "Pending", "Succeeded"
```

## Advanced Usage

### Iterating Over Pods with Selectors

Combine `Pod.get()` with `execute()` or `log()` to debug across multiple pods:

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ExecOnPodError
from ocp_resources.resource import get_client

client = get_client()

for pod in Pod.get(client=client, namespace="production", label_selector="app=web"):
    try:
        uptime = pod.execute(command=["uptime"])
        print(f"{pod.name} on {pod.node.name}: {uptime.strip()}")
    except ExecOnPodError as e:
        print(f"{pod.name}: command failed — {e}")
```

### Executing in Multi-Container Pods

When pods contain sidecar containers (e.g., logging or proxy containers), always specify the target container explicitly:

```python
# Main application container
app_output = pod.execute(command=["cat", "/app/config.yaml"], container="app")

# Envoy sidecar
proxy_stats = pod.execute(command=["curl", "localhost:15000/stats"], container="istio-proxy")

# Logs from each container
app_logs = pod.log(container="app", tail_lines=100)
proxy_logs = pod.log(container="istio-proxy", tail_lines=100)
```

### Collecting Debug Information

A practical recipe for gathering debugging data from a running pod:

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ExecOnPodError
from ocp_resources.resource import get_client

client = get_client()
pod = Pod(client=client, name="my-app", namespace="default")

debug_info = {
    "pod_name": pod.name,
    "node": pod.node.name,
    "ip": pod.ip,
    "status": pod.status,
    "logs_tail": pod.log(tail_lines=20),
}

# Safely attempt commands that might fail
for cmd_name, cmd in [("env", ["env"]), ("df", ["df", "-h"]), ("ps", ["ps", "aux"])]:
    try:
        debug_info[cmd_name] = pod.execute(command=cmd, timeout=10)
    except ExecOnPodError:
        debug_info[cmd_name] = "command failed"

for key, value in debug_info.items():
    print(f"--- {key} ---\n{value}\n")
```

### Using Pod Context Manager for Temporary Pods

Create a pod, run commands, and clean up automatically:

```python
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

with Pod(
    client=client,
    name="debug-pod",
    namespace="default",
    containers=[{
        "name": "debug",
        "image": "registry.access.redhat.com/ubi9/ubi:latest",
        "command": ["sleep", "3600"],
    }],
) as pod:
    pod.wait_for_status(status=Pod.Status.RUNNING, timeout=120)
    output = pod.execute(command=["cat", "/etc/os-release"])
    print(output)
# Pod is automatically deleted when the context manager exits
```

> **Note:** See [Creating and Managing Resources](creating-and-managing-resources.html) for more about context managers and resource lifecycle.

## Troubleshooting

### `ExecOnPodError` with `"stream resp is closed"`

This typically means the command timed out. Increase the `timeout` parameter:

```python
pod.execute(command=["long-running-script.sh"], timeout=300)
```

### `AssertionError: Node not found for pod <name>`

The pod is not yet scheduled. Wait for it to be running before accessing `pod.node`:

```python
pod.wait_for_status(status=Pod.Status.RUNNING, timeout=120)
node = pod.node
```

### `ExecOnPodError` with return code `-1` and an error channel dict

This indicates a Kubernetes-level failure (e.g., container not found, pod not running). Check that:

- The pod is in `Running` status
- The `container` name matches one defined in the pod spec
- The container has started and is not in `CrashLoopBackOff`

### Logs return empty string

The container may not have produced any output yet, or it may have restarted. Try:

```python
# Check previous container instance
logs = pod.log(previous=True)

# Or narrow to recent logs
logs = pod.log(since_seconds=60)
```

## Related Pages

- [Common Resource Patterns](common-patterns.html)
- [Working with Kubernetes Events](working-with-events.html)
- [Error Handling and Exception Reference](error-handling.html)
- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Working with Virtual Machines (KubeVirt)](working-with-virtual-machines.html)

---

Source: error-handling.md

Let me explore the repository structure and find all relevant exception definitions and usages.

Now let me look at the `ConditionError` usage more closely and the `MissingTemplateVariables` and `MissingResourceError` exceptions:

Now let me look at how the deploy method uses ValidationError:

Now let me see where `validate` is called during `create`:

Now let me look at the `Pod.execute` method more closely:

Now let me look at how `client_with_basic_auth` works:

Now let me see the NNCP methods in full context:

Let me also look at `MissingTemplateVariables` usage more closely:

Now let me check the `__init__` of Resource to see how `schema_validation_enabled` is handled:

Now let me also check the `wait_for_condition` method signature more closely:

Let me also look at `MissingResourceResError` usage:

Now let me check the test for `ResourceTeardownError`:

Let me see how SecretTestExit is defined:

Let me check the import path for all exceptions:

Now I have a thorough understanding of the codebase. Let me write the reference documentation.

# Error Handling and Exception Reference

All custom exceptions are defined in `ocp_resources.exceptions` and can be imported directly:

```python
from ocp_resources.exceptions import (
    ExecOnPodError,
    MissingRequiredArgumentError,
    ResourceTeardownError,
    ValidationError,
    ConditionError,
    NNCPConfigurationFailed,
    ClientWithBasicAuthError,
    MissingResourceError,
    MissingTemplateVariables,
)
```

> **Tip:** All exceptions inherit from Python's built-in `Exception` class and can be caught with a bare `except Exception` if needed.

---

## ExecOnPodError

Raised when a command executed inside a pod via `Pod.execute()` fails. See [Executing Commands in Pods and Retrieving Logs](pod-execution-and-logs.html) for full usage details.

**Import:**

```python
from ocp_resources.exceptions import ExecOnPodError
```

**Constructor Parameters:**

| Parameter | Type | Description |
|-----------|------|-------------|
| `command` | `list[str]` | The command that was executed |
| `rc` | `int` | Return code (`-1` if the return code could not be determined) |
| `out` | `str` | Standard output captured from the command |
| `err` | `Any` | Standard error output or error channel details |

**Attributes on the caught exception:**

| Attribute | Type | Description |
|-----------|------|-------------|
| `cmd` | `list[str]` | The command that was executed |
| `rc` | `int` | The return code |
| `out` | `str` | Standard output |
| `err` | `Any` | Standard error or error channel dict |

**Raised by:** `Pod.execute()` in the following scenarios:

- Command times out (stream response closes before completion) — `rc=-1`
- Error channel returns no status — `rc=-1`
- Error channel status is `"Failure"` — `rc=-1`, `err` contains the full error channel dict
- Command exits with a non-zero exit code — `rc` contains the actual exit code

**String representation:**

```
Command execution failure: ['ls', '/nonexistent'], RC: 2, OUT: , ERR: ls: cannot access '/nonexistent'
```

**Example:**

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ExecOnPodError

pod = Pod(client=client, name="my-pod", namespace="default")

try:
    output = pod.execute(command=["ls", "/nonexistent"], timeout=30)
except ExecOnPodError as e:
    print(f"Command: {e.cmd}")
    print(f"Return code: {e.rc}")
    print(f"Stdout: {e.out}")
    print(f"Stderr: {e.err}")
```

**Handling non-zero exit codes without exceptions:**

```python
# Use ignore_rc=True to suppress ExecOnPodError on non-zero exit codes
output = pod.execute(command=["grep", "pattern", "/var/log/messages"], ignore_rc=True)
```

---

## MissingRequiredArgumentError

Raised when a resource is instantiated without providing required arguments and no `yaml_file` or `kind_dict` was supplied. This error is raised during `to_dict()`, which is called automatically by `create()` and `deploy()`.

**Import:**

```python
from ocp_resources.exceptions import MissingRequiredArgumentError
```

**Constructor Parameters:**

| Parameter | Type | Description |
|-----------|------|-------------|
| `argument` | `str` | Name of the missing required argument(s) |

**Attributes on the caught exception:**

| Attribute | Type | Description |
|-----------|------|-------------|
| `argument` | `str` | The missing argument name |

**String representation:**

```
Missing required argument/s. Either provide yaml_file, kind_dict or pass self.containers
```

**Raised by:** The `to_dict()` method of resource subclasses when required spec fields are not provided. Examples of resources that raise this exception:

| Resource Class | Required Arguments |
|---|---|
| `Pod` | `containers` |
| `StorageClass` | `provisioner` |
| `ClusterRoleBinding` | `cluster_role` |
| `ClusterResourceQuota` | `quota`, `selector` |
| `CronJob` | `job_template`, `schedule` |
| `InferenceService` | `predictor` |
| `IPAddressPool` | `addresses` |
| `ResourceQuota` | `hard` |
| `UserDefinedNetwork` | `topology` |

> **Note:** This exception is **not** raised if you construct the resource using `yaml_file` or `kind_dict`, since those bypass the `to_dict()` logic.

**Example:**

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import MissingRequiredArgumentError

try:
    pod = Pod(client=client, name="my-pod", namespace="default")
    pod.deploy()
except MissingRequiredArgumentError as e:
    print(f"Missing: {e.argument}")
    # Output: Missing required argument/s. Either provide yaml_file, kind_dict or pass self.containers
```

---

## ResourceTeardownError

Raised when a resource's `clean_up()` method returns `False` during context manager exit. This indicates the resource could not be deleted.

**Import:**

```python
from ocp_resources.exceptions import ResourceTeardownError
```

**Constructor Parameters:**

| Parameter | Type | Description |
|-----------|------|-------------|
| `resource` | `Any` | The resource object that failed to tear down |

**Attributes on the caught exception:**

| Attribute | Type | Description |
|-----------|------|-------------|
| `resource` | `Any` | The resource object that failed teardown |

**String representation:**

```
Failed to execute teardown for resource <resource>
```

**Raised by:** `Resource.__exit__()` — the context manager exit handler. Specifically, this is raised when:

1. The resource was created with `teardown=True` (the default).
2. The `clean_up()` method returns `False`.

**Example:**

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ResourceTeardownError

try:
    with Pod(
        client=client,
        name="my-pod",
        namespace="default",
        containers=[{"name": "nginx", "image": "nginx:latest"}],
    ) as pod:
        # Use pod...
        pass
    # clean_up() is called automatically here
except ResourceTeardownError as e:
    print(f"Could not delete: {e.resource}")
```

> **Tip:** Set `teardown=False` on the resource constructor if you do not want automatic cleanup on context manager exit, and thus never want this exception raised.

---

## ValidationError

Raised when a resource fails schema validation against the OpenAPI specification. See [Validating Resources Against OpenAPI Schemas](validating-resources.html) for full validation guide.

**Import:**

```python
from ocp_resources.exceptions import ValidationError
```

**Constructor Parameters:**

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `message` | `str` | *(required)* | Human-readable error description |
| `path` | `str` | `""` | JSONPath to the invalid field (e.g., `"spec.containers[0].image"`) |
| `schema_error` | `Any` | `None` | Original `jsonschema` validation error for debugging |

**Attributes on the caught exception:**

| Attribute | Type | Description |
|-----------|------|-------------|
| `message` | `str` | Human-readable error message |
| `path` | `str` | JSONPath to the invalid field |
| `schema_error` | `Any` | Original `jsonschema.ValidationError` if available |

**String representation:**

```
Validation error at 'spec.containers[0].image': Invalid type
```

When `path` is empty:

```
Validation error: Field is required
```

**Raised by:**

| Method | Trigger |
|--------|---------|
| `resource.validate()` | Called explicitly by user |
| `resource.create()` | When `schema_validation_enabled=True` |
| `resource.update_replace()` | When `schema_validation_enabled=True` |
| `Resource.validate_dict()` | Class method for validating raw dicts |

> **Note:** `resource.update()` does **not** trigger validation even when `schema_validation_enabled=True`, because updates send partial patches that would fail full schema validation.

**Example — explicit validation:**

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ValidationError

pod = Pod(client=client, name="my-pod", namespace="default",
          containers=[{"name": "nginx", "image": "nginx:latest"}])

try:
    pod.validate()
except ValidationError as e:
    print(f"Error: {e.message}")
    print(f"Path: {e.path}")
    if e.schema_error:
        print(f"Original error: {e.schema_error}")
```

**Example — auto-validation on create:**

```python
from ocp_resources.pod import Pod
from ocp_resources.exceptions import ValidationError

try:
    pod = Pod(
        client=client,
        name="my-pod",
        namespace="default",
        containers=[{"name": "nginx", "image": "nginx:latest"}],
        schema_validation_enabled=True,
    )
    pod.deploy()
except ValidationError as e:
    print(f"Invalid resource: {e}")
```

**Example — validate a raw dictionary:**

```python
from ocp_resources.deployment import Deployment
from ocp_resources.exceptions import ValidationError

deployment_dict = {
    "apiVersion": "apps/v1",
    "kind": "Deployment",
    "metadata": {"name": "my-deploy"},
    "spec": {"replicas": "three"},  # Wrong type — should be int
}

try:
    Deployment.validate_dict(resource_dict=deployment_dict)
except ValidationError as e:
    print(f"Validation failed: {e}")
```

---

## ConditionError

Raised when a resource reaches an undesired stop condition during `wait_for_condition()`. See [Waiting for Resource Conditions and Status](waiting-for-conditions.html) for details.

**Import:**

```python
from ocp_resources.exceptions import ConditionError
```

**Constructor Parameters:** Standard `Exception` — accepts a single string message.

**Raised by:** `Resource.wait_for_condition()` when the `stop_condition` parameter matches before the desired condition is met.

**`wait_for_condition()` parameters relevant to `ConditionError`:**

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `condition` | `str` | *(required)* | Condition type to wait for |
| `status` | `str` | *(required)* | Expected status value |
| `timeout` | `int` | `300` | Maximum wait time in seconds |
| `sleep_time` | `int` | `1` | Polling interval in seconds |
| `stop_condition` | `str \| None` | `None` | Condition type that should abort the wait |
| `stop_status` | `str` | `"True"` | Status value for the stop condition |

**String representation:**

```
Deployment my-deploy reached stop_condition 'Failed' in status 'True':
{'type': 'Failed', 'status': 'True', 'reason': 'DeadlineExceeded', 'message': '...'}
```

> **Note:** When `stop_condition` is `None` (the default), `ConditionError` is never raised. Instead, `TimeoutExpiredError` is raised if the desired condition is not met within the timeout.

**Example:**

```python
from ocp_resources.deployment import Deployment
from ocp_resources.exceptions import ConditionError
from timeout_sampler import TimeoutExpiredError

deploy = Deployment(client=client, name="my-deploy", namespace="default")

try:
    deploy.wait_for_condition(
        condition="Available",
        status="True",
        timeout=120,
        stop_condition="Failed",
        stop_status="True",
    )
except ConditionError as e:
    print(f"Resource entered failure state: {e}")
except TimeoutExpiredError:
    print("Timed out waiting for condition")
```

---

## NNCPConfigurationFailed

Raised when a `NodeNetworkConfigurationPolicy` (NNCP) fails to configure on one or more nodes.

**Import:**

```python
from ocp_resources.exceptions import NNCPConfigurationFailed
```

**Constructor Parameters:** Standard `Exception` — accepts a single string message describing the failure reason and error details.

**Raised by:** `NodeNetworkConfigurationPolicy.wait_for_status_success()` in two scenarios:

| Scenario | Message Pattern |
|----------|-----------------|
| No matching node found | `"{name}. Reason: NoMatchingNode"` |
| Configuration failed on nodes | `"Reason: FailedToConfigure\n{error_details}"` |

> **Note:** When `wait_for_status_success()` catches `TimeoutExpiredError` or `NNCPConfigurationFailed`, it logs the error with node details and re-raises the exception.

**Example:**

```python
from ocp_resources.node_network_configuration_policy import NodeNetworkConfigurationPolicy
from ocp_resources.exceptions import NNCPConfigurationFailed
from timeout_sampler import TimeoutExpiredError

nncp = NodeNetworkConfigurationPolicy(
    client=client,
    name="my-nncp",
    desired_state={"interfaces": [{"name": "eth1", "type": "ethernet", "state": "up"}]},
)

try:
    nncp.deploy()
    nncp.wait_for_status_success()
except NNCPConfigurationFailed as e:
    print(f"NNCP configuration failed: {e}")
except TimeoutExpiredError:
    print("NNCP configuration timed out")
```

---

## ClientWithBasicAuthError

Raised during OAuth-based client authentication when using username/password credentials to connect to an OpenShift cluster. See [Connecting to Clusters](connecting-to-clusters.html) for connection methods.

**Import:**

```python
from ocp_resources.exceptions import ClientWithBasicAuthError
```

**Constructor Parameters:** Standard `Exception` — accepts a single string message.

**Raised by:** `client_configuration_with_basic_auth()` in the following scenarios:

| Scenario | Message |
|----------|---------|
| OAuth well-known endpoint not reachable | `"No well-known file found at endpoint"` |
| Authorization code not returned after login | `"No authorization code found"` |
| No `authorization_endpoint` in OAuth config | `"No authorization_endpoint found in well-known file"` |
| Token exchange fails | `"Failed to authenticate with basic auth"` |

**Example:**

```python
from ocp_resources.exceptions import ClientWithBasicAuthError
from ocp_resources.resource import client_configuration_with_basic_auth

import kubernetes

configuration = kubernetes.client.Configuration()
configuration.verify_ssl = False

try:
    api_client = client_configuration_with_basic_auth(
        username="admin",
        password="password",
        host="https://api.cluster.example.com:6443",
        configuration=configuration,
    )
except ClientWithBasicAuthError as e:
    print(f"Authentication failed: {e}")
```

---

## MissingResourceError

Raised when a resource object fails to generate its internal representation.

**Import:**

```python
from ocp_resources.exceptions import MissingResourceError
```

**Constructor Parameters:**

| Parameter | Type | Description |
|-----------|------|-------------|
| `name` | `str` | Name of the resource that could not be generated |

**Attributes on the caught exception:**

| Attribute | Type | Description |
|-----------|------|-------------|
| `resource_name` | `str` | The resource name |

**String representation:**

```
Failed to generate resource: my-resource
```

**Example:**

```python
from ocp_resources.exceptions import MissingResourceError

try:
    # Operations that may fail to generate a resource
    ...
except MissingResourceError as e:
    print(f"Resource generation failed: {e.resource_name}")
```

---

## MissingTemplateVariables

Raised when rendering a YAML template and not all required template variables have been provided.

**Import:**

```python
from ocp_resources.exceptions import MissingTemplateVariables
```

**Constructor Parameters:**

| Parameter | Type | Description |
|-----------|------|-------------|
| `var` | `str` | Name of the missing template variable |
| `template` | `str` | Path to the template file |

**Attributes on the caught exception:**

| Attribute | Type | Description |
|-----------|------|-------------|
| `var` | `str` | The missing variable name |
| `template` | `str` | The template file path |

**String representation:**

```
Missing variables image for template tests/manifests/vm.yaml
```

**Example:**

```python
from ocp_resources.exceptions import MissingTemplateVariables

try:
    result = generate_yaml_from_template(name="my-vm")
    # If template also requires 'image', raises MissingTemplateVariables
except MissingTemplateVariables as e:
    print(f"Variable '{e.var}' missing in template '{e.template}'")
```

---

## Common Error-Handling Patterns

### Catching Multiple Library Exceptions

```python
from ocp_resources.exceptions import (
    ExecOnPodError,
    MissingRequiredArgumentError,
    ResourceTeardownError,
    ValidationError,
)

try:
    with Pod(client=client, name="worker", namespace="default",
             containers=[{"name": "app", "image": "myapp:latest"}],
             schema_validation_enabled=True) as pod:
        pod.execute(command=["python", "run_task.py"])
except ValidationError as e:
    print(f"Invalid resource spec: {e}")
except ExecOnPodError as e:
    print(f"Task failed (rc={e.rc}): {e.err}")
except ResourceTeardownError as e:
    print(f"Pod cleanup failed: {e}")
except MissingRequiredArgumentError as e:
    print(f"Missing field: {e.argument}")
```

### Safe Condition Waiting with Early Abort

```python
from ocp_resources.exceptions import ConditionError
from timeout_sampler import TimeoutExpiredError

try:
    resource.wait_for_condition(
        condition="Ready",
        status="True",
        timeout=180,
        stop_condition="Failed",
        stop_status="True",
    )
except ConditionError:
    # Resource entered a terminal failure state — no point waiting further
    resource.clean_up()
    raise
except TimeoutExpiredError:
    # Condition not met within timeout — may be transient
    print("Resource is taking too long, investigating...")
```

### Ignoring Non-Critical Command Failures

```python
from ocp_resources.exceptions import ExecOnPodError

# Method 1: Use ignore_rc=True
output = pod.execute(command=["grep", "ERROR", "/var/log/app.log"], ignore_rc=True)

# Method 2: Catch and inspect the return code
try:
    output = pod.execute(command=["test", "-f", "/tmp/lockfile"])
except ExecOnPodError as e:
    if e.rc == 1:
        print("Lock file does not exist — proceeding")
    else:
        raise  # Re-raise unexpected errors
```

### Validation Before Batch Deployment

```python
from ocp_resources.exceptions import ValidationError

resources = [pod1, pod2, deployment1, service1]

# Pre-validate all resources before deploying any
errors = []
for resource in resources:
    try:
        resource.validate()
    except ValidationError as e:
        errors.append((resource.name, str(e)))

if errors:
    for name, err in errors:
        print(f"  {name}: {err}")
    raise SystemExit("Fix validation errors before deploying")

for resource in resources:
    resource.deploy()
```

---

## Exception Hierarchy Summary

| Exception | Parent | Module |
|-----------|--------|--------|
| `ExecOnPodError` | `Exception` | `ocp_resources.exceptions` |
| `MissingRequiredArgumentError` | `Exception` | `ocp_resources.exceptions` |
| `ResourceTeardownError` | `Exception` | `ocp_resources.exceptions` |
| `ValidationError` | `Exception` | `ocp_resources.exceptions` |
| `ConditionError` | `Exception` | `ocp_resources.exceptions` |
| `NNCPConfigurationFailed` | `Exception` | `ocp_resources.exceptions` |
| `ClientWithBasicAuthError` | `Exception` | `ocp_resources.exceptions` |
| `MissingResourceError` | `Exception` | `ocp_resources.exceptions` |
| `MissingTemplateVariables` | `Exception` | `ocp_resources.exceptions` |

> **Warning:** `MissingResourceResError` is deprecated and will be removed in a future release. Use `MissingResourceError` instead.

## Related Pages

- [Creating and Managing Resources](creating-and-managing-resources.html)
- [Waiting for Resource Conditions and Status](waiting-for-conditions.html)
- [Validating Resources Against OpenAPI Schemas](validating-resources.html)
- [Executing Commands in Pods and Retrieving Logs](pod-execution-and-logs.html)
- [Connecting to Clusters](connecting-to-clusters.html)

---