From e5ff5025c34fa093c63c355a26b4ec9e3af03cea Mon Sep 17 00:00:00 2001 From: Adam Setch Date: Sat, 3 Oct 2026 22:26:35 -0400 Subject: [PATCH] refactor(sonar): extract helpers to reduce main() complexity (python:S3776) --- scripts/test-live.py | 228 +++++++++++++++++++++++++------------------ 1 file changed, 131 insertions(+), 97 deletions(-) diff --git a/scripts/test-live.py b/scripts/test-live.py index c7a530d..62970a8 100644 --- a/scripts/test-live.py +++ b/scripts/test-live.py @@ -52,27 +52,20 @@ def until(predicate, description, seconds=10): raise AssertionError(f"Timed out waiting for {description}") -def main(): - call("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", - "RequestName", "(su)", (SERVICE, 0)) - xml = f'' - registration = BUS.register_object("/Test", Gio.DBusNodeInfo.new_for_xml(xml).interfaces[0], receive, None, None) - plugin = f"gitify-live-test-{os.getpid()}" - probe_name = f"{plugin}-probe" - installed, = scripting("isScriptLoaded", "(s)", ("gitify",)) - process = None - try: - with tempfile.TemporaryDirectory(prefix="gitify-kwin-test-") as directory: - temporary = Path(directory) - env = dict(os.environ, XDG_CONFIG_HOME=str(temporary / "config"), - XDG_CACHE_HOME=str(temporary / "cache"), XDG_DATA_HOME=str(temporary / "data")) - env.pop("ELECTRON_RUN_AS_NODE", None) - with (temporary / "app.log").open("w+") as log: - process = subprocess.Popen(["gitify", "--inspect=127.0.0.1:0"], env=env, stdout=log, stderr=log) - - def inspect(mutation=""): - reports.clear() - source = f""" +def start_gitify(temporary): + """Launch Gitify against isolated XDG directories, logging to app.log.""" + env = dict(os.environ, XDG_CONFIG_HOME=str(temporary / "config"), + XDG_CACHE_HOME=str(temporary / "cache"), XDG_DATA_HOME=str(temporary / "data")) + env.pop("ELECTRON_RUN_AS_NODE", None) + log = (temporary / "app.log").open("w+") + process = subprocess.Popen(["gitify", "--inspect=127.0.0.1:0"], env=env, stdout=log, stderr=log) + return process, log + + +def inspect(process, temporary, probe_name, mutation=""): + """Load the probe script into KWin and return the latest report it writes.""" + reports.clear() + source = f""" const targets = workspace.windowList().filter(w => w.pid === {process.pid} && w.caption === 'Gitify'); {mutation} function rect(r) {{ return {{ x:r.x, y:r.y, width:r.width, height:r.height }}; }} @@ -81,78 +74,81 @@ def inspect(mutation=""): panels: workspace.windowList().filter(w => w.dock && String(w.resourceClass) === 'plasmashell').map(w => rect(w.frameGeometry)) }})); """ - probe = temporary / "probe.js" - probe.write_text(source) - try: - load_script(probe, probe_name) - until(lambda: reports, "KWin report", 5) - return reports[-1] - finally: - scripting("unloadScript", "(s)", (probe_name,)) - - def visible(): - return next((window for window in inspect()["windows"] if not window["hidden"]), None) - - def tray_item(): - items, = call("org.kde.StatusNotifierWatcher", "/StatusNotifierWatcher", - "org.freedesktop.DBus.Properties", "Get", "(ss)", - ("org.kde.StatusNotifierWatcher", "RegisteredStatusNotifierItems")) - for item in items: - service, _, path = item.partition("/") - try: - pid, = call("org.freedesktop.DBus", "/org/freedesktop/DBus", - "org.freedesktop.DBus", "GetConnectionUnixProcessID", "(s)", (service,)) - if pid == process.pid: - return service, "/" + path - except GLib.Error: - continue - return None - - tray_service, tray_path = until(tray_item, "Gitify's tray icon", 20) - def activate(): - call(tray_service, tray_path, "org.kde.StatusNotifierItem", "Activate", "(ii)", (0, 0)) - - if not visible(): - activate() - before = until(visible, "Gitify's initial window", 20) - print('Before:', json.dumps(before), flush=True) - if not installed: - load_script(ROOT / "gitify/contents/code/main.js", plugin) - print('After loading script:', json.dumps(inspect()), flush=True) - - def beside_panel(): - report = inspect() - for window in report["windows"]: - frame = window["frame"] - if window["hidden"]: - continue - for panel in report["panels"]: - gaps = [frame["y"] - panel["y"] - panel["height"], - panel["y"] - frame["y"] - frame["height"], - frame["x"] - panel["x"] - panel["width"], - panel["x"] - frame["x"] - frame["width"]] - if any(abs(gap - 8) < 1 for gap in gaps): - return window - return None - - after = until(beside_panel, "panel placement") - screen, frame = after["screen"], after["frame"] - assert frame["x"] >= screen["x"] and frame["y"] >= screen["y"] - assert frame["x"] + frame["width"] <= screen["x"] + screen["width"] - assert frame["y"] + frame["height"] <= screen["y"] + screen["height"] - if not installed: - assert after["frame"] != before["frame"], "Script did not change placement" - - log.flush() - # Gitify is launched with --inspect=127.0.0.1:0, so its DevTools inspector only binds - # to the loopback interface. The plain "ws" URL below therefore stays on this machine - # and cannot be reached from outside; the DevTools endpoint also offers no TLS. - endpoint = re.search(r"ws://127\.0\.0\.1:\d+/[\w-]+", (temporary / "app.log").read_text()).group() # NOSONAR(S5332) - expression = f"""(() => {{ + probe = temporary / "probe.js" + probe.write_text(source) + try: + load_script(probe, probe_name) + until(lambda: reports, "KWin report", 5) + return reports[-1] + finally: + scripting("unloadScript", "(s)", (probe_name,)) + + +def visible(process, temporary, probe_name): + """Return Gitify's first visible window, or None if it is hidden.""" + windows = inspect(process, temporary, probe_name)["windows"] + return next((window for window in windows if not window["hidden"]), None) + + +def tray_item(process): + """Return the StatusNotifier service and object path owned by Gitify.""" + items, = call("org.kde.StatusNotifierWatcher", "/StatusNotifierWatcher", + "org.freedesktop.DBus.Properties", "Get", "(ss)", + ("org.kde.StatusNotifierWatcher", "RegisteredStatusNotifierItems")) + for item in items: + service, _, path = item.partition("/") + try: + pid, = call("org.freedesktop.DBus", "/org/freedesktop/DBus", + "org.freedesktop.DBus", "GetConnectionUnixProcessID", "(s)", (service,)) + if pid == process.pid: + return service, "/" + path + except GLib.Error: + continue + return None + + +def beside_panel(process, temporary, probe_name): + """Return the Gitify window placed with the 8px gap next to a Plasma panel.""" + report = inspect(process, temporary, probe_name) + for window in report["windows"]: + if window["hidden"]: + continue + for panel in report["panels"]: + if has_8px_gap(window["frame"], panel): + return window + return None + + +def has_8px_gap(frame, panel): + gaps = [frame["y"] - panel["y"] - panel["height"], + panel["y"] - frame["y"] - frame["height"], + frame["x"] - panel["x"] - panel["width"], + panel["x"] - frame["x"] - frame["width"]] + return any(abs(gap - 8) < 1 for gap in gaps) + + +def activate_tray(tray_service, tray_path): + call(tray_service, tray_path, "org.kde.StatusNotifierItem", "Activate", "(ii)", (0, 0)) + + +def inspector_endpoint(temporary): + """Return the loopback-only Chrome DevTools inspector URL from Gitify's log.""" + # Gitify is launched with --inspect=127.0.0.1:0, so its DevTools inspector only binds + # to the loopback interface. The plain "ws" URL below therefore stays on this machine + # and cannot be reached from outside; the DevTools endpoint also offers no TLS. + log = (temporary / "app.log").read_text() + return re.search(r"ws://127\.0\.0\.1:\d+/[\w-]+", log).group() # NOSONAR(S5332) + + +def resize_expression(frame): + return f"""(() => {{ const electron = process.getBuiltinModule('module').createRequire(process.resourcesPath + '/app.asar/package.json')('electron'); electron.BrowserWindow.getAllWindows()[0].setSize({frame['width']}, {frame['height'] + 100}); }})()""" - subprocess.run(["node", "--input-type=module", "-e", """ + + +def resize_window(endpoint, expression): + subprocess.run(["node", "--input-type=module", "-e", """ const socket = new WebSocket(process.argv[1]); const timeout = setTimeout(() => { console.error('Inspector timed out'); process.exit(1); }, 5000); socket.onopen = () => socket.send(JSON.stringify({ id: 1, method: 'Runtime.evaluate', params: { expression: process.argv[2] } })); @@ -167,13 +163,49 @@ def beside_panel(): } }; """, endpoint, expression], check=True) - resized = until(lambda: (window := beside_panel()) and window["frame"]["height"] != frame["height"] and window, "anchored resize") - activate() - until(lambda: not visible(), "Gitify to hide") - activate() - reopened = until(beside_panel, "Gitify to reopen beside the panel") - print(json.dumps({"before": before, "after": after, "resized": resized, "reopened": reopened}, indent=2)) - print("Native KWin placement, resize, and tray reopen passed. Gitify ran with normal sandboxing.") + + +def main(): + call("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", + "RequestName", "(su)", (SERVICE, 0)) + xml = f'' + registration = BUS.register_object("/Test", Gio.DBusNodeInfo.new_for_xml(xml).interfaces[0], receive, None, None) + plugin = f"gitify-live-test-{os.getpid()}" + probe_name = f"{plugin}-probe" + installed, = scripting("isScriptLoaded", "(s)", ("gitify",)) + process = None + log = None + try: + with tempfile.TemporaryDirectory(prefix="gitify-kwin-test-") as directory: + temporary = Path(directory) + process, log = start_gitify(temporary) + tray_service, tray_path = until(lambda: tray_item(process), "Gitify's tray icon", 20) + + if not visible(process, temporary, probe_name): + activate_tray(tray_service, tray_path) + before = until(lambda: visible(process, temporary, probe_name), "Gitify's initial window", 20) + print('Before:', json.dumps(before), flush=True) + if not installed: + load_script(ROOT / "gitify/contents/code/main.js", plugin) + print('After loading script:', json.dumps(inspect(process, temporary, probe_name)), flush=True) + + after = until(lambda: beside_panel(process, temporary, probe_name), "panel placement") + screen, frame = after["screen"], after["frame"] + assert frame["x"] >= screen["x"] and frame["y"] >= screen["y"] + assert frame["x"] + frame["width"] <= screen["x"] + screen["width"] + assert frame["y"] + frame["height"] <= screen["y"] + screen["height"] + if not installed: + assert after["frame"] != before["frame"], "Script did not change placement" + + log.flush() + resize_window(inspector_endpoint(temporary), resize_expression(frame)) + resized = until(lambda: (window := beside_panel(process, temporary, probe_name)) and window["frame"]["height"] != frame["height"] and window, "anchored resize") + activate_tray(tray_service, tray_path) + until(lambda: not visible(process, temporary, probe_name), "Gitify to hide") + activate_tray(tray_service, tray_path) + reopened = until(lambda: beside_panel(process, temporary, probe_name), "Gitify to reopen beside the panel") + print(json.dumps({"before": before, "after": after, "resized": resized, "reopened": reopened}, indent=2)) + print("Native KWin placement, resize, and tray reopen passed. Gitify ran with normal sandboxing.") finally: scripting("unloadScript", "(s)", (plugin,)) if process is not None: @@ -183,6 +215,8 @@ def beside_panel(): except subprocess.TimeoutExpired: process.kill() process.wait() + if log is not None: + log.close() BUS.unregister_object(registration)