werkrevisor/tools/host_config_block.py

"""Compares or replaces the inline host config WERKREVISOR_CONFIG_TOML in the machine configs of a Werkator host.

Usage on the host: python3 host_config_block.py check|apply < {"<repo>": "<content of config/<repo>.toml>", ...}
It prints line numbers and TOML key names of the block only, never values, since a machine config holds tokens.
Exit code: 0 if every block equals its file, 1 if a block differs in check mode, 2 if a block is missing.
"""
import datetime
import difflib
import json
import os
import re
import shutil
import sys
import tomllib

# The chomping indicators only change the last line break, which the review does not read.
KEY = re.compile(r"^(\s*)WERKREVISOR_CONFIG_TOML: \|[-+]?\s*$")
BACKUP = ".werkator.yml.bak-config-"


class NoBlock(Exception):
    pass


def _indent(line):
    return len(line) - len(line.lstrip())


def find_block(lines):
    """Returns (first, end, indent) of the block lines; trailing blank lines belong to the YAML around it."""
    starts = [i for i, line in enumerate(lines) if KEY.match(line)]
    if len(starts) != 1:
        raise NoBlock(f"{len(starts)} WERKREVISOR_CONFIG_TOML blocks")
    key = _indent(lines[starts[0]])
    first = end = starts[0] + 1
    while end < len(lines) and (not lines[end].strip() or _indent(lines[end]) > key):
        end += 1
    while end > first and not lines[end - 1].strip():
        end -= 1
    if end == first:
        raise NoBlock("an empty WERKREVISOR_CONFIG_TOML block")
    return first, end, min(_indent(line) for line in lines[first:end] if line.strip())


def block_text(lines, first, end, indent):
    return [line[indent:] if line.strip() else "" for line in lines[first:end]]


def differing_lines(old, new):
    """Line ranges of the old block, 1-based, that the new content changes; an insertion names the line before it."""
    ranges = []
    for tag, a1, a2, _, _ in difflib.SequenceMatcher(None, old, new, autojunk=False).get_opcodes():
        if tag != "equal":
            ranges.append(f"after {a1}" if a2 == a1 else f"{a1 + 1}" if a2 == a1 + 1 else f"{a1 + 1}-{a2}")
    return ranges


def _flat(table, prefix=""):
    flat = {}
    for key, value in table.items():
        if isinstance(value, dict):
            flat.update(_flat(value, f"{prefix}{key}."))
        else:
            flat[f"{prefix}{key}"] = value
    return flat


def differing_keys(old, new):
    """Names the keys by kind of difference, so that a drift shows its meaning without a value."""
    try:
        host, file = _flat(tomllib.loads("\n".join(old))), _flat(tomllib.loads("\n".join(new)))
    except tomllib.TOMLDecodeError:
        return "invalid TOML"
    groups = (("changed", sorted(k for k in host.keys() & file.keys() if host[k] != file[k])),
              # A machine config may leave out the keys that the .werkrevisor.toml of its repository sets.
              ("only in the file", sorted(file.keys() - host.keys())),
              ("only on the host", sorted(host.keys() - file.keys())))
    named = [f"{name}: {', '.join(keys)}" for name, keys in groups if keys]
    return "; ".join(named) if named else "no key, comments or layout only"


def replaced(lines, first, end, indent, new):
    return lines[:first] + [" " * indent + line if line else "" for line in new] + lines[end:]


def backups_to_remove(names, keep=2):
    return sorted((n for n in names if n.startswith(BACKUP)), reverse=True)[keep:]


def sync(path, content, apply, now):
    """Returns (status, lines) for one machine config; status is "equal", "differs", "replaced" or "no block"."""
    try:
        with open(path) as f:
            lines = f.read().split("\n")
    except FileNotFoundError:
        return "no block", ["no machine config"]
    try:
        first, end, indent = find_block(lines)
    except NoBlock as e:
        return "no block", [str(e)]
    new = content.rstrip("\n").split("\n")
    old = block_text(lines, first, end, indent)
    changed = differing_lines(old, new)
    if not changed:
        return "equal", []
    changed = [f"lines {', '.join(changed)}; {differing_keys(old, new)}"]
    if not apply:
        return "differs", changed
    shutil.copy2(path, os.path.join(os.path.dirname(path), BACKUP + now))
    # Writing into the existing file keeps its mode 600.
    with open(path, "w") as f:
        f.write("\n".join(replaced(lines, first, end, indent, new)))
    directory = os.path.dirname(path)
    for old in backups_to_remove(os.listdir(directory)):
        os.remove(os.path.join(directory, old))
    return "replaced", changed


def main(argv, stdin, home, now=None):
    mode = argv[0] if argv else ""
    if mode not in ("check", "apply"):
        print(__doc__.strip(), file=sys.stderr)
        return 2
    now = now or datetime.datetime.now().strftime("%Y%m%d%H%M%S")
    worst = 0
    for repo, content in json.load(stdin).items():
        path = os.path.join(home, repo, ".git", "werkator", ".werkator.yml")
        status, details = sync(path, content, mode == "apply", now)
        listed = f": {details[0]}" if status in ("differs", "replaced") else ""
        reason = f": {details[0]}" if status == "no block" else ""
        backup = f"; backup {BACKUP}{now}" if status == "replaced" else ""
        print(f"{repo}: {status}{listed}{reason}{backup}")
        worst = max(worst, {"equal": 0, "replaced": 0, "differs": 1, "no block": 2}[status])
    return worst


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:], sys.stdin, os.path.expanduser("~")))