Zhengshou Lai
f14c39d7c0
- launcher _BACKENDS 加 dsh：profile 透传、zstd 会话列表（session/title 或首条用户消息）、 --resume 透传、DSH_HOME 环境变量与 expanduser - entrypoints/cli/completion/打包/卸载全触点 + TestDsh 用例（裸跑/版本/透传/列表/缺 bin）
922 lines
34 KiB
Python
922 lines
34 KiB
Python
"""Tests for myagents CLI."""
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
import myagents.launcher
|
|
from myagents.cli import cli
|
|
|
|
|
|
def _backend_which(*names: str):
|
|
"""Patch ``shutil.which`` to find backend binaries and nothing else.
|
|
|
|
The launcher also probes ``which("npm")`` while resolving npm-global
|
|
installs; returning None there keeps that probe from spawning a real
|
|
``subprocess.run`` in tests (which would break ``assert_called_once``).
|
|
"""
|
|
|
|
def _which(name: str) -> str | None:
|
|
return f"/usr/bin/{name}" if name in names else None
|
|
|
|
return _which
|
|
|
|
|
|
class TestMyagentsHelp:
|
|
"""Tests for top-level myagents command."""
|
|
|
|
def test_help_shows_agent_subcommands(self) -> None:
|
|
"""--help should list claude, kimi, codex, hermes, cursor and dsh."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "claude" in result.output
|
|
assert "kimi" in result.output
|
|
assert "codex" in result.output
|
|
assert "hermes" in result.output
|
|
assert "cursor" in result.output
|
|
assert "dsh" in result.output
|
|
assert "update" in result.output
|
|
assert "upgrade" in result.output
|
|
|
|
def test_version_shows_version(self) -> None:
|
|
"""--version should show package version."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--version"])
|
|
assert result.exit_code == 0
|
|
assert "version" in result.output.lower()
|
|
|
|
def test_bare_invocation_shows_help(self) -> None:
|
|
"""Running myagents without subcommands should show help, not launch agent."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, [])
|
|
assert result.exit_code == 0
|
|
assert "Usage:" in result.output
|
|
|
|
def test_backends_lists_all(self) -> None:
|
|
"""``myagents backends`` lists all supported agent backends."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["backends"])
|
|
assert result.exit_code == 0
|
|
assert "claude" in result.output
|
|
assert "kimi" in result.output
|
|
assert "codex" in result.output
|
|
assert "hermes" in result.output
|
|
assert "cursor" in result.output
|
|
|
|
def test_help_lists_backends(self) -> None:
|
|
"""--help should list backends subcommand."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "backends" in result.output
|
|
|
|
|
|
class TestClaudeSubcommand:
|
|
"""Tests for ``myagents claude``."""
|
|
|
|
def test_help_shows_options(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["claude", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "--cwd" in result.output
|
|
assert "--list" in result.output
|
|
assert "--dangerously-skip-permissions" not in result.output
|
|
|
|
def test_runs_claude(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["claude"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once()
|
|
assert mock_run.call_args[0][0] == [
|
|
"/usr/bin/claude",
|
|
"--dangerously-skip-permissions",
|
|
]
|
|
|
|
def test_missing_binary_error(self) -> None:
|
|
runner = CliRunner()
|
|
with patch("myagents.launcher.shutil.which", return_value=None):
|
|
result = runner.invoke(cli, ["claude"])
|
|
assert result.exit_code == 127
|
|
assert "not found" in result.output.lower()
|
|
|
|
def test_cwd_option_passed(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
test_dir = tmp_path / "test_cwd"
|
|
test_dir.mkdir()
|
|
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["claude", "--cwd", str(test_dir)])
|
|
assert result.exit_code == 0
|
|
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
|
|
|
|
def test_cwd_invalid_directory(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
bad_dir = tmp_path / "does_not_exist"
|
|
result = runner.invoke(cli, ["claude", "--cwd", str(bad_dir)])
|
|
assert result.exit_code == 1
|
|
assert "not a directory" in result.output.lower()
|
|
|
|
|
|
class TestClaudePassthrough:
|
|
"""Unknown leading flags forward to claude instead of erroring."""
|
|
|
|
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(
|
|
cli, ["claude", "--cwd", str(tmp_path), *args]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
mock_run.assert_called_once()
|
|
return mock_run.call_args[0][0]
|
|
|
|
def test_bare_resume_opens_picker_not_launch(self, tmp_path: Path) -> None:
|
|
"""Bare --resume opens the unified picker; non-tty lists and exits."""
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("claude")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["claude", "--cwd", str(tmp_path), "--resume"])
|
|
assert result.exit_code == 0, result.output
|
|
mock_run.assert_not_called()
|
|
|
|
def test_resume_with_session_id(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-r", "abc123"], tmp_path)
|
|
assert cmd[-2:] == ["-r", "abc123"]
|
|
|
|
def test_continue_flag_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--continue"], tmp_path)
|
|
assert cmd[-1] == "--continue"
|
|
|
|
|
|
class TestClaudeListSessions:
|
|
"""``myagents claude --list`` reports resumable sessions."""
|
|
|
|
def test_list_empty_directory(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
with patch("myagents.launcher.subprocess.run") as mock_run:
|
|
result = runner.invoke(cli, ["claude", "--cwd", str(tmp_path), "--list"])
|
|
assert result.exit_code == 0
|
|
assert "no sessions" in result.output.lower()
|
|
mock_run.assert_not_called()
|
|
|
|
|
|
class TestKimiSubcommand:
|
|
"""Tests for ``myagents kimi``."""
|
|
|
|
def test_help_shows_options(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["kimi", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "--cwd" in result.output
|
|
assert "--list" in result.output
|
|
assert "--dangerously-skip-permissions" not in result.output
|
|
|
|
def test_runs_kimi(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("kimi")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["kimi"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once()
|
|
assert mock_run.call_args[0][0] == ["/usr/bin/kimi"]
|
|
|
|
def test_missing_binary_error(self) -> None:
|
|
runner = CliRunner()
|
|
with patch("myagents.launcher.shutil.which", return_value=None):
|
|
result = runner.invoke(cli, ["kimi"])
|
|
assert result.exit_code == 127
|
|
assert "not found" in result.output.lower()
|
|
|
|
def test_cwd_option_passed(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
test_dir = tmp_path / "test_cwd"
|
|
test_dir.mkdir()
|
|
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("kimi")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["kimi", "--cwd", str(test_dir)])
|
|
assert result.exit_code == 0
|
|
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
|
|
|
|
|
|
class TestKimiPassthrough:
|
|
"""Unknown leading flags forward to kimi instead of erroring."""
|
|
|
|
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("kimi")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["kimi", "--cwd", str(tmp_path), *args])
|
|
assert result.exit_code == 0, result.output
|
|
mock_run.assert_called_once()
|
|
return mock_run.call_args[0][0]
|
|
|
|
def test_resume_flag_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--resume"], tmp_path)
|
|
assert cmd == ["/usr/bin/kimi", "--resume"]
|
|
|
|
def test_resume_with_session_id(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-r", "abc123"], tmp_path)
|
|
assert cmd[-2:] == ["-r", "abc123"]
|
|
|
|
def test_continue_flag_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--continue"], tmp_path)
|
|
assert cmd[-1] == "--continue"
|
|
|
|
|
|
class TestUpdateSubcommand:
|
|
"""Tests for update/upgrade subcommands registration."""
|
|
|
|
def test_update_subcommand_exists(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["update", "--help"])
|
|
assert result.exit_code == 0
|
|
assert (
|
|
"update" in result.output.lower()
|
|
or "reinstall" in result.output.lower()
|
|
)
|
|
|
|
def test_upgrade_alias_exists(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["upgrade", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
class TestCodexSubcommand:
|
|
"""Tests for ``myagents codex``."""
|
|
|
|
def test_help_shows_options(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["codex", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "--cwd" in result.output
|
|
assert "--list" in result.output
|
|
|
|
def test_runs_codex(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["codex"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once()
|
|
assert mock_run.call_args[0][0] == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox"]
|
|
|
|
def test_missing_binary_error(self) -> None:
|
|
runner = CliRunner()
|
|
with patch("myagents.launcher.shutil.which", return_value=None):
|
|
result = runner.invoke(cli, ["codex"])
|
|
assert result.exit_code == 127
|
|
assert "not found" in result.output.lower()
|
|
|
|
def test_cwd_option_passed(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
test_dir = tmp_path / "test_cwd"
|
|
test_dir.mkdir()
|
|
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["codex", "--cwd", str(test_dir)])
|
|
assert result.exit_code == 0
|
|
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
|
|
|
|
|
|
class TestCodexPassthrough:
|
|
"""Codex resume subcommand forwards to codex instead of erroring."""
|
|
|
|
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(
|
|
cli, ["codex", "--cwd", str(tmp_path), *args]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
mock_run.assert_called_once()
|
|
return mock_run.call_args[0][0]
|
|
|
|
def test_resume_subcommand_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["resume", "abc123"], tmp_path)
|
|
assert cmd == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox", "resume", "abc123"]
|
|
|
|
def test_continue_subcommand_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["continue"], tmp_path)
|
|
assert cmd == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox", "continue"]
|
|
|
|
|
|
class TestCodexResumeOptions:
|
|
"""Map myagents-style resume flags to codex resume subcommand."""
|
|
|
|
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_backend_which("codex")),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(
|
|
cli, ["codex", "--cwd", str(tmp_path), *args]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
mock_run.assert_called_once()
|
|
return mock_run.call_args[0][0]
|
|
|
|
def test_r_flag_maps_to_resume(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-r", "abc123"], tmp_path)
|
|
assert cmd == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox", "resume", "abc123"]
|
|
|
|
def test_resume_flag_maps_to_resume(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--resume", "abc123"], tmp_path)
|
|
assert cmd == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox", "resume", "abc123"]
|
|
|
|
def test_continue_flag_maps_to_continue(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--continue"], tmp_path)
|
|
assert cmd == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox", "continue"]
|
|
|
|
def test_last_flag_maps_to_resume_last(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--last"], tmp_path)
|
|
assert cmd == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox", "resume", "--last"]
|
|
|
|
def test_all_flag_maps_to_resume_all(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--all"], tmp_path)
|
|
assert cmd == ["/usr/bin/codex", "--dangerously-bypass-approvals-and-sandbox", "resume", "--all"]
|
|
|
|
|
|
class TestCodexListSessions:
|
|
"""``myagents codex --list`` reports resumable sessions."""
|
|
|
|
def test_list_empty_directory(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
with patch("myagents.launcher.subprocess.run") as mock_run:
|
|
result = runner.invoke(cli, ["codex", "--cwd", str(tmp_path), "--list"])
|
|
assert result.exit_code == 0
|
|
assert "no sessions" in result.output.lower()
|
|
mock_run.assert_not_called()
|
|
|
|
def test_list_shows_session_id_and_snippet(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
sessions_root = tmp_path / "sessions"
|
|
sessions_root.mkdir(parents=True)
|
|
session_file = (
|
|
sessions_root
|
|
/ "2026"
|
|
/ "01"
|
|
/ "02"
|
|
/ "rollout-2026-01-02T12-00-00-019aeb67-0620-71c1-8cbb-756bc8845c6e.jsonl"
|
|
)
|
|
session_file.parent.mkdir(parents=True)
|
|
session_file.write_text(
|
|
json.dumps(
|
|
{
|
|
"type": "session_meta",
|
|
"timestamp": "2026-01-02T12:00:00Z",
|
|
"payload": {
|
|
"id": "019aeb67-0620-71c1-8cbb-756bc8845c6e",
|
|
"cwd": str(tmp_path),
|
|
},
|
|
}
|
|
)
|
|
+ "\n"
|
|
+ json.dumps(
|
|
{
|
|
"type": "user_message",
|
|
"payload": {
|
|
"type": "user_message",
|
|
"message": "hello codex",
|
|
},
|
|
}
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with patch(
|
|
"myagents.launcher._BACKENDS",
|
|
{
|
|
"codex": {
|
|
**myagents.launcher._BACKENDS["codex"],
|
|
"sessions_root": lambda: sessions_root,
|
|
}
|
|
},
|
|
):
|
|
result = runner.invoke(cli, ["codex", "--cwd", str(tmp_path), "--list"])
|
|
assert result.exit_code == 0, result.output
|
|
assert "019aeb67-0620-71c1-8cbb-756bc8845c6e" in result.output
|
|
assert "hello codex" in result.output
|
|
|
|
|
|
class TestHermesSubcommand:
|
|
"""Tests for ``myagents hermes``."""
|
|
|
|
def test_help_shows_options(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["hermes", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "--cwd" in result.output
|
|
assert "--list" in result.output
|
|
|
|
def test_runs_hermes(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("hermes")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["hermes"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once()
|
|
assert mock_run.call_args[0][0] == ["/usr/bin/hermes", "--yolo"]
|
|
|
|
def test_missing_binary_error(self) -> None:
|
|
runner = CliRunner()
|
|
with patch("myagents.launcher.shutil.which", return_value=None):
|
|
result = runner.invoke(cli, ["hermes"])
|
|
assert result.exit_code == 127
|
|
assert "not found" in result.output.lower()
|
|
|
|
def test_cwd_option_passed(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
test_dir = tmp_path / "test_cwd"
|
|
test_dir.mkdir()
|
|
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("hermes")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["hermes", "--cwd", str(test_dir)])
|
|
assert result.exit_code == 0
|
|
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
|
|
|
|
|
|
class TestHermesPassthrough:
|
|
"""Unknown leading flags forward to hermes instead of erroring."""
|
|
|
|
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
|
|
runner = CliRunner()
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("hermes")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(
|
|
cli, ["hermes", "--cwd", str(tmp_path), *args]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
mock_run.assert_called_once()
|
|
return mock_run.call_args[0][0]
|
|
|
|
def test_resume_with_session_id(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--resume", "abc123"], tmp_path)
|
|
assert cmd == [
|
|
"/usr/bin/hermes",
|
|
"--yolo",
|
|
"--resume",
|
|
"abc123",
|
|
"--no-restore-cwd",
|
|
]
|
|
|
|
def test_bare_r_flag_opens_session_picker(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-r"], tmp_path)
|
|
assert cmd == ["/usr/bin/hermes", "--yolo", "sessions", "browse"]
|
|
|
|
def test_bare_resume_flag_opens_session_picker(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--resume"], tmp_path)
|
|
assert cmd == ["/usr/bin/hermes", "--yolo", "sessions", "browse"]
|
|
|
|
def test_r_flag_with_session_id(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-r", "abc123"], tmp_path)
|
|
assert cmd == [
|
|
"/usr/bin/hermes",
|
|
"--yolo",
|
|
"--resume",
|
|
"abc123",
|
|
"--no-restore-cwd",
|
|
]
|
|
|
|
def test_c_flag_maps_to_continue(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-c"], tmp_path)
|
|
assert cmd == [
|
|
"/usr/bin/hermes",
|
|
"--yolo",
|
|
"--continue",
|
|
"--no-restore-cwd",
|
|
]
|
|
|
|
def test_continue_flag_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--continue"], tmp_path)
|
|
assert cmd == [
|
|
"/usr/bin/hermes",
|
|
"--yolo",
|
|
"--continue",
|
|
"--no-restore-cwd",
|
|
]
|
|
|
|
def test_last_flag_maps_to_continue(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--last"], tmp_path)
|
|
assert cmd == [
|
|
"/usr/bin/hermes",
|
|
"--yolo",
|
|
"--continue",
|
|
"--no-restore-cwd",
|
|
]
|
|
|
|
|
|
class TestHermesListSessions:
|
|
"""``myagents hermes --list`` reads the global SQLite session store."""
|
|
|
|
def _patch_state_db(self, db_path: Path):
|
|
return patch(
|
|
"myagents.launcher._BACKENDS",
|
|
{
|
|
"hermes": {
|
|
**myagents.launcher._BACKENDS["hermes"],
|
|
"state_db": lambda: db_path,
|
|
}
|
|
},
|
|
)
|
|
|
|
def test_list_missing_database(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
with self._patch_state_db(tmp_path / "missing.db"):
|
|
result = runner.invoke(
|
|
cli, ["hermes", "--cwd", str(tmp_path), "--list"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "no sessions" in result.output.lower()
|
|
|
|
def test_list_shows_session_id_and_snippet(self, tmp_path: Path) -> None:
|
|
db_path = tmp_path / "state.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute(
|
|
"CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT NOT NULL,"
|
|
" started_at REAL NOT NULL, title TEXT)"
|
|
)
|
|
conn.execute(
|
|
"CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT,"
|
|
" session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT,"
|
|
" timestamp REAL NOT NULL)"
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO sessions (id, source, started_at)"
|
|
" VALUES ('sess-abc123', 'cli', 1767220800)"
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO messages (session_id, role, content, timestamp)"
|
|
" VALUES ('sess-abc123', 'user', 'hello hermes', 1767220801)"
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
runner = CliRunner()
|
|
with self._patch_state_db(db_path):
|
|
result = runner.invoke(
|
|
cli, ["hermes", "--cwd", str(tmp_path), "--list"]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "sess-abc123" in result.output
|
|
assert "hello hermes" in result.output
|
|
|
|
|
|
class TestCursorSubcommand:
|
|
"""Tests for ``myagents cursor``."""
|
|
|
|
def test_help_shows_options(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["cursor", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "--cwd" in result.output
|
|
assert "--list" in result.output
|
|
|
|
def test_runs_agent(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("agent")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["cursor"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once()
|
|
assert mock_run.call_args[0][0] == ["/usr/bin/agent", "--force"]
|
|
|
|
def test_falls_back_to_cursor_agent(self) -> None:
|
|
runner = CliRunner()
|
|
|
|
def _which(name: str) -> str | None:
|
|
return "/usr/bin/cursor-agent" if name == "cursor-agent" else None
|
|
|
|
with (
|
|
patch("myagents.launcher.shutil.which", side_effect=_which),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
patch.dict("os.environ", {}, clear=False) as env,
|
|
):
|
|
env.pop("CURSOR_BIN", None)
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["cursor"])
|
|
assert result.exit_code == 0
|
|
assert mock_run.call_args[0][0] == [
|
|
"/usr/bin/cursor-agent",
|
|
"--force",
|
|
]
|
|
|
|
def test_missing_binary_error(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", return_value=None),
|
|
patch.dict("os.environ", {}, clear=False) as env,
|
|
):
|
|
env.pop("CURSOR_BIN", None)
|
|
result = runner.invoke(cli, ["cursor"])
|
|
assert result.exit_code == 127
|
|
assert "not found" in result.output.lower()
|
|
|
|
def test_cwd_option_passed(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
test_dir = tmp_path / "test_cwd"
|
|
test_dir.mkdir()
|
|
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("agent")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["cursor", "--cwd", str(test_dir)])
|
|
assert result.exit_code == 0
|
|
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
|
|
|
|
|
|
class TestDshSubcommand:
|
|
"""Tests for ``myagents dsh``."""
|
|
|
|
def test_help_shows_options(self) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["dsh", "--help"])
|
|
assert result.exit_code == 0
|
|
assert "--cwd" in result.output
|
|
assert "--list" in result.output
|
|
|
|
def test_runs_dsh(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("dsh")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["dsh"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once()
|
|
# dsh is a profile launcher: no default args injected.
|
|
assert mock_run.call_args[0][0] == ["/usr/bin/dsh"]
|
|
|
|
def test_missing_binary_error(self) -> None:
|
|
runner = CliRunner()
|
|
with (
|
|
patch("myagents.launcher.shutil.which", return_value=None),
|
|
patch.dict("os.environ", {}, clear=False) as env,
|
|
):
|
|
env.pop("DSH_BIN", None)
|
|
result = runner.invoke(cli, ["dsh"])
|
|
assert result.exit_code == 127
|
|
assert "dsh CLI not found" in result.output
|
|
|
|
def test_cwd_option_passed(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
test_dir = tmp_path / "test_cwd"
|
|
test_dir.mkdir()
|
|
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("dsh")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(cli, ["dsh", "--cwd", str(test_dir)])
|
|
assert result.exit_code == 0
|
|
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
|
|
|
|
|
|
class TestDshListSessions:
|
|
"""``mydsh -l`` reads zstd-compressed dsh session logs."""
|
|
|
|
def _write_session(self, dsh_home: Path, slug: str, sid: str, cwd: str) -> Path:
|
|
zstd = pytest.importorskip("zstandard")
|
|
log = dsh_home / "sessions" / slug / sid / "session.jsonl.zstd"
|
|
log.parent.mkdir(parents=True, exist_ok=True)
|
|
lines = [
|
|
json.dumps(
|
|
{"type": "session", "version": 0, "id": sid, "cwd": cwd}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"type": "session/title",
|
|
"seq": 1,
|
|
"data": {"title": "My dsh test"},
|
|
}
|
|
),
|
|
]
|
|
with log.open("wb") as fh:
|
|
with zstd.ZstdCompressor().stream_writer(fh) as w:
|
|
w.write(("\n".join(lines) + "\n").encode("utf-8"))
|
|
return log
|
|
|
|
def test_list_shows_dsh_sessions(self, tmp_path: Path, monkeypatch) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
dsh_home = tmp_path / "dshhome"
|
|
self._write_session(dsh_home, "slug", "sess-1", str(workspace))
|
|
monkeypatch.setenv("DSH_HOME", str(dsh_home))
|
|
|
|
runner = CliRunner()
|
|
with patch(
|
|
"myagents.launcher.shutil.which", return_value="/usr/bin/dsh"
|
|
):
|
|
result = runner.invoke(
|
|
cli, ["dsh", "-l", "--cwd", str(workspace)]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "sess-1" in result.output
|
|
assert "My dsh test" in result.output
|
|
|
|
def test_first_prompt_reads_zstd(self, tmp_path: Path) -> None:
|
|
from myagents.launcher import _first_prompt_dsh
|
|
|
|
workspace = tmp_path / "ws"
|
|
workspace.mkdir()
|
|
log = self._write_session(tmp_path / "home", "slug", "sid-1", str(workspace))
|
|
assert _first_prompt_dsh(log) == "My dsh test"
|
|
|
|
|
|
class TestCursorPassthrough:
|
|
"""Map myagents-style resume flags to Cursor Agent CLI flags."""
|
|
|
|
def _invoke(self, args: list[str], tmp_path: Path) -> list[str]:
|
|
runner = CliRunner()
|
|
with (
|
|
patch(
|
|
"myagents.launcher.shutil.which", side_effect=_backend_which("agent")
|
|
),
|
|
patch("myagents.launcher.subprocess.run") as mock_run,
|
|
):
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = runner.invoke(
|
|
cli, ["cursor", "--cwd", str(tmp_path), *args]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
mock_run.assert_called_once()
|
|
return mock_run.call_args[0][0]
|
|
|
|
def test_r_flag_maps_to_resume(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-r", "abc123"], tmp_path)
|
|
assert cmd == ["/usr/bin/agent", "--force", "--resume", "abc123"]
|
|
|
|
def test_bare_r_flag_maps_to_resume(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-r"], tmp_path)
|
|
assert cmd == ["/usr/bin/agent", "--force", "--resume"]
|
|
|
|
def test_resume_flag_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--resume", "abc123"], tmp_path)
|
|
assert cmd == ["/usr/bin/agent", "--force", "--resume", "abc123"]
|
|
|
|
def test_c_flag_maps_to_continue(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["-c"], tmp_path)
|
|
assert cmd == ["/usr/bin/agent", "--force", "--continue"]
|
|
|
|
def test_continue_flag_passes_through(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--continue"], tmp_path)
|
|
assert cmd == ["/usr/bin/agent", "--force", "--continue"]
|
|
|
|
def test_last_flag_maps_to_continue(self, tmp_path: Path) -> None:
|
|
cmd = self._invoke(["--last"], tmp_path)
|
|
assert cmd == ["/usr/bin/agent", "--force", "--continue"]
|
|
|
|
|
|
class TestCursorListSessions:
|
|
"""``myagents cursor --list`` reads ~/.cursor/chats/<cwd-md5>/."""
|
|
|
|
def test_list_empty_directory(self, tmp_path: Path) -> None:
|
|
runner = CliRunner()
|
|
with patch(
|
|
"myagents.launcher._BACKENDS",
|
|
{
|
|
"cursor": {
|
|
**myagents.launcher._BACKENDS["cursor"],
|
|
"sessions_root": lambda: tmp_path / "chats",
|
|
}
|
|
},
|
|
):
|
|
result = runner.invoke(
|
|
cli, ["cursor", "--cwd", str(tmp_path), "--list"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "no sessions" in result.output.lower()
|
|
|
|
def test_list_shows_session_id_and_name(self, tmp_path: Path) -> None:
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
|
|
cwd = tmp_path / "proj"
|
|
cwd.mkdir()
|
|
chats_root = tmp_path / "chats"
|
|
munged = hashlib.md5(str(cwd.resolve()).encode("utf-8")).hexdigest()
|
|
session_dir = chats_root / munged / "chat-abc123"
|
|
session_dir.mkdir(parents=True)
|
|
db_path = session_dir / "store.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)")
|
|
conn.execute("CREATE TABLE blobs (id TEXT PRIMARY KEY, data BLOB)")
|
|
meta = {
|
|
"agentId": "chat-abc123",
|
|
"name": "Fix mycursor entrypoint",
|
|
"mode": "default",
|
|
"createdAt": 1,
|
|
}
|
|
conn.execute(
|
|
"INSERT INTO meta (key, value) VALUES ('0', ?)",
|
|
(json.dumps(meta).encode().hex(),),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO blobs (id, data) VALUES ('1', ?)",
|
|
(
|
|
json.dumps(
|
|
{
|
|
"role": "user",
|
|
"content": "<user_query>\nhello cursor\n</user_query>",
|
|
}
|
|
).encode(),
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
runner = CliRunner()
|
|
with patch(
|
|
"myagents.launcher._BACKENDS",
|
|
{
|
|
"cursor": {
|
|
**myagents.launcher._BACKENDS["cursor"],
|
|
"sessions_root": lambda: chats_root,
|
|
}
|
|
},
|
|
):
|
|
result = runner.invoke(
|
|
cli, ["cursor", "--cwd", str(cwd), "--list"],
|
|
env={"COLUMNS": "120"},
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "chat-abc123" in result.output
|
|
assert "Fix mycursor entrypoint" in result.output
|
|
assert "mycursor" in result.output
|