marketing-shibata50/cmux
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'origin/main' into task-browser-open-trailing-json-flags

Browse source 
# Conflicts:
#	CLI/cmux.swift
This commit is contained in:
Lawrence Chen 2026-03-05 20:03:19 -08:00
commit 0e7daae4a6
22 changed files with 2604 additions and 369 deletions
Download patch file Download diff file Expand all files Collapse all files

6
tests_v2/test_browser_cli_agent_port.py
View file

@ -203,6 +203,12 @@ def main() -> int:
snapshot_text = _run_cli_text(cli, ["browser", surface, "snapshot", "--interactive"])
_must("ref=e" in snapshot_text, f"Expected snapshot text with refs from CLI: {snapshot_text!r}")
blank_opened = _run_cli_json(cli, ["browser", "open", "about:blank", "--workspace", workspace])
blank_surface = str(blank_opened.get("surface_ref") or blank_opened.get("surface_id") or "")
_must(bool(blank_surface), f"Expected about:blank browser open to return a surface: {blank_opened}")
blank_snapshot = _run_cli_text(cli, ["browser", blank_surface, "snapshot", "--interactive"])
_must("about:blank" in blank_snapshot and "get url" in blank_snapshot, f"Expected empty snapshot diagnostics for about:blank: {blank_snapshot!r}")
opened_routed = _run_cli_json(cli, ["browser", "open", page_url, "--workspace", workspace])
routed_surface = str(opened_routed.get("surface_ref") or opened_routed.get("surface_id") or "")
_must(bool(routed_surface), f"browser open --workspace returned no surface handle: {opened_routed}")

146
tests_v2/test_browser_cli_wait_and_screenshot.py Normal file
View file

@ -0,0 +1,146 @@
#!/usr/bin/env python3
"""Regression: browser wait/snapshot and screenshot CLI return usable file locations."""
import glob
import json
import os
import subprocess
import sys
import tempfile
import urllib.parse
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from cmux import cmux, cmuxError
SOCKET_PATH = os.environ.get("CMUX_SOCKET", "/tmp/cmux-debug.sock")
def _must(cond: bool, msg: str) -> None:
if not cond:
raise cmuxError(msg)
def _find_cli_binary() -> str:
env_cli = os.environ.get("CMUXTERM_CLI")
if env_cli and os.path.isfile(env_cli) and os.access(env_cli, os.X_OK):
return env_cli
fixed = os.path.expanduser(
"~/Library/Developer/Xcode/DerivedData/cmux-tests-v2/Build/Products/Debug/cmux"
)
if os.path.isfile(fixed) and os.access(fixed, os.X_OK):
return fixed
candidates = glob.glob(
os.path.expanduser("~/Library/Developer/Xcode/DerivedData/**/Build/Products/Debug/cmux"),
recursive=True,
)
candidates += glob.glob("/tmp/cmux-*/Build/Products/Debug/cmux")
candidates = [p for p in candidates if os.path.isfile(p) and os.access(p, os.X_OK)]
if not candidates:
raise cmuxError("Could not locate cmux CLI binary; set CMUXTERM_CLI")
candidates.sort(key=lambda p: os.path.getmtime(p), reverse=True)
return candidates[0]
def _run_cli(cli: str, *args: str) -> subprocess.CompletedProcess[str]:
cmd = [cli, "--socket", SOCKET_PATH, *args]
proc = subprocess.run(cmd, capture_output=True, text=True, check=False)
if proc.returncode != 0:
merged = f"{proc.stdout}\n{proc.stderr}".strip()
raise cmuxError(f"CLI failed ({' '.join(cmd)}): {merged}")
return proc
def main() -> int:
cli = _find_cli_binary()
with cmux(SOCKET_PATH) as c:
opened = c._call("browser.open_split", {"url": "about:blank"}) or {}
target = str(opened.get("surface_id") or opened.get("surface_ref") or "")
_must(target != "", f"browser.open_split returned no surface handle: {opened}")
html = """
<!doctype html>
<html>
<head><title>cmux-browser-cli-regression</title></head>
<body>
<main>
<h1>browser cli regression</h1>
<p id="status">ready</p>
</main>
</body>
</html>
""".strip()
data_url = "data:text/html;charset=utf-8," + urllib.parse.quote(html)
c._call("browser.navigate", {"surface_id": target, "url": data_url})
wait_proc = _run_cli(
cli,
"browser",
target,
"wait",
"--load-state",
"interactive",
"--timeout-ms",
"5000",
)
_must(wait_proc.stdout.strip() == "OK", f"Expected browser wait OK output: {wait_proc.stdout!r}")
snapshot_payload = c._call("browser.snapshot", {"surface_id": target}) or {}
refs = snapshot_payload.get("refs") or {}
_must(isinstance(refs, dict) and len(refs) > 0, f"Expected snapshot refs for ref-based wait coverage: {snapshot_payload}")
ref_selector = str(next(iter(refs.keys())))
ref_wait_proc = _run_cli(
cli,
"browser",
target,
"wait",
"--selector",
ref_selector,
"--timeout-ms",
"2000",
)
_must(ref_wait_proc.stdout.strip() == "OK", f"Expected browser wait to resolve snapshot refs: {ref_wait_proc.stdout!r}")
snapshot_proc = _run_cli(cli, "browser", target, "snapshot", "--compact")
_must(
snapshot_proc.stdout.strip().startswith("- document"),
f"Expected snapshot command to succeed with structured output: {snapshot_proc.stdout!r}",
)
screenshot_json_proc = _run_cli(cli, "browser", target, "screenshot", "--json")
screenshot_json_text = screenshot_json_proc.stdout.strip()
payload = json.loads(screenshot_json_text or "{}")
_must("\\/" not in screenshot_json_text, f"Expected screenshot JSON without escaped slashes: {screenshot_json_text!r}")
_must("png_base64" not in payload, f"Expected screenshot JSON to omit png_base64 when file location is available: {payload}")
screenshot_path = str(payload.get("path") or "")
screenshot_url = str(payload.get("url") or "")
_must(screenshot_path.startswith("/"), f"Expected screenshot path in JSON payload: {payload}")
_must(screenshot_url.startswith("file://"), f"Expected screenshot file URL in JSON payload: {payload}")
_must(Path(screenshot_path).is_file(), f"Expected screenshot file to exist: {payload}")
out_dir = Path(tempfile.mkdtemp(prefix="cmux-browser-screenshot-cli-")) / "nested" / "dir"
out_path = out_dir / "capture.png"
screenshot_out_proc = _run_cli(
cli,
"browser",
target,
"screenshot",
"--out",
str(out_path),
)
_must(screenshot_out_proc.stdout.strip() == f"OK {out_path}", f"Expected --out to print the requested path: {screenshot_out_proc.stdout!r}")
_must("file://" not in screenshot_out_proc.stdout, f"Expected --out to print a path, not a file URL: {screenshot_out_proc.stdout!r}")
_must(out_path.is_file(), f"Expected --out screenshot file to exist: {out_path}")
print("PASS: browser CLI wait/snapshot and screenshot output work end-to-end")
return 0
if __name__ == "__main__":
raise SystemExit(main())

12
tests_v2/test_focus_notification_dismiss.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
E2E: focusing a panel clears its notification and triggers a flash.
E2E: focusing a panel preserves its notification and triggers a flash.
Note: This uses the socket focus command (no assistive access needed).
"""
@ -74,8 +74,12 @@ def main() -> int:
client.send("x")
time.sleep(0.2)
if not wait_for_notification(client, surface_id, is_read=True, timeout=2.0):
print("FAIL: Notification did not become read after focus")
if wait_for_notification(client, surface_id, is_read=True, timeout=2.0):
print("FAIL: Notification became read after focus")
return 1
items = client.list_notifications()
if not any(item["surface_id"] == surface_id and not item["is_read"] for item in items):
print("FAIL: Notification did not remain present and unread after focus")
return 1
final_flash = client.flash_count(term_b)
@ -93,7 +97,7 @@ def main() -> int:
except Exception:
pass
print("PASS: Focus clears notification and flashes panel")
print("PASS: Focus preserves notification and flashes panel")
return 0
except (cmuxError, RuntimeError) as exc:
print(f"FAIL: {exc}")

95
tests_v2/test_notifications.py
View file

@ -58,6 +58,15 @@ def wait_for_flash_count(client: cmux, surface: str, minimum: int = 1, timeout:
return last
def wait_for_current_workspace(client: cmux, expected: str, timeout: float = 2.0) -> bool:
start = time.time()
while time.time() - start < timeout:
if client.current_workspace() == expected:
return True
time.sleep(0.05)
return client.current_workspace() == expected
def ensure_two_surfaces(client: cmux) -> list[tuple[int, str, bool]]:
surfaces = client.list_surfaces()
if len(surfaces) < 2:
@ -215,8 +224,8 @@ def test_rxvt_notification_osc777(client: cmux) -> TestResult:
return result
def test_mark_read_on_focus_change(client: cmux) -> TestResult:
result = TestResult("Mark Read On Panel Focus")
def test_preserve_unread_on_focus_change(client: cmux) -> TestResult:
result = TestResult("Preserve Unread On Panel Focus")
try:
client.clear_notifications()
client.reset_flash_counts()
@ -229,81 +238,88 @@ def test_mark_read_on_focus_change(client: cmux) -> TestResult:
client.set_app_focus(False)
client.notify_surface(other[0], "focusread")
time.sleep(0.1)
items = wait_for_notifications(client, 1)
target = next((n for n in items if n["surface_id"] == other[1]), None)
if target is None or target["is_read"]:
result.failure("Expected unread notification for target surface before focus")
return result
client.set_app_focus(True)
client.focus_surface(other[0])
time.sleep(0.1)
count = wait_for_flash_count(client, other[1], minimum=1, timeout=2.0)
if count < 1:
result.failure("Expected flash on panel focus")
return result
items = client.list_notifications()
target = next((n for n in items if n["surface_id"] == other[1]), None)
if target is None:
result.failure("Expected notification for target surface")
elif not target["is_read"]:
result.failure("Expected notification to be marked read on focus")
elif target["is_read"]:
result.failure("Expected notification to remain unread on focus")
else:
count = wait_for_flash_count(client, other[1], minimum=1, timeout=2.0)
if count < 1:
result.failure("Expected flash on panel focus dismissal")
else:
result.success("Notification marked read on focus")
result.success("Notification persisted across panel focus")
except Exception as e:
result.failure(f"Exception: {e}")
return result
def test_mark_read_on_app_active(client: cmux) -> TestResult:
result = TestResult("Mark Read On App Active")
def test_preserve_unread_on_app_active(client: cmux) -> TestResult:
result = TestResult("Preserve Unread On App Active")
try:
client.clear_notifications()
client.set_app_focus(False)
client.notify("activate")
time.sleep(0.1)
items = client.list_notifications()
items = wait_for_notifications(client, 1)
if not items or items[0]["is_read"]:
result.failure("Expected unread notification before activation")
return result
client.simulate_app_active()
time.sleep(0.1)
items = client.list_notifications()
items = wait_for_notifications(client, 1)
if not items:
result.failure("Expected notification to remain after activation")
elif not items[0]["is_read"]:
result.failure("Expected notification to be marked read on app active")
elif items[0]["is_read"]:
result.failure("Expected notification to remain unread on app active")
else:
result.success("Notification marked read on app active")
result.success("Notification persisted across app activation")
except Exception as e:
result.failure(f"Exception: {e}")
return result
def test_mark_read_on_tab_switch(client: cmux) -> TestResult:
result = TestResult("Mark Read On Tab Switch")
def test_preserve_unread_on_tab_switch(client: cmux) -> TestResult:
result = TestResult("Preserve Unread On Tab Switch")
try:
client.clear_notifications()
client.set_app_focus(False)
tab1 = client.current_workspace()
client.notify("tabswitch")
time.sleep(0.1)
items = wait_for_notifications(client, 1)
target = next((n for n in items if n["workspace_id"] == tab1), None)
if target is None or target["is_read"]:
result.failure("Expected unread notification for original tab before switching")
return result
tab2 = client.new_workspace()
time.sleep(0.1)
if not wait_for_current_workspace(client, tab2):
result.failure("Expected new workspace to become selected")
return result
client.set_app_focus(True)
client.select_workspace(tab1)
time.sleep(0.1)
if not wait_for_current_workspace(client, tab1):
result.failure("Expected original workspace to become selected again")
return result
items = client.list_notifications()
items = wait_for_notifications(client, 1)
target = next((n for n in items if n["workspace_id"] == tab1), None)
if target is None:
result.failure("Expected notification for original tab")
elif not target["is_read"]:
result.failure("Expected notification to be marked read on tab switch")
elif target["is_read"]:
result.failure("Expected notification to remain unread on tab switch")
else:
result.success("Notification marked read on tab switch")
result.success("Notification persisted across tab switch")
except Exception as e:
result.failure(f"Exception: {e}")
return result
@ -371,11 +387,20 @@ def test_focus_on_notification_click(client: cmux) -> TestResult:
result.failure("Expected notification surface to be focused")
return result
items = client.list_notifications()
notification = next((n for n in items if n["surface_id"] == other[1]), None)
if notification is None:
result.failure("Expected notification to remain listed after notification click")
return result
if notification["is_read"]:
result.failure("Expected notification click to preserve unread state")
return result
count = wait_for_flash_count(client, other[1], minimum=1, timeout=2.0)
if count < 1:
result.failure(f"Expected flash count >= 1, got {count}")
else:
result.success("Notification click focuses and flashes panel")
result.success("Notification click focuses, flashes, and preserves unread state")
except Exception as e:
result.failure(f"Exception: {e}")
return result
@ -455,9 +480,9 @@ def run_tests() -> int:
results.append(test_kitty_notification_simple(client))
results.append(test_kitty_notification_chunked(client))
results.append(test_rxvt_notification_osc777(client))
results.append(test_mark_read_on_focus_change(client))
results.append(test_mark_read_on_app_active(client))
results.append(test_mark_read_on_tab_switch(client))
results.append(test_preserve_unread_on_focus_change(client))
results.append(test_preserve_unread_on_app_active(client))
results.append(test_preserve_unread_on_tab_switch(client))
results.append(test_flash_on_tab_switch(client))
results.append(test_focus_on_notification_click(client))
results.append(test_restore_focus_on_tab_switch(client))