alphard/ComfyUI-Lora-Manager
1
0
Fork 0
mirror of https://github.com/willmiao/ComfyUI-Lora-Manager.git synced 2026-08-09 23:40:15 -03:00
Code Issues Packages Projects Releases Wiki Activity

chore(skill): harden lora-manager-e2e for sandboxed E2E

Browse Source
This commit is contained in:
Will Miao
2026-08-09 11:31:04 +08:00
parent 420530f532
commit d0bc4be0dc
6 changed files with 576 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
.agents/skills/lora-manager-e2e/scripts/example_e2e_test.py
View File

@@ -8,186 +8,208 @@ This script shows how to:
3. Verify functionality end-to-end
Note: This is a template. Actual execution requires Chrome DevTools MCP.
Port: pick a FREE port for the run — 8188 is commonly occupied by a live
ComfyUI (see the skill's Port Selection section). Set PORT below to e.g. 8199
when 8188 is taken. Always run against a SANDBOXED standalone server.
"""
import subprocess
import sys
import time
# Choose the E2E port. 8188 is only the default candidate; use 8199 (or any
# free port checked with `ss -tlnp`) when 8188 is occupied by a live ComfyUI.
PORT = "8188"
def run_test():
"""Run example E2E test flow."""
print("=" * 60)
print("LoRa Manager E2E Test Example")
print("=" * 60)
# Step 1: Start server
# Step 1: Start server (detached so it survives the shell)
print("\n[1/5] Starting LoRa Manager standalone server...")
result = subprocess.run(
[sys.executable, "start_server.py", "--port", "8188", "--wait", "--timeout", "30"],
[sys.executable, "start_server.py", "--port", PORT, "--wait", "--timeout", "30", "--detach"],
capture_output=True,
text=True
text=True,
)
if result.returncode != 0:
print(f"Failed to start server: {result.stderr}")
return 1
print("Server ready!")
# Step 2: Open Chrome (manual step - show command)
print("\n[2/5] Open Chrome with debug mode:")
print("google-chrome --remote-debugging-port=9222 --user-data-dir=/tmp/chrome-lora-manager http://127.0.0.1:8188/loras")
print(
f"google-chrome --remote-debugging-port=9222 "
f"--user-data-dir=/tmp/chrome-lora-manager http://127.0.0.1:{PORT}/loras"
)
print("(In actual test, this would be automated via MCP)")
# Step 3: Navigate and verify page load
print("\n[3/5] Page Load Verification:")
print("""
print(
f"""
MCP Commands to execute:
1. navigate_page(type="url", url="http://127.0.0.1:8188/loras")
1. navigate_page(type="url", url="http://127.0.0.1:{PORT}/loras")
2. wait_for(text="LoRAs", timeout=10000)
3. snapshot = take_snapshot()
""")
"""
)
# Step 4: Test search functionality
print("\n[4/5] Search Functionality Test:")
print("""
print(
"""
MCP Commands to execute:
1. fill(uid="search-input", value="test")
2. press_key(key="Enter")
3. wait_for(text="Results", timeout=5000)
4. result = evaluate_script(function="""
4. result = evaluate_script(function=`
() => {
const cards = document.querySelectorAll('.lora-card');
return { count: cards.length };
}
""")
""")
`)
"""
)
# Step 5: Verify API
print("\n[5/5] API Verification:")
print("""
print(
"""
MCP Commands to execute:
1. api_result = evaluate_script(function="""
1. api_result = evaluate_script(function=`
async () => {
const response = await fetch('/loras/api/list');
const data = await response.json();
return { count: data.length, status: response.status };
}
""")
`)
2. Verify api_result['status'] == 200
""")
"""
)
print("\n" + "=" * 60)
print("Test flow completed!")
print("=" * 60)
return 0
def example_restart_flow():
"""Example: Testing configuration change that requires restart."""
print("\n" + "=" * 60)
print("Example: Server Restart Flow")
print("=" * 60)
print("""
print(
f"""
Scenario: Change setting and verify after restart
Steps:
1. Navigate to settings page
- navigate_page(type="url", url="http://127.0.0.1:8188/settings")
- navigate_page(type="url", url="http://127.0.0.1:{PORT}/settings")
2. Change a setting (e.g., theme)
- fill(uid="theme-select", value="dark")
- click(uid="save-settings-button")
3. Restart server
- subprocess.run([python, "start_server.py", "--restart", "--wait"])
- subprocess.run([python, "start_server.py", "--port", "{PORT}", "--restart", "--wait", "--detach"])
4. Refresh browser
- navigate_page(type="reload", ignoreCache=True)
- wait_for(text="LoRAs", timeout=15000)
5. Verify setting persisted
- navigate_page(type="url", url="http://127.0.0.1:8188/settings")
- navigate_page(type="url", url="http://127.0.0.1:{PORT}/settings")
- theme = evaluate_script(function="() => document.querySelector('#theme-select').value")
- assert theme == "dark"
""")
"""
)
def example_modal_interaction():
"""Example: Testing modal dialog interaction."""
print("\n" + "=" * 60)
print("Example: Modal Dialog Interaction")
print("=" * 60)
print("""
print(
"""
Scenario: Add new LoRA via modal
Steps:
1. Open modal
- click(uid="add-lora-button")
- wait_for(text="Add LoRA", timeout=3000)
2. Fill form
- fill_form(elements=[
{"uid": "lora-name", "value": "Test Character"},
{"uid": "lora-path", "value": "/models/test.safetensors"},
])
3. Submit
- click(uid="modal-submit-button")
4. Verify success
- wait_for(text="Successfully added", timeout=5000)
- snapshot = take_snapshot()
""")
"""
)
def example_network_monitoring():
"""Example: Network request monitoring."""
print("\n" + "=" * 60)
print("Example: Network Request Monitoring")
print("=" * 60)
print("""
print(
f"""
Scenario: Verify API calls during user interaction
Steps:
1. Clear network log (implicit on navigation)
- navigate_page(type="url", url="http://127.0.0.1:8188/loras")
- navigate_page(type="url", url="http://127.0.0.1:{PORT}/loras")
2. Perform action that triggers API call
- fill(uid="search-input", value="character")
- press_key(key="Enter")
3. List network requests
- requests = list_network_requests(resourceTypes=["xhr", "fetch"])
4. Find search API call
- search_requests = [r for r in requests if "/api/search" in r.get("url", "")]
- assert len(search_requests) > 0, "Search API was not called"
5. Get request details
- if search_requests:
details = get_network_request(reqid=search_requests[0]["reqid"])
- Verify request method, response status, etc.
""")
"""
)
if __name__ == "__main__":
print("LoRa Manager E2E Test Examples\n")
print("This script demonstrates E2E testing patterns.\n")
print("Note: Actual execution requires Chrome DevTools MCP connection.\n")
run_test()
example_restart_flow()
example_modal_interaction()
example_network_monitoring()
print("\n" + "=" * 60)
print("All examples shown!")
print("=" * 60)

271
.agents/skills/lora-manager-e2e/scripts/start_server.py
View File

@@ -1,15 +1,78 @@
#!/usr/bin/env python3
"""
Start or restart LoRa Manager standalone server for E2E testing.
Backward-compatible CLI: --port, --restart, --wait, --timeout all work as before.
New options: --detach (setsid-style fully detached launch, survives shell death).
Safety rules implemented here:
- Never kill processes the script did not start. The script tracks the PIDs it
manages in a pidfile (/tmp/lora-manager-e2e-server-{PORT}.pid).
- If the port is held by an unrelated process (e.g. a live ComfyUI) the script
reports the conflict and exits early instead of killing it.
- --restart only kills managed PIDs; if unrelated processes still hold the port
afterwards, the script reports them and aborts.
"""
from __future__ import annotations
import argparse
import os
import signal
import socket
import subprocess
import sys
import time
import socket
import signal
import os
PIDFILE_PREFIX = "/tmp/lora-manager-e2e-server"
def pidfile_path(port: int) -> str:
"""Path of the pidfile that records PIDs this script started for a port."""
return f"{PIDFILE_PREFIX}-{port}.pid"
def read_managed_pids(port: int) -> list[int]:
"""Read PIDs this script previously managed for the port (may be stale)."""
path = pidfile_path(port)
if not os.path.exists(path):
return []
try:
with open(path, "r", encoding="utf-8") as fh:
return [int(line.strip()) for line in fh if line.strip().isdigit()]
except (OSError, ValueError):
return []
def write_managed_pids(port: int, pids: list[int]) -> None:
"""Record PIDs this script manages for the port."""
try:
with open(pidfile_path(port), "w", encoding="utf-8") as fh:
for pid in pids:
fh.write(f"{pid}\n")
except OSError as exc:
print(f"Warning: could not write pidfile for port {port}: {exc}")
def clear_managed_pids(port: int) -> None:
"""Remove the pidfile for the port (no longer managed)."""
path = pidfile_path(port)
try:
if os.path.exists(path):
os.remove(path)
except OSError as exc:
print(f"Warning: could not remove pidfile {path}: {exc}")
def process_alive(pid: int) -> bool:
"""Return True if a process with the given pid exists."""
try:
os.kill(pid, 0)
return True
except ProcessLookupError:
return False
except PermissionError:
return True # exists but owned by someone else
def find_server_process(port: int) -> list[int]:
@@ -19,7 +82,7 @@ def find_server_process(port: int) -> list[int]:
["lsof", "-ti", f":{port}"],
capture_output=True,
text=True,
check=False
check=False,
)
if result.returncode == 0 and result.stdout.strip():
return [int(pid) for pid in result.stdout.strip().split("\n") if pid]
@@ -30,7 +93,7 @@ def find_server_process(port: int) -> list[int]:
["netstat", "-tlnp"],
capture_output=True,
text=True,
check=False
check=False,
)
pids = []
for line in result.stdout.split("\n"):
@@ -49,30 +112,48 @@ def find_server_process(port: int) -> list[int]:
return []
def kill_server(port: int) -> None:
"""Kill processes using the specified port."""
pids = find_server_process(port)
def describe_processes(pids: list[int]) -> str:
"""Human-readable description of a pid list (pid + command line)."""
descriptions = []
for pid in pids:
cmdline = ""
try:
with open(f"/proc/{pid}/cmdline", "rb") as fh:
raw = fh.read().replace(b"\x00", b" ").decode("utf-8", "replace")
cmdline = raw.strip()
except OSError:
pass
descriptions.append(f"pid {pid}{' (' + cmdline + ')' if cmdline else ''}")
return ", ".join(descriptions) if descriptions else "none"
def kill_pids(pids: list[int], what: str) -> None:
"""Send SIGTERM (then SIGKILL) to the given PIDs, only after reporting."""
for pid in pids:
print(f"Sent SIGTERM to {what} pid {pid}")
try:
os.kill(pid, signal.SIGTERM)
print(f"Sent SIGTERM to process {pid}")
except ProcessLookupError:
pass
# Wait for processes to terminate
time.sleep(1)
deadline = time.time() + 5
while time.time() < deadline:
if not any(process_alive(pid) for pid in pids):
break
time.sleep(0.2)
# Force kill if still running
pids = find_server_process(port)
for pid in pids:
try:
os.kill(pid, signal.SIGKILL)
print(f"Sent SIGKILL to process {pid}")
except ProcessLookupError:
pass
if process_alive(pid):
try:
os.kill(pid, signal.SIGKILL)
print(f"Sent SIGKILL to {what} pid {pid}")
except ProcessLookupError:
pass
def is_server_ready(port: int, timeout: float = 0.5) -> bool:
def is_server_ready(port: int, timeout: float = 2.0) -> bool:
"""Check if server is accepting connections."""
try:
with socket.create_connection(("127.0.0.1", port), timeout=timeout):
@@ -84,9 +165,15 @@ def is_server_ready(port: int, timeout: float = 0.5) -> bool:
def wait_for_server(port: int, timeout: int = 30) -> bool:
"""Wait for server to become ready."""
start = time.time()
last_report = 0.0
while time.time() - start < timeout:
if is_server_ready(port):
return True
# Report progress every ~5s so a slow boot is visible, not silent.
elapsed = time.time() - start
if elapsed - last_report >= 5:
print(f" ...still waiting ({int(elapsed)}s/{timeout}s)")
last_report = elapsed
time.sleep(0.5)
return False
@@ -99,68 +186,148 @@ def main() -> int:
"--port",
type=int,
default=8188,
help="Server port (default: 8188)"
help="Server port (default: 8188)",
)
parser.add_argument(
"--restart",
action="store_true",
help="Kill existing server before starting"
help="Kill the E2E server previously managed by this script for the port "
"(tracked via pidfile) before starting; refuse to kill unrelated processes",
)
parser.add_argument(
"--wait",
action="store_true",
help="Wait for server to be ready before exiting"
help="Wait for server to be ready before exiting",
)
parser.add_argument(
"--timeout",
type=int,
default=30,
help="Timeout for waiting (default: 30)"
help="Timeout for waiting (default: 30)",
)
parser.add_argument(
"--detach",
action="store_true",
help="Launch the server fully detached (setsid-style) so it survives shell "
"death. REQUIRED for E2E: a plain background process dies with the shell",
)
args = parser.parse_args()
# Get project root (parent of .agents directory)
script_dir = os.path.dirname(os.path.abspath(__file__))
skill_dir = os.path.dirname(script_dir)
project_root = os.path.dirname(os.path.dirname(os.path.dirname(skill_dir)))
# Restart if requested
managed_pids = read_managed_pids(args.port)
# Restart if requested: kill ONLY managed PIDs.
if args.restart:
print(f"Killing existing server on port {args.port}...")
kill_server(args.port)
alive_managed = [pid for pid in managed_pids if process_alive(pid)]
if alive_managed:
print(
f"Killing E2E server previously started by this script on port "
f"{args.port} ({describe_processes(alive_managed)})..."
)
kill_pids(alive_managed, "managed E2E server")
else:
print(
f"No live managed E2E server for port {args.port} "
f"(pidfile: {pidfile_path(args.port)})"
)
time.sleep(1)
# Check if already running
if is_server_ready(args.port):
print(f"Server already running on port {args.port}")
return 0
# Refuse to kill anything the script did not manage.
remaining = find_server_process(args.port)
if remaining:
print(
f"ERROR: port {args.port} is still held by process(es) this script "
f"did not start: {describe_processes(remaining)}."
)
print(
"These may be unrelated (e.g. a live ComfyUI). The script will NOT "
"kill them. Pick a different --port, or stop them manually if you "
"are certain they are stale E2E servers."
)
return 2
clear_managed_pids(args.port)
# Port conflict check before starting: never blind-kill.
port_pids = find_server_process(args.port)
if port_pids:
alive_managed = [pid for pid in port_pids if pid in managed_pids]
unmanaged = [pid for pid in port_pids if pid not in managed_pids]
if alive_managed and not unmanaged:
print(
f"Server already running on port {args.port} "
f"({describe_processes(alive_managed)}, started by this script). "
f"Use --restart to recycle it."
)
return 0
print(
f"ERROR: port {args.port} is already in use by process(es): "
f"{describe_processes(port_pids)}."
)
print(
"This is likely an unrelated process (e.g. a live ComfyUI holding 8188). "
"The script will NOT kill it. Pick a free port with --port, e.g. 8199."
)
return 2
# Start server
print(f"Starting LoRa Manager standalone server on port {args.port}...")
cmd = [sys.executable, "standalone.py", "--port", str(args.port)]
# Start in background
process = subprocess.Popen(
cmd,
cwd=project_root,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
start_new_session=True
)
print(f"Server process started with PID {process.pid}")
cmd = [
sys.executable,
"standalone.py",
"--host",
"127.0.0.1",
"--port",
str(args.port),
]
if args.detach:
# Fully detached launch: new session (setsid), no controlling terminal,
# stdin from /dev/null, stdout/stderr to a log file. Survives the shell.
log_dir = os.path.join(script_dir, "logs")
os.makedirs(log_dir, exist_ok=True)
log_path = os.path.join(log_dir, f"server-{args.port}.log")
with open(log_path, "ab") as log_fh:
process = subprocess.Popen(
cmd,
cwd=project_root,
stdin=subprocess.DEVNULL,
stdout=log_fh,
stderr=subprocess.STDOUT,
start_new_session=True,
close_fds=True,
)
print(f"Detached server process started with PID {process.pid} (setsid)")
print(f"Log: {log_path}")
else:
# Plain background process (legacy behavior): dies with the shell.
process = subprocess.Popen(
cmd,
cwd=project_root,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
start_new_session=True,
)
print(f"Server process started with PID {process.pid}")
print(
"NOTE: not detached — this process dies when the launching shell exits. "
"For E2E use --detach."
)
write_managed_pids(args.port, [process.pid])
# Wait for ready if requested
if args.wait:
print(f"Waiting for server to be ready (timeout: {args.timeout}s)...")
if wait_for_server(args.port, args.timeout):
print(f"Server ready at http://127.0.0.1:{args.port}/loras")
return 0
else:
print(f"Timeout waiting for server")
return 1
print(f"Timeout waiting for server on port {args.port}")
return 1
print(f"Server starting at http://127.0.0.1:{args.port}/loras")
return 0

28
.agents/skills/lora-manager-e2e/scripts/wait_for_server.py
View File

@@ -1,15 +1,20 @@
#!/usr/bin/env python3
"""
Wait for LoRa Manager server to become ready.
Timeout is configurable via --timeout (default 30s); the script polls the port
until the server accepts connections or the timeout expires.
"""
from __future__ import annotations
import argparse
import socket
import sys
import time
def is_server_ready(port: int, timeout: float = 0.5) -> bool:
def is_server_ready(port: int, timeout: float = 2.0) -> bool:
"""Check if server is accepting connections."""
try:
with socket.create_connection(("127.0.0.1", port), timeout=timeout):
@@ -21,9 +26,15 @@ def is_server_ready(port: int, timeout: float = 0.5) -> bool:
def wait_for_server(port: int, timeout: int = 30) -> bool:
"""Wait for server to become ready."""
start = time.time()
last_report = 0.0
while time.time() - start < timeout:
if is_server_ready(port):
return True
# Report progress every ~5s so a slow boot is visible, not silent.
elapsed = time.time() - start
if elapsed - last_report >= 5:
print(f" ...still waiting ({int(elapsed)}s/{timeout}s)")
last_report = elapsed
time.sleep(0.5)
return False
@@ -36,25 +47,24 @@ def main() -> int:
"--port",
type=int,
default=8188,
help="Server port (default: 8188)"
help="Server port (default: 8188)",
)
parser.add_argument(
"--timeout",
type=int,
default=30,
help="Timeout in seconds (default: 30)"
help="Timeout in seconds (default: 30)",
)
args = parser.parse_args()
print(f"Waiting for server on port {args.port} (timeout: {args.timeout}s)...")
if wait_for_server(args.port, args.timeout):
print(f"Server ready at http://127.0.0.1:{args.port}/loras")
return 0
else:
print(f"Timeout: Server not ready after {args.timeout}s")
return 1
print(f"Timeout: Server not ready after {args.timeout}s")
return 1
if __name__ == "__main__":