"""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//.""" 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": "\nhello cursor\n", } ).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