new-rule

name	new-rule
description	Use when creating a new in-cluster validation rule from scratch - provides rule templates, domain registration, and test scaffolding

Create a new in-cluster rule based on the following description:

$ARGUMENTS

Follow the framework guidelines and development rules defined in: @.claude/rules/in-cluster-check.md

---

Step 1: Determine the Rule Type

Based on the description, choose ONE type (see "Rule Types" in the linked guidelines above):

• Rule — most common, runs on specific nodes
• OrchestratorRule — coordinates data collection across ALL nodes, requires a DataCollector
• DataCollector — collects data from nodes, used BY OrchestratorRule

Step 2: Create the Rule Class

Place the rule in the appropriate file under src/in_cluster_checks/rules/<domain>/. For example: src/in_cluster_checks/rules/hw/hw_validations.py

Use an existing domain folder or create a new one with an __init__.py.

Standard Rule Template

from in_cluster_checks.core.rule import Rule
from in_cluster_checks.core.rule_result import RuleResult, PrerequisiteResult
from in_cluster_checks.utils.enums import Objectives
from in_cluster_checks.utils.safe_cmd_string import SafeCmdString

class MyRuleName(Rule):
    """Rule description - what this rule verifies."""

    objective_hosts = [Objectives.ALL_NODES]  # or MASTERS, WORKERS, etc.
    unique_name = "my_rule_name"  # Must be unique, lowercase with underscores
    title = "Human-readable rule title"
    links = [
        "https://link-to-documentation-or-kb-article",
    ]

    def is_prerequisite_fulfilled(self):
        """Optional: Check if rule can run on this node."""
        return_code, _, _ = self.run_cmd("which some_tool")
        if return_code != 0:
            return PrerequisiteResult.not_met("some_tool is not available on this system")
        return PrerequisiteResult.met()

    def run_rule(self):
        """Execute the rule logic."""
        return_code, stdout, stderr = self.run_cmd("your-command")

        # Parse output and determine pass/fail
        if return_code == 0:
            return RuleResult.passed("Rule passed")
        else:
            return RuleResult.failed(f"Rule failed: {stderr}")

Best Practices:

• Use descriptive unique_name values (e.g., is_disk_space_sufficient, is_network_reachable)
• Include helpful error messages in failed results
• Use RuleResult.warning() for non-critical issues
• Parse command output carefully and handle edge cases
• Use parsing_utils helpers (parse_json, parse_int, get_dict_from_string) when parsing command output

OrchestratorRule Template (multi-node comparison)

from in_cluster_checks.core.operations import DataCollector
from in_cluster_checks.core.rule import OrchestratorRule, RuleResult
from in_cluster_checks.utils.enums import Objectives

class MyDataCollector(DataCollector):
    """Collect data from each node."""
    objective_hosts = [Objectives.ALL_NODES]
    unique_name = "collect_my_data"
    title = "Collect my data"

    def collect_data(self, **kwargs):
        output = self.get_output_from_run_cmd("some_command")
        return parsed_data

class MyOrchestratorRule(OrchestratorRule):
    """Compare data across all nodes."""
    objective_hosts = [Objectives.ORCHESTRATOR]
    unique_name = "my_orchestrator_rule"
    title = "Compare data across nodes"

    def run_rule(self):
        all_data = self.run_data_collector(MyDataCollector)
        if mismatch_found:
            return RuleResult.failed("Data mismatch across nodes")
        return RuleResult.passed()

Using Cluster API (oc_api):

OrchestratorRule provides self.oc_api for cluster resource access:

# Use existing oc_api methods when available
pods = self.oc_api.get_pods(namespace="openshift-etcd")
network = self.oc_api.select_resources("network.operator/cluster", single=True)

# Run commands inside pods
cmd = SafeCmdString("etcdctl version")
rc, out, err = self.oc_api.run_rsh_cmd("openshift-etcd", "etcd-pod", cmd)

# Use run_oc_command for other oc commands
rc, out, err = self.oc_api.run_oc_command("get", ["nodes", "-o", "json"])

Important: Don't add new methods to oc_api if run_oc_command() can achieve the same result. Keep the API minimal.

Step 3: Register in Domain

Add the new rule class to the appropriate domain in src/in_cluster_checks/domains/:

from in_cluster_checks.rules.<domain>.<file> import MyRuleName

class SomeDomain(RuleDomain):
    def get_rule_classes(self) -> List[type]:
        return [
            # ... existing rules ...
            MyRuleName,
        ]

If a new domain is needed, create it following hw_domain.py as a template.

Step 4: Write Tests

Create tests in tests/rules/<domain>/test_<file>.py:

import pytest

from in_cluster_checks.rules.<domain>.<file> import MyRuleName
from tests.pytest_tools.test_operator_base import CmdOutput
from tests.pytest_tools.test_rule_base import RuleTestBase, RuleScenarioParams


class TestMyRuleName(RuleTestBase):
    """Test MyRuleName rule."""

    tested_type = MyRuleName

    good_output = "expected good output here"
    bad_output = "expected bad output here"

    scenario_passed = [
        RuleScenarioParams(
            "description of passing scenario",
            {"exact_command_string": CmdOutput(good_output)},
        ),
    ]

    scenario_failed = [
        RuleScenarioParams(
            "description of failing scenario",
            {"exact_command_string": CmdOutput(bad_output)},
            failed_msg="exact expected failure message",
        ),
    ]

    @pytest.mark.parametrize("scenario_params", scenario_passed)
    def test_scenario_passed(self, scenario_params, tested_object):
        RuleTestBase.test_scenario_passed(self, scenario_params, tested_object)

    @pytest.mark.parametrize("scenario_params", scenario_failed)
    def test_scenario_failed(self, scenario_params, tested_object):
        RuleTestBase.test_scenario_failed(self, scenario_params, tested_object)

Also update the domain test file to include the new rule in assertions.

CmdOutput usage

CmdOutput("stdout text")                          # Success (rc=0)
CmdOutput("stdout text", return_code=1)            # Failed command
CmdOutput("stdout text", return_code=0, err="")    # Full form

For OrchestratorRule tests, use data_collector_dict

RuleScenarioParams(
    "scenario name",
    cmd_input_output_dict={},
    data_collector_dict={
        MyDataCollector: {"node1": data1, "node2": data2},
    },
)

Step 5: Run Tests

source .venv/bin/activate
pytest tests/rules/<domain>/test_<file>.py -v

---

Important Guidelines

NEVER use self.logger in rules - Return error messages via RuleResult.failed() or RuleResult.warning() instead. The framework handles logging automatically.

Command Security - SafeCmdString:

REQUIRED for all run_cmd(), get_output_from_run_cmd(), and run_rsh_cmd() to prevent command injection.

Examples:

# Static command
self.run_cmd(SafeCmdString("systemctl status"))
# Named placeholder
cmd = SafeCmdString("cat {file}").format(file="/etc/hostname")
self.run_cmd(cmd)

# Positional placeholder
cmd = SafeCmdString("cat {}").format("/etc/hostname")
self.run_cmd(cmd)

# Concatenation with + operator
self.run_cmd(SafeCmdString("cat /etc/hostname") + SafeCmdString("| grep localhost"))

# SafeCmdString as variable (bypasses validation - already safe)
cmd1 = SafeCmdString("etcdctl version")
cmd2 = SafeCmdString("Running: {cmd}").format(cmd=cmd1)
self.oc_api.run_rsh_cmd(namespace, pod, cmd2)

Allowed patterns in format() variables:

• Absolute paths: /path/to/file or /path/to/file.ext (one dot max for extension)
• Generic identifiers: [a-zA-Z0-9][a-zA-Z0-9.- ]* (alphanumeric start, then letters/digits/dots/dashes/spaces)
• Etcd URLs: https://etcd-N.etcd.openshift-etcd.svc:2379/path
• PCI addresses: 01:00.0 or 0000:01:00.0

Pre-commit linter enforces:

• Template must be string literal (not variable/f-string/expression)
• One SafeCmdString per line (except SafeCmdString() + SafeCmdString() is allowed)

Step 6: Set Supported Profile

If the rule may be specific to a deployment type (e.g., telco, AI, Spectrum-X), ask the user which profile it should target before finalizing.

Check src/profiles/profiles.yaml for available profiles and their hierarchy. Only leave the default {"general"} if the rule applies to ALL cluster types.

class MyRuleName(Rule):
    supported_profiles = {"spectrum-x"}  # Only runs for spectrum-x profile

See "Supported Profiles" in @.claude/rules/in-cluster-check.md for details.

Checklist

• Rule class created with objective_hosts, unique_name, title
• Asked user about supported_profiles if rule may be specific to a deployment type
• run_rule() implemented returning RuleResult
• is_prerequisite_fulfilled() added if rule requires specific tools/conditions
• UnExpectedSystemOutput used for command failures
• NO use of self.logger in rule - return error messages via RuleResult
• All commands use SafeCmdString for run_cmd(), get_output_from_run_cmd(), run_rsh_cmd()
• Rule registered in the appropriate domain's get_rule_classes()
• Tests written with both scenario_passed and scenario_failed
• Domain test updated to include new rule
• All command strings in tests match EXACTLY what the rule executes
• Tests pass: pytest tests/ -v
• Imports at top of file, following PEP 8 order