vacuum-wall/lib/common.py

"""Shared utilities for Vacuum Wall lib/ modules.

Provides common helpers for JSON persistence, subprocess execution,
deep merging, and directory creation used across all subsystem modules.
"""

import hashlib
import json
import os
import re
import subprocess
from copy import deepcopy
from pathlib import Path
from typing import Any

_APPLY_HASH_KEY = "_last_applied_hash"


def config_hash(cfg: dict[str, Any]) -> str:
    """Compute a SHA-256 hash of *cfg* excluding the ``_last_applied_hash`` key.

    Args:
        cfg: Config dict, possibly containing ``_last_applied_hash``.

    Returns:
        Hex digest of the stripped config JSON.
    """
    clean = {k: v for k, v in cfg.items() if k != _APPLY_HASH_KEY}
    return hashlib.sha256(json.dumps(clean, sort_keys=True).encode()).hexdigest()


def validate_interface_name(name: str) -> str:
    """Validate a Linux network interface name.

    Args:
        name: Interface name to validate.

    Returns:
        The validated (stripped) name.

    Raises:
        ValueError: When the name is empty, contains path components,
            or does not match Linux interface naming rules.
    """
    if not name or not isinstance(name, str):
        raise ValueError("Interface name must be a non-empty string")
    name = name.strip()
    if not name:
        raise ValueError("Interface name must not be blank")
    if "/" in name or ".." in name or " " in name:
        raise ValueError(f"Invalid interface name: {name!r}")
    if not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*$", name):
        raise ValueError(f"Invalid interface name: {name!r}")
    return name


def run(
    cmd: list[str],
    check: bool = True,
    sudo: bool = False,
    timeout: int | None = None,
) -> str:
    """Run a command and return stripped stdout.

    Args:
        cmd: Command arguments.
        check: Raise RuntimeError on non-zero exit.
        sudo: Prefix command with ``sudo``.
        timeout: Timeout in seconds (``None`` → no timeout).

    Returns:
        ``stdout`` with trailing whitespace removed.

    Raises:
        RuntimeError: When ``check=True`` and the process exits non-zero.
    """
    full_cmd = ["sudo", *cmd] if sudo else list(cmd)
    try:
        result = subprocess.run(
            full_cmd,
            capture_output=True,
            text=True,
            check=check,
            timeout=timeout,
        )
        return result.stdout.strip()
    except subprocess.CalledProcessError as exc:
        raise RuntimeError(
            f"Command failed: {' '.join(full_cmd)} (rc={exc.returncode}): "
            f"{exc.stderr.strip()}"
        ) from exc


def run_proc(
    cmd: list[str],
    check: bool = True,
    sudo: bool = False,
    timeout: int | None = None,
    input: str | None = None,
) -> subprocess.CompletedProcess[str]:
    """Run a command and return the full ``CompletedProcess``.

    Args:
        cmd: Command arguments.
        check: Raise ``subprocess.CalledProcessError`` on non-zero exit.
        sudo: Prefix command with ``sudo``.
        timeout: Timeout in seconds.
        input: String to pass as stdin to the subprocess.

    Returns:
        The completed process object.
    """
    full_cmd = ["sudo", *cmd] if sudo else list(cmd)
    return subprocess.run(
        full_cmd,
        capture_output=True,
        text=True,
        check=check,
        timeout=timeout,
        input=input,
    )


def load_json(path: Path, default: dict[str, Any] | None = None) -> dict[str, Any]:
    """Load JSON from *path*.

    Returns *default* (default ``{}``) if the file does not exist.
    """
    if default is None:
        default = {}
    if not path.exists():
        return deepcopy(default)
    with open(path) as f:
        return json.load(f)


def save_json(path: Path, data: dict[str, Any], indent: int = 4) -> None:
    """Atomically write *data* as JSON to *path*.

    Writes to ``path.tmp`` first, then replaces *path* via ``os.replace()``
    to avoid partial writes.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    with open(tmp, "w") as f:
        json.dump(data, f, indent=indent)
        f.write("\n")
    os.replace(tmp, path)


def deep_merge(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
    """Recursively merge *overrides* into a deep copy of *base*.

    For nested dicts the merge recurses; for all other values
    *overrides* wins.
    """
    result = deepcopy(base)
    for k, v in overrides.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = deep_merge(result[k], v)
        else:
            result[k] = deepcopy(v)
    return result


def ensure_dirs(*dirs: Path) -> None:
    """Create each directory (and parents) if it does not exist."""
    for d in dirs:
        d.mkdir(parents=True, exist_ok=True)


__all__ = [
    "_APPLY_HASH_KEY",
    "config_hash",
    "deep_merge",
    "ensure_dirs",
    "load_json",
    "run",
    "run_proc",
    "save_json",
    "validate_interface_name",
]