134 lines
4.2 KiB
Python
134 lines
4.2 KiB
Python
"""Manually start sloppy.py, then run the test against the same GUI process."""
|
|
import subprocess
|
|
import os
|
|
import sys
|
|
import time
|
|
import socket
|
|
from pathlib import Path
|
|
|
|
# Start sloppy.py
|
|
project_root = Path("C:/projects/manual_slop").absolute()
|
|
gui_script = project_root / "sloppy.py"
|
|
test_workspace = project_root / "tests" / "artifacts" / "live_gui_workspace"
|
|
|
|
# Clean up old workspace
|
|
if test_workspace.exists():
|
|
import shutil
|
|
for _ in range(5):
|
|
try:
|
|
shutil.rmtree(test_workspace)
|
|
break
|
|
except PermissionError:
|
|
time.sleep(0.5)
|
|
|
|
test_workspace.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create minimal files
|
|
(test_workspace / "manual_slop.toml").write_text("[project]\nname = 'TestProject'\n\n[conductor]\ndir = 'conductor'\n", encoding="utf-8")
|
|
(test_workspace / "conductor" / "tracks").mkdir(parents=True, exist_ok=True)
|
|
|
|
config_content = {
|
|
'ai': {'provider': 'gemini', 'model': 'gemini-2.5-flash-lite'},
|
|
'projects': {
|
|
'paths': [str((test_workspace / 'manual_slop.toml').absolute())],
|
|
'active': str((test_workspace / 'manual_slop.toml').absolute())
|
|
},
|
|
'paths': {
|
|
'logs_dir': str((test_workspace / "logs").absolute()),
|
|
'scripts_dir': str((test_workspace / "scripts" / "generated").absolute())
|
|
},
|
|
}
|
|
import tomli_w
|
|
with open(test_workspace / 'config.toml', 'wb') as f:
|
|
tomli_w.dump(config_content, f)
|
|
|
|
# Start sloppy.py
|
|
os.makedirs("logs", exist_ok=True)
|
|
log_file = open("logs/sloppy_py_test_2.log", "w", encoding="utf-8")
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = str(project_root.absolute())
|
|
env["SLOP_CONFIG"] = str((test_workspace / "config.toml").absolute())
|
|
env["SLOP_GLOBAL_PRESETS"] = str((test_workspace / "presets.toml").absolute())
|
|
env["SLOP_GLOBAL_TOOL_PRESETS"] = str((test_workspace / "tool_presets.toml").absolute())
|
|
|
|
print("Starting sloppy.py...")
|
|
proc = subprocess.Popen(
|
|
["uv", "run", "python", "-u", str(gui_script), "--enable-test-hooks"],
|
|
stdout=log_file,
|
|
stderr=log_file,
|
|
text=True,
|
|
cwd=str(test_workspace.absolute()),
|
|
env=env,
|
|
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == 'nt' else 0
|
|
)
|
|
print(f"Started PID: {proc.pid}")
|
|
|
|
# Wait for hook server
|
|
import requests
|
|
for i in range(30):
|
|
try:
|
|
resp = requests.get("http://127.0.0.1:8999/status", timeout=0.5)
|
|
if resp.status_code == 200:
|
|
print(f"Hook server ready after {i*0.5}s")
|
|
break
|
|
except Exception:
|
|
time.sleep(0.5)
|
|
else:
|
|
print("Hook server didn't start!")
|
|
proc.kill()
|
|
sys.exit(1)
|
|
|
|
# Wait extra for imgui to fully initialize
|
|
print("Waiting 3s for imgui to stabilize...")
|
|
time.sleep(3.0)
|
|
|
|
# Now run the actual test flow
|
|
from src.api_hook_client import ApiHookClient
|
|
client = ApiHookClient()
|
|
|
|
print("\n[1] set_value show_windows {Diagnostics: True}")
|
|
client.set_value('show_windows', {'Diagnostics': True})
|
|
time.sleep(1.0)
|
|
|
|
print("\n[2] push_event save_workspace_profile")
|
|
client.push_event('custom_callback', {'callback': 'save_workspace_profile', 'args': ['Tier3Profile', 'project']})
|
|
time.sleep(1.0)
|
|
|
|
print("\n[3] set_value show_windows {Diagnostics: False}")
|
|
client.set_value('show_windows', {'Diagnostics': False})
|
|
|
|
print("\n[4] set_value ui_auto_switch_layout")
|
|
client.set_value('ui_auto_switch_layout', True)
|
|
|
|
print("\n[5] set_value ui_tier_layout_bindings")
|
|
client.set_value('ui_tier_layout_bindings', {'Tier 1': '', 'Tier 2': '', 'Tier 3': 'Tier3Profile', 'Tier 4': ''})
|
|
|
|
def trigger_tier(tier):
|
|
client.push_event("mma_state_update", {"status": "running", "active_tier": tier})
|
|
|
|
print("\n[6] trigger Tier 2")
|
|
trigger_tier('Tier 2 (Tech Lead)')
|
|
time.sleep(1.0)
|
|
val = client.get_value('show_windows')
|
|
print(f"[after Tier 2] show_windows: {val!r}")
|
|
assert val is not None, "show_windows is None"
|
|
assert val.get('Diagnostics', False) == False, f"Expected False, got {val}"
|
|
|
|
print("\n[7] trigger Tier 3")
|
|
trigger_tier('Tier 3 (Worker): task-1')
|
|
time.sleep(1.0)
|
|
val = client.get_value('show_windows')
|
|
print(f"[after Tier 3] show_windows: {val!r}")
|
|
assert val.get('Diagnostics', False) == True, f"Expected True, got {val}"
|
|
|
|
print("\nALL ASSERTIONS PASSED!")
|
|
|
|
# Cleanup
|
|
print("Killing sloppy.py...")
|
|
proc.kill()
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except:
|
|
pass
|
|
log_file.close()
|