Gabriel Brown
e0c0e53ae0
A suite runner walking the tests directory executes it with no arguments and gets an IndexError, which reads as a failing test rather than a helper being used wrongly. Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
299 lines
12 KiB
Python
299 lines
12 KiB
Python
#!/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:
|
|
# Driven by settings-write-sweep-contract.sh, which starts the harness this
|
|
# needs and restores the desktop afterwards. Run bare -- by a suite runner
|
|
# walking the directory, say -- it says so instead of failing on argv.
|
|
if len(sys.argv) < 3:
|
|
print("settings_write_sweep is driven by tests/quickshell/"
|
|
"settings-write-sweep-contract.sh; run that instead.")
|
|
return 0
|
|
|
|
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())
|