mr-kot 0.7.0

Pytest-inspired invariant checker

Mr. Kot

Mr. Kot is a pytest-inspired invariant checker. It is designed to describe and verify system invariants: conditions that must hold for a system to remain functional.

Mr. Kot is specialized for health checks. It provides:

• Facts: small functions that describe system state.
• Checks: functions that use facts (and optionally fixtures) to verify invariants.
• Selectors: conditions based on facts that decide whether a check should run.
• Fixtures: reusable resources injected into checks with setup/teardown support.
• Parametrization: run the same check with multiple values or fact-provided inputs.
• Runner: an engine that resolves facts, applies selectors, expands parametrization, runs checks, and produces machine-readable results.

---

Concepts

Facts

Facts provide info that can be used by checks and other facts. They are registered with @fact. Facts may depend on other facts via function parameters, and are memoized per run.

Example:

@fact
def os_release():
    return {"id": "ubuntu", "version": "22.04"}

# fact name in the function parameters means dependency on it
@fact
def os_is_ubuntu(os_release: dict) -> bool:
    return os_release["id"] == "ubuntu"

Checks

Checks verify invariants. They are registered with @check. Checks must return a tuple (status, evidence) where status is a Status enum: PASS, FAIL, WARN, SKIP, or ERROR.

You can use fact values inside a check to make a decision and craft evidence:

from mr_kot import check, Status, fact
@fact
def cpu_count() -> int:
    import os
    return os.cpu_count() or 1

@check
def has_enough_cpus(cpu_count: int):
    required = 4
    if cpu_count >= required:
        return (Status.PASS, f"cpus={cpu_count} (>= {required})")
    return (Status.FAIL, f"cpus={cpu_count} (< {required})")

If you want a fact or fixture for its side effect (e.g. you don't need its value inside the function), use @depends instead of adding it as a function parameter.

from mr_kot import check, depends, fixture, Status

@fixture
def side_effectful_fixture():
    mount("/data")
    yield True
    umount("/data")

@check
@depends("side_effectful_fixture")
def fs_write_smoke():
    with open("/data/test", "w") as f:
        f.write("ok")
        return (Status.PASS, "ok")

Selectors

Selectors decide whether a check instance should run, based on fact values. They are passed as a parameter of @check(selector=...). If selector is not passed, the check runs unconditionally after parametrization.

There are two forms of the selectors:

• String shorthand (recommended for common cases): a comma-separated list of fact names. All listed facts must exist and evaluate truthy for the check to run. This is equivalent to ALL(<fact_name>, <fact_name>, ...).

• Predicate callable (advanced): a function taking facts as arguments and returning a boolean. There are helper predicates available: ALL, ANY, NOT.

• Only facts are allowed in selectors; fixtures are not allowed.

• Facts used solely as check arguments (not in the selector) are produced during execution; if they fail, that instance becomes ERROR and the run continues.

Helper predicates (for common boolean checks):

from mr_kot import check, Status, ALL, ANY, NOT

# Run only if both boolean facts are truthy
@check(selector=ALL("has_systemd", "has_network"))
def service_reachable(unit: str):
    return (Status.PASS, f"unit={unit}")

# Run if any of the flags is truthy
@check(selector=ANY("has_systemd", "has_sysvinit"))
def service_manager_present():
    return (Status.PASS, "present")

# Negate another predicate
@check(selector=NOT(ALL("maintenance_mode")))
def system_not_in_maintenance():
    return (Status.PASS, "ok")

Advanced selectors (predicate)

Use a predicate when you need to inspect values. Predicates are evaluated with facts only (fixtures are not allowed) and must return a boolean.

from mr_kot import check

from mr_kot import check, Status

@check(selector=lambda os_release: os_release["id"] == "ubuntu")
def ubuntu_version_is_supported(os_release):
    """Pass if Ubuntu version is >= 20.04, else fail.

    Selector fail-fast guarantees os_release exists and was produced without error.
    """
    def _parse(v: str) -> tuple[int, int]:
        parts = (v.split(".") + ["0", "0"])[:2]
        try:
            return int(parts[0]), int(parts[1])
        except Exception:
            return (0, 0)

    min_major, min_minor = (20, 4)
    major, minor = _parse(os_release.get("version", "0.0"))  # type: ignore[call-arg]
    if (major, minor) >= (min_major, min_minor):
        return (Status.PASS, f"ubuntu {major}.{minor} >= {min_major}.{min_minor}")
    return (Status.FAIL, f"ubuntu {major}.{minor} < {min_major}.{min_minor}")

Notes:

• Selectors are evaluated per-instance after parametrization expansion.
• If a selector evaluates to False for an instance, the runner emits a SKIP item with evidence selector=false.
• Unknown fact name in a selector (or helper) → planning error, run aborts.
• Fact production error during selector evaluation → planning error, run aborts.
• Fixtures are not allowed in selectors.
• Facts used only as check arguments are produced at execution; failures mark that instance ERROR and the run continues.

Fixtures

Fixtures are reusable resources. They are registered with @fixture. They can return a value directly, or yield a value and perform teardown afterward. For now, fixtures are per-check: each check call receives a fresh instance.

Example:

@fixture
def tmp_path():
    import tempfile, shutil
    path = tempfile.mkdtemp()
    try:
        yield path
    finally:
        shutil.rmtree(path)

@check
def can_write_tmp(tmp_path):
    import os
    test_file = os.path.join(tmp_path, "test")
    with open(test_file, "w") as f:
        f.write("ok")
    return (Status.PASS, f"wrote to {test_file}")

Parametrization

Checks can be expanded into multiple instances with different arguments using @parametrize.

Inline values:

@parametrize("mount", values=["/data", "/logs"])
@check
def mount_present(mount):
    import os
    if os.path.exists(mount):
        return (Status.PASS, f"{mount} present")
    return (Status.FAIL, f"{mount} missing")

Values from a fact:

@fact
def systemd_units():
    return ["cron.service", "sshd.service"]

@parametrize("unit", source="systemd_units")
@check
def unit_active(unit):
    return (Status.PASS, f"{unit} is active")

Use fail_fast=True in @parametrize to stop executing remaining instances of the same check after the first FAIL or ERROR:

@parametrize("mount", values=["/data", "/data/logs"], fail_fast=True)
@check
def mount_present(mount):
    import os
    return (Status.PASS, f"{mount} present") if os.path.exists(mount) else (Status.FAIL, f"{mount} missing")

When fail-fast triggers, remaining instances are emitted as SKIP.

Runner

The runner discovers all facts, fixtures, and checks, evaluates selectors, expands parametrization, resolves dependencies, executes checks, and collects results.

Output structure:

{
  "overall": "PASS",
  "counts": {"PASS": 2, "FAIL": 1, "WARN": 0, "SKIP": 0, "ERROR": 0},
  "items": [
    {"id": "os_is_ubuntu", "status": "PASS", "evidence": "os=ubuntu"},
    {"id": "mount_present[/data]", "status": "PASS", "evidence": "/data present"},
    {"id": "mount_present[/logs]", "status": "FAIL", "evidence": "/logs missing"}
  ]
}

The `overall` field is computed by severity ordering: `ERROR > FAIL > WARN > PASS`.

---

## Plugins

Mr. Kot can discover and load external plugins that register facts, fixtures, and checks at import time.

- **Discovery sources**
  - Entry points: installed distributions that declare the entry-point group `mr_kot.plugins` are discovered and imported.
  - Explicit list: pass modules via the CLI `--plugins pkg1,pkg2` to import them first, in order.

- **Order and dedup**
  - CLI `--plugins` are imported first (left to right), then entry-point plugins sorted by their entry-point name.
  - If the same module appears in both, it is imported only once.

- **Uniqueness rule**
  - IDs (function names) must be unique across all loaded plugins and the current module; collisions abort the run.