Panama/tests/quickshell/settings_write_sweep.py

#!/usr/bin/env python3

"""Flip every compositor-backed setting, read it back, and put it back.

The existing write contract does this for five settings. There are sixty-five,
and the ones it does not cover are exactly where a silently-dead setting would
live: nothing fails, nothing logs, the row just does not do anything.

Each setting is driven through SystemSettings.commitPreference -- the same entry
point a settings row uses -- so this exercises the real routing, the real
compositor write, and the real read-back verification, not a shape.

Every original value is captured before anything changes and restored at the
end, through hyprctl directly, so a broken write path cannot also break the
restore.
"""

from __future__ import annotations

import ast
import json
import os
import re
import subprocess
import sys
import time

SCHEMA = "config/dot/quickshell/config/PreferenceSchema.qml"

# Read-back shapes this sweep knows how to compare. A gradient or a vec2 has no
# single scalar to diff, and guessing one would produce false failures; those
# are reported as skipped rather than quietly counted as verified.
COMPARABLE = {"bool", "int", "float", "str"}

# Settings whose value this must not choose freely.
#
# kb_layout is the one that can lock someone out of their own keyboard if a
# sweep picks a layout they cannot type in, and the existing write contract
# already covers it deliberately.
SKIP_KEYS = {"keyboardLayout", "keyboardVariant", "keyboardOptions", "keyboardModel"}


def run(command: list[str], timeout: float = 20.0) -> subprocess.CompletedProcess:
    return subprocess.run(command, capture_output=True, text=True,
                          timeout=timeout, check=False)


def local_entries() -> list[dict]:
    """User-facing settings that Panama stores itself.

    These never reach the compositor, so the compositor cannot be asked whether
    they took. What can be asked is whether the value came back out of the
    preference store -- a setting that silently fails to persist is the same
    dead row from the user's side.
    """
    source = open(SCHEMA, encoding="utf-8").read()
    entries = []
    for chunk in source.split("key: ")[1:]:
        head = chunk[:1400]
        if re.search(r"internal:\s*true", head) or re.search(r"hypr:\s*\{", head):
            continue
        key = re.match(r'"([A-Za-z0-9_]+)"', chunk)
        kind = re.search(r'type:\s*"([a-z]+)"', head)
        if not (key and kind) or kind.group(1) == "json":
            continue
        entry = {"key": key.group(1), "type": kind.group(1)}
        for bound in ("min", "max", "step"):
            found = re.search(rf"\b{bound}:\s*(-?[0-9.]+)", head)
            if found:
                entry[bound] = float(found.group(1))
        options = re.search(r"options:\s*\[(.*?)\]", head, re.S)
        if options:
            entry["values"] = [
                ast.literal_eval(value) if value.lstrip().startswith(("'", '"'))
                else int(value)
                for value in re.findall(r"value:\s*(\"[^\"]*\"|-?\d+)", options.group(1))
            ]
        entries.append(entry)
    return entries


def schema_entries() -> list[dict]:
    """Every user-facing setting that maps onto a compositor option."""
    source = open(SCHEMA, encoding="utf-8").read()
    entries = []
    for chunk in source.split("key: ")[1:]:
        head = chunk[:1400]
        if re.search(r"internal:\s*true", head):
            continue
        hypr = re.search(r"hypr:\s*\{(.*?)\n\s*\}", head, re.S)
        if not hypr:
            continue
        key = re.match(r'"([A-Za-z0-9_]+)"', chunk)
        kind = re.search(r'type:\s*"([a-z]+)"', head)
        option = re.search(r'option:\s*"([^"]+)"', hypr.group(1))
        read_as = re.search(r'readAs:\s*"([a-z]+)"', hypr.group(1))
        if not (key and kind and option):
            continue
        entry = {
            "key": key.group(1),
            "type": kind.group(1),
            "option": option.group(1),
            "readAs": read_as.group(1) if read_as else "int",
            "invert": bool(re.search(r"invert:\s*true", hypr.group(1))),
        }
        for bound in ("min", "max", "step"):
            found = re.search(rf"\b{bound}:\s*(-?[0-9.]+)", head)
            if found:
                entry[bound] = float(found.group(1))
        options = re.search(r"options:\s*\[(.*?)\]", head, re.S)
        if options:
            entry["values"] = [
                ast.literal_eval(value) if value.lstrip().startswith(("'", '"'))
                else int(value)
                for value in re.findall(r"value:\s*(\"[^\"]*\"|-?\d+)", options.group(1))
            ]
        entries.append(entry)
    return entries


def read_option(entry: dict):
    """What the compositor currently holds, in the shape the schema declares."""
    result = run(["hyprctl", "-j", "getoption", entry["option"]])
    if result.returncode != 0:
        return None
    try:
        payload = json.loads(result.stdout)
    except json.JSONDecodeError:
        return None
    field = {"bool": "bool", "int": "int", "float": "float", "str": "str"}.get(entry["readAs"])
    if field is None or field not in payload:
        return None
    return payload[field]


def stored_from_option(entry: dict, raw):
    """The value a preference would hold for this compositor reading."""
    if entry["readAs"] == "bool":
        value = bool(raw)
        return (not value) if entry["invert"] else value
    if entry["readAs"] == "int" and entry["type"] == "bool":
        value = bool(raw)
        return (not value) if entry["invert"] else value
    return raw


def pick_target(entry: dict, current):
    """A valid value this setting does not currently hold."""
    if entry["type"] == "bool":
        return not bool(current)
    if entry["type"] == "enum":
        for candidate in entry.get("values", []):
            if candidate != current:
                return candidate
        return None
    if entry["type"] in ("int", "real"):
        low = entry.get("min", 0.0)
        high = entry.get("max", max(low + 1.0, float(current or 0) + 1.0))
        step = entry.get("step", 1.0)
        candidate = float(current or 0) + step
        if candidate > high:
            candidate = float(current or 0) - step
        if candidate < low:
            candidate = low if low != current else high
        if entry["type"] == "int":
            candidate = int(round(candidate))
        if candidate == current:
            return None
        return candidate
    return None


def ipc(config_home: str, harness: str, *arguments: str) -> subprocess.CompletedProcess:
    environment = dict(os.environ, XDG_CONFIG_HOME=config_home)
    return subprocess.run(["qs", "-p", harness, "ipc", "call", "settings-system-test", *arguments],
                          capture_output=True, text=True, timeout=30, check=False,
                          env=environment)


def main() -> int:
    config_home = sys.argv[1]
    harness = sys.argv[2]

    entries = [entry for entry in schema_entries() if entry["key"] not in SKIP_KEYS]
    verified, skipped, failures, originals = [], [], [], {}

    for entry in entries:
        raw = read_option(entry)
        if raw is None or entry["readAs"] not in COMPARABLE:
            skipped.append((entry["key"], f"read-back shape {entry['readAs']}"))
            continue

        current = stored_from_option(entry, raw)
        originals[entry["key"]] = (entry, raw)
        target = pick_target(entry, current)
        if target is None:
            skipped.append((entry["key"], "no second valid value"))
            continue

        # Whatever happens below, this setting goes back to where it was
        # before the next one is touched. A sweep that dies holding sixty
        # settings at test values would be worse than no sweep.
        try:
            answer = ipc(config_home, harness, "commit", entry["key"], json.dumps(target))
            if answer.returncode != 0:
                failures.append((entry["key"], f"commit failed: {answer.stderr.strip()[:80]}"))
                continue
            if answer.stdout.strip().lower() not in ("true", ""):
                failures.append((entry["key"], f"commit refused the value ({answer.stdout.strip()[:40]})"))
                continue

            # The compositor is asked again for up to a second. commitPreference
            # verifies before storing, so a correct write is usually visible
            # immediately -- but "usually" makes a flaky test, and a flaky test
            # is worse than none. A write that never happens never matches, no
            # matter how long this waits.
            after = None
            matched = False
            deadline = time.monotonic() + 1.0
            while True:
                after = stored_from_option(entry, read_option(entry))
                if isinstance(target, float) or isinstance(after, float):
                    matched = after is not None and abs(float(after) - float(target)) < 0.02
                else:
                    matched = after == target
                if matched or time.monotonic() > deadline:
                    break
                time.sleep(0.05)
            if not matched:
                failures.append((entry["key"],
                                 f"set to {target!r} but the compositor reports {after!r} "
                                 f"({entry['option']})"))
                continue
            verified.append(entry["key"])
        finally:
            ipc(config_home, harness, "commit", entry["key"], json.dumps(current))

    # ── Settings Panama stores itself ────────────────────────────────────────
    local_verified, local_skipped, local_failures = [], [], []
    for entry in local_entries():
        raw = ipc(config_home, harness, "stored", entry["key"])
        if raw.returncode != 0:
            local_failures.append((entry["key"], "could not be read from the store"))
            continue
        try:
            current = json.loads(raw.stdout.strip() or "null")
        except json.JSONDecodeError:
            local_failures.append((entry["key"], f"stored value is not readable: {raw.stdout.strip()[:40]}"))
            continue

        target = pick_target(entry, current)
        if target is None:
            local_skipped.append((entry["key"], "no second valid value"))
            continue

        try:
            answer = ipc(config_home, harness, "commit", entry["key"], json.dumps(target))
            if answer.returncode != 0 or answer.stdout.strip().lower() not in ("true", ""):
                local_failures.append((entry["key"], "the store refused the value"))
                continue
            back = ipc(config_home, harness, "stored", entry["key"])
            try:
                after = json.loads(back.stdout.strip() or "null")
            except json.JSONDecodeError:
                after = None
            if isinstance(target, float) or isinstance(after, float):
                matched = after is not None and abs(float(after) - float(target)) < 0.02
            else:
                matched = after == target
            if not matched:
                local_failures.append((entry["key"], f"set to {target!r} but the store reports {after!r}"))
                continue
            local_verified.append(entry["key"])
        finally:
            ipc(config_home, harness, "commit", entry["key"], json.dumps(current))

    print(json.dumps({
        "verified": verified,
        "skipped": skipped,
        "failures": failures,
        "total": len(entries),
        "localVerified": local_verified,
        "localSkipped": local_skipped,
        "localFailures": local_failures,
    }))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())