glances-rs/tools/parity.py

#!/usr/bin/env python3
"""Differential test harness: compares glances-rs stdout-json output with the
Python Glances implementation (the reference oracle).

Usage:
  tools/parity.py [--rust BIN] [--python BIN] [--iterations N] [plugin ...]

With no plugin list, a default set of core plugins is tested.
"""

import argparse
import json
import os
import subprocess
import sys

DEFAULT_PLUGINS = [
    "cpu",
    "percpu",
    "mem",
    "memswap",
    "load",
    "fs",
    "network",
    "diskio",
    "sensors",
    "now",
    "uptime",
    "system",
    "processcount",
    "version",
    "psutilversion",
    "quicklook",
]


def flatten(value, prefix=""):
    out = {}
    if isinstance(value, dict):
        for key, item in value.items():
            out.update(flatten(item, f"{prefix}{key}."))
    elif isinstance(value, list):
        key_field = value[0].get("key") if value and isinstance(value[0], dict) else None
        for index, item in enumerate(value):
            identity = item.get(key_field) if key_field and isinstance(item, dict) else index
            out.update(flatten(item, f"{prefix}{identity}."))
    else:
        out[prefix[:-1]] = value
    return out


def type_name(value):
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "bool"
    if isinstance(value, (int, float)):
        return "number"
    if isinstance(value, str):
        return "string"
    if isinstance(value, list):
        return "array"
    if isinstance(value, dict):
        return "object"
    return type(value).__name__


def run(binary, plugins, iterations, extra):
    plugins_arg = ",".join(plugins)
    cmd = [
        binary,
        "--stdout-json",
        plugins_arg,
        "--disable-plugin",
        "all",
        "--enable-plugin",
        plugins_arg,
        "--stop-after",
        str(iterations),
        "-t",
        "1",
        "--disable-check-update",
    ] + extra
    env = dict(os.environ)
    env["PYTHONUNBUFFERED"] = "1"
    result = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=120)
    if result.returncode != 0:
        return None, result.stderr.strip().splitlines()[-3:]
    lines = [line for line in result.stdout.splitlines() if line.strip()]
    if not lines:
        return None, ["no output"]
    return json.loads(lines[-1]), None


ORACLE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "glances_oracle.py")


def run_python(python, conf, plugins, iterations, extra):
    plugins_arg = ",".join(plugins)
    cmd = [
        python,
        ORACLE,
        "--iterations",
        str(iterations),
        "--refresh",
        "1",
        "--plugins",
        plugins_arg,
    ]
    if conf:
        cmd += ["--config", conf]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    if result.returncode != 0:
        return None, result.stderr.strip().splitlines()[-3:]
    lines = [line for line in result.stdout.splitlines() if line.strip()]
    if not lines:
        return None, ["no output"]
    return json.loads(lines[-1]), None


def compare(plugin, rust, python, errors):
    rust_plugin = rust.get(plugin, {})
    python_plugin = python.get(plugin, {})
    rust_flat = flatten(rust_plugin)
    python_flat = flatten(python_plugin)

    if plugin not in rust:
        errors.append(f"{plugin}: missing in Rust output")
        return
    if plugin not in python:
        return

    for key in sorted(set(rust_flat) | set(python_flat)):
        if key not in python_flat:
            errors.append(f"{plugin}.{key}: only in Rust ({rust_flat[key]!r})")
        elif key not in rust_flat:
            errors.append(f"{plugin}.{key}: only in Python ({python_flat[key]!r})")
        else:
            rust_type = type_name(rust_flat[key])
            python_type = type_name(python_flat[key])
            if rust_type != python_type and not (
                rust_type == "null" or python_type == "null"
            ):
                errors.append(
                    f"{plugin}.{key}: type mismatch Rust={rust_type} Python={python_type} "
                    f"({rust_flat[key]!r} vs {python_flat[key]!r})"
                )


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--rust", default="./target/debug/glances")
    parser.add_argument("--python", default="/tmp/opencode/glances-py/bin/python")
    parser.add_argument("--config", default="conf/glances.conf")
    parser.add_argument("--iterations", type=int, default=3)
    parser.add_argument("plugins", nargs="*")
    args = parser.parse_args()

    plugins = args.plugins or DEFAULT_PLUGINS
    errors = []
    for plugin in plugins:
        extra = ["-C", args.config] if args.config else []
        rust, rust_error = run(args.rust, [plugin], args.iterations, extra)
        if rust_error is not None:
            errors.append(f"{plugin}: Rust run failed: {rust_error}")
            continue
        python, python_error = run_python(args.python, args.config, [plugin], args.iterations, [])
        if python_error is not None:
            errors.append(f"{plugin}: Python run failed: {python_error}")
            continue
        compare(plugin, rust, python, errors)
        status = "OK" if not any(e.startswith(f"{plugin}.") or e.startswith(f"{plugin}:") for e in errors) else "DIFF"
        print(f"  {plugin}: {status}")

    if errors:
        print("\nDifferences:")
        for error in errors:
            print(f"  - {error}")
        return 1
    print("\nNo differences found")
    return 0


if __name__ == "__main__":
    sys.exit(main())