fix: delete command auto-confirms prompts + add request-level logging

Bug 1: amplifier-dev --destroy prompts 'Are you sure? [y/N]'. subprocess.run
had no stdin input, so the prompt hung until timeout. Fix: input='y\n' auto-
confirms interactive prompts in delete_session().

Bug 2: No request-level logging — only startup messages visible in journalctl.
Added INFO-level log lines for session create, delete (command + success/fail),
and connect operations. Set uvicorn log_level='info' so application logs
appear in service logs (journalctl / launchd).

Tests: 5 new tests covering stdin confirmation kwarg, INFO-level logs for
delete/create/connect, and uvicorn log level assertion.
This commit is contained in:
Brian Krabach
2026-04-01 11:18:57 -07:00
parent 66f58a6f4b
commit 98d552557a
3 changed files with 138 additions and 2 deletions
+1 -1
View File
@@ -178,7 +178,7 @@ def serve(
from muxplex.main import app # noqa: PLC0415
print(f" muxplex → http://{host}:{port}")
uvicorn.run(app, host=host, port=port, log_level="warning")
uvicorn.run(app, host=host, port=port, log_level="info")
def doctor() -> None:
+7 -1
View File
@@ -368,6 +368,7 @@ async def create_session(payload: CreateSessionPayload) -> dict:
name = payload.name
settings = load_settings()
command = settings["new_session_template"].replace("{name}", name)
_log.info("Creating session '%s' with command: %s", name, command)
try:
subprocess.Popen(
command,
@@ -394,6 +395,7 @@ async def connect_session(name: str) -> dict:
if known and name not in known:
raise HTTPException(status_code=404, detail=f"Session '{name}' not found")
_log.info("Connecting to session '%s'", name)
await kill_ttyd()
await spawn_ttyd(name)
@@ -444,15 +446,19 @@ async def delete_session(name: str) -> dict:
"delete_session_template", "tmux kill-session -t {name}"
).replace("{name}", name)
_log.info("Deleting session '%s' with command: %s", name, command)
try:
result = subprocess.run(
command,
shell=True,
input="y\n", # auto-confirm interactive prompts (e.g. amplifier-dev --destroy)
capture_output=True,
text=True,
timeout=30,
)
if result.returncode != 0:
if result.returncode == 0:
_log.info("Session '%s' deleted successfully", name)
else:
_log.warning(
"Delete command failed (rc=%d): %s",
result.returncode,
+130
View File
@@ -1387,3 +1387,133 @@ def test_get_auth_token_returns_401_when_not_authenticated(monkeypatch):
# No cookie set — endpoint must return 401 with application/json accept
response = c.get("/api/auth/token", headers={"Accept": "application/json"})
assert response.status_code == 401
# ---------------------------------------------------------------------------
# Bug fix: delete_session must pass input="y\n" to subprocess.run
# ---------------------------------------------------------------------------
def test_delete_session_passes_stdin_y_to_subprocess(client, monkeypatch, tmp_path):
"""DELETE /api/sessions/{name} must pass input='y\\n' to subprocess.run.
When delete_session_template uses an interactive command (e.g. amplifier-dev
--destroy), the confirmation prompt must be auto-answered via stdin.
Without input='y\\n', subprocess.run hangs until 30s timeout and the
session is never actually deleted.
"""
from unittest.mock import MagicMock, patch
import muxplex.settings as settings_mod
monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["my-session"])
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "no-settings.json")
captured_kwargs = []
def mock_run(cmd, **kwargs):
captured_kwargs.append(kwargs)
result = MagicMock()
result.returncode = 0
result.stderr = ""
return result
with patch("muxplex.main.subprocess.run", side_effect=mock_run):
response = client.delete("/api/sessions/my-session")
assert response.status_code == 200
assert len(captured_kwargs) == 1, "subprocess.run must be called once"
kwargs = captured_kwargs[0]
assert "input" in kwargs, (
"subprocess.run must receive input= kwarg to auto-answer interactive prompts"
)
assert kwargs["input"] == "y\n", (
f"input must be 'y\\n' to confirm deletion, got: {kwargs['input']!r}"
)
# ---------------------------------------------------------------------------
# Bug fix: request-level INFO logging for session operations
# ---------------------------------------------------------------------------
def test_delete_session_logs_command_at_info(client, monkeypatch, tmp_path, caplog):
"""DELETE /api/sessions/{name} must log the command being run at INFO level."""
import logging
from unittest.mock import MagicMock, patch
import muxplex.settings as settings_mod
monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["logged-session"])
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "no-settings.json")
def mock_run(cmd, **kwargs):
result = MagicMock()
result.returncode = 0
result.stderr = ""
return result
with caplog.at_level(logging.INFO, logger="muxplex.main"):
with patch("muxplex.main.subprocess.run", side_effect=mock_run):
client.delete("/api/sessions/logged-session")
log_messages = "\n".join(caplog.messages)
assert "logged-session" in log_messages, (
f"delete_session must log the session name at INFO level, got logs:\n{log_messages}"
)
def test_create_session_logs_command(client, monkeypatch, tmp_path, caplog):
"""POST /api/sessions must log the command being launched at INFO level."""
import logging
from unittest.mock import MagicMock, patch
import muxplex.settings as settings_mod
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", tmp_path / "no-settings.json")
with caplog.at_level(logging.INFO, logger="muxplex.main"):
with patch("muxplex.main.subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock()
client.post("/api/sessions", json={"name": "new-session"})
log_messages = "\n".join(caplog.messages)
assert "new-session" in log_messages, (
f"create_session must log session name at INFO level, got logs:\n{log_messages}"
)
def test_connect_session_logs_session_name(client, monkeypatch, caplog):
"""POST /api/sessions/{name}/connect must log the session name at INFO level."""
import logging
monkeypatch.setattr("muxplex.main.get_session_list", lambda: ["target-session"])
async def mock_kill_ttyd():
pass
async def mock_spawn_ttyd(name):
pass
monkeypatch.setattr("muxplex.main.kill_ttyd", mock_kill_ttyd)
monkeypatch.setattr("muxplex.main.spawn_ttyd", mock_spawn_ttyd)
with caplog.at_level(logging.INFO, logger="muxplex.main"):
client.post("/api/sessions/target-session/connect")
log_messages = "\n".join(caplog.messages)
assert "target-session" in log_messages, (
f"connect_session must log session name at INFO level, got logs:\n{log_messages}"
)
def test_cli_uvicorn_log_level_is_info():
"""cli.py serve() must pass log_level='info' to uvicorn.run so logs appear in journalctl."""
import inspect
from muxplex import cli
source = inspect.getsource(cli.serve)
assert 'log_level="info"' in source or "log_level='info'" in source, (
"serve() must call uvicorn.run(..., log_level='info') so application "
"logs appear in journalctl; currently set to 'warning' which suppresses them"
)