129 lines
5.1 KiB
Python
129 lines
5.1 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import subprocess
|
|
import pytest
|
|
|
|
def test_bug2_headless_layout_does_not_overflow():
|
|
"""Verify Bug 2: w=0, h=0 in headless mode outputs 'right' (not 'overflow')."""
|
|
calc_script = """
|
|
import sys, json, os
|
|
min_cols = int(os.environ.get('MAM_MIN_COLS', 60))
|
|
min_rows = int(os.environ.get('MAM_MIN_ROWS', 20))
|
|
try:
|
|
d = json.loads(sys.stdin.read()).get('result', {})
|
|
focused_id = d.get('focused_pane_id', '')
|
|
panes = d.get('panes', [])
|
|
anchor = None
|
|
for p in panes:
|
|
if p.get('pane_id') == focused_id:
|
|
anchor = p.get('rect', {})
|
|
break
|
|
if not anchor and panes:
|
|
anchor = panes[0].get('rect', {})
|
|
if anchor:
|
|
w = anchor.get('width', 0)
|
|
h = anchor.get('height', 0)
|
|
if w <= 0 or h <= 0:
|
|
print('right')
|
|
elif w // 2 >= min_cols:
|
|
print('right')
|
|
elif h // 2 >= min_rows:
|
|
print('down')
|
|
else:
|
|
print('overflow')
|
|
except Exception:
|
|
pass
|
|
"""
|
|
# 1. Headless 0x0
|
|
payload_0x0 = json.dumps({"result": {"panes": [{"rect": {"width": 0, "height": 0}}]}})
|
|
res = subprocess.run([sys.executable, "-c", calc_script], input=payload_0x0, capture_output=True, text=True)
|
|
assert res.stdout.strip() == "right", f"Headless 0x0 should default to 'right', got {res.stdout.strip()}"
|
|
|
|
# 2. Genuine small pane (overflow)
|
|
payload_small = json.dumps({"result": {"panes": [{"rect": {"width": 50, "height": 30}}]}})
|
|
res = subprocess.run([sys.executable, "-c", calc_script], input=payload_small, capture_output=True, text=True)
|
|
assert res.stdout.strip() == "overflow", f"Small pane should be 'overflow', got {res.stdout.strip()}"
|
|
|
|
# 3. Wide pane (split right)
|
|
payload_wide = json.dumps({"result": {"panes": [{"rect": {"width": 160, "height": 30}}]}})
|
|
res = subprocess.run([sys.executable, "-c", calc_script], input=payload_wide, capture_output=True, text=True)
|
|
assert res.stdout.strip() == "right", f"Wide pane should be 'right', got {res.stdout.strip()}"
|
|
|
|
# 4. Tall pane (split down)
|
|
payload_tall = json.dumps({"result": {"panes": [{"rect": {"width": 80, "height": 60}}]}})
|
|
res = subprocess.run([sys.executable, "-c", calc_script], input=payload_tall, capture_output=True, text=True)
|
|
assert res.stdout.strip() == "down", f"Tall pane should be 'down', got {res.stdout.strip()}"
|
|
|
|
|
|
def test_bug3_reconcile_skills_dir_passed_and_fallback():
|
|
"""Verify Bug 3: reconcile.sh passes SKILLS_DIR to env_python/atomic_dump_yaml and RECON_SRC has fallback."""
|
|
recon_path = os.path.abspath(".agents/skills/multi-agent-mux-monitor/scripts/reconcile.sh")
|
|
with open(recon_path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
|
|
# Assert SKILLS_DIR is explicitly passed to env_python / atomic_dump_yaml
|
|
assert 'SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" env_python' in content
|
|
assert 'SKILLS_DIR="$SKILLS_DIR" LIB_SH="$LIB_SH" atomic_dump_yaml' in content
|
|
|
|
# Assert fallback exists inside RECON_SRC
|
|
assert "if not skills_dir:" in content
|
|
assert "skills_dir = os.path.join(_ws_root, '.agents/skills')" in content
|
|
|
|
|
|
def test_bug4_send_keys_safe_gating_order():
|
|
"""Verify Bug 4: send_keys_safe checks quiescence and dialogs before agent prompt fast-path."""
|
|
lib_path = os.path.abspath(".agents/skills/lib.sh")
|
|
with open(lib_path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
|
|
sks_idx = content.find("send_keys_safe() {")
|
|
assert sks_idx != -1
|
|
|
|
sks_body = content[sks_idx:sks_idx + 2500]
|
|
quiescent_idx = sks_body.find("_pane_quiescent")
|
|
dialog_idx = sks_body.find("_pane_dialog_open")
|
|
prompt_idx = sks_body.find("agent prompt")
|
|
|
|
assert quiescent_idx != -1, "_pane_quiescent not found in send_keys_safe"
|
|
assert dialog_idx != -1, "_pane_dialog_open not found in send_keys_safe"
|
|
assert prompt_idx != -1, "agent prompt not found in send_keys_safe"
|
|
|
|
# Crucial ordering check: quiescence and dialog checks MUST precede agent prompt
|
|
assert quiescent_idx < prompt_idx, "_pane_quiescent must execute before agent prompt fast-path"
|
|
assert dialog_idx < prompt_idx, "_pane_dialog_open must execute before agent prompt fast-path"
|
|
|
|
|
|
def test_bug4_no_duplicate_input_on_rpc_success(tmp_path):
|
|
"""Verify Bug 4: when herdr agent prompt succeeds, send_keys_safe returns 0 without calling paste-buffer."""
|
|
test_script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
SKILL_DIR="{os.path.abspath('.agents/skills')}"
|
|
source "$SKILL_DIR/lib.sh"
|
|
|
|
_pane_quiescent() {{ return 0; }}
|
|
_pane_dialog_open() {{ return 1; }}
|
|
|
|
PASTE_CALLED=0
|
|
_sks_herdr() {{
|
|
if [ "${{1:-}}" = "agent" ] && [ "${{2:-}}" = "prompt" ]; then
|
|
return 0
|
|
fi
|
|
if [ "${{1:-}}" = "paste-buffer" ]; then
|
|
PASTE_CALLED=1
|
|
fi
|
|
return 0
|
|
}}
|
|
|
|
send_keys_safe "test-sess" "my prompt" "job-1"
|
|
if [ "$PASTE_CALLED" = "1" ]; then
|
|
echo "ERROR: paste-buffer was called after agent prompt success (duplicate input)" >&2
|
|
exit 1
|
|
fi
|
|
echo "SUCCESS"
|
|
"""
|
|
res = subprocess.run(["bash", "-c", test_script], capture_output=True, text=True)
|
|
assert res.returncode == 0, f"Expected clean exit 0 without duplicate paste-buffer call, got {res.returncode}. Stderr: {res.stderr}"
|
|
assert "SUCCESS" in res.stdout
|
|
|