"""Comprehensive tests for cli.py subcommands.""" import argparse import json import signal from contextlib import ExitStack from unittest.mock import patch, MagicMock import pytest import cli # --------------------------------------------------------------------------- # Helpers / fixtures # --------------------------------------------------------------------------- def _args(**kwargs): """Create an argparse.Namespace with given keyword attrs.""" return argparse.Namespace(**kwargs) @pytest.fixture def iso(tmp_path, monkeypatch): """Redirect all cli module-level paths to tmp_path for isolation.""" pid_file = tmp_path / "echo-core.pid" logs_dir = tmp_path / "logs" logs_dir.mkdir() log_file = logs_dir / "echo-core.log" sess_dir = tmp_path / "sessions" sess_dir.mkdir() sessions_file = sess_dir / "active.json" config_file = tmp_path / "config.json" monkeypatch.setattr(cli, "PID_FILE", pid_file) monkeypatch.setattr(cli, "LOG_FILE", log_file) monkeypatch.setattr(cli, "SESSIONS_FILE", sessions_file) monkeypatch.setattr(cli, "CONFIG_FILE", config_file) monkeypatch.setattr(cli, "PROJECT_ROOT", tmp_path) return { "pid_file": pid_file, "log_file": log_file, "sessions_file": sessions_file, "config_file": config_file, } # --------------------------------------------------------------------------- # cmd_status # --------------------------------------------------------------------------- class TestStatus: def test_offline_no_pid(self, iso, capsys): cli.cmd_status(_args()) out = capsys.readouterr().out assert "OFFLINE" in out assert "no PID file" in out def test_offline_invalid_pid(self, iso, capsys): iso["pid_file"].write_text("garbage") cli.cmd_status(_args()) out = capsys.readouterr().out assert "OFFLINE" in out assert "invalid PID file" in out def test_offline_stale_pid(self, iso, capsys): iso["pid_file"].write_text("999999") with patch("os.kill", side_effect=OSError): cli.cmd_status(_args()) out = capsys.readouterr().out assert "OFFLINE" in out assert "not running" in out def test_online(self, iso, capsys): iso["pid_file"].write_text("1234") iso["sessions_file"].write_text(json.dumps({"ch1": {}})) with patch("os.kill"): cli.cmd_status(_args()) out = capsys.readouterr().out assert "ONLINE" in out assert "1234" in out assert "1 active" in out def test_sessions_count_zero(self, iso, capsys): cli.cmd_status(_args()) out = capsys.readouterr().out assert "0 active" in out # --------------------------------------------------------------------------- # cmd_doctor # --------------------------------------------------------------------------- class TestDoctor: """Tests for the doctor diagnostic command.""" def _run_doctor(self, iso, capsys, *, token="tok", claude="/usr/bin/claude", disk_bavail=1_000_000, disk_frsize=4096): """Run cmd_doctor with mocked externals, return (stdout, exit_code).""" stat = MagicMock(f_bavail=disk_bavail, f_frsize=disk_frsize) patches = [ patch("cli.get_secret", return_value=token), patch("keyring.get_password", return_value=None), patch("shutil.which", return_value=claude), patch("os.statvfs", return_value=stat), ] with ExitStack() as stack: for p in patches: stack.enter_context(p) try: cli.cmd_doctor(_args()) return capsys.readouterr().out, 0 except SystemExit as exc: return capsys.readouterr().out, exc.code def test_all_pass(self, iso, capsys): iso["config_file"].write_text('{"bot":{}}') out, code = self._run_doctor(iso, capsys) assert "All checks passed" in out assert "[FAIL]" not in out assert code == 0 def test_missing_token(self, iso, capsys): iso["config_file"].write_text('{"bot":{}}') out, code = self._run_doctor(iso, capsys, token=None) assert "[FAIL] Discord token" in out assert code == 1 def test_missing_claude(self, iso, capsys): iso["config_file"].write_text('{"bot":{}}') out, code = self._run_doctor(iso, capsys, claude=None) assert "[FAIL] Claude CLI found" in out assert code == 1 def test_invalid_config(self, iso, capsys): # config_file not created -> FileNotFoundError out, code = self._run_doctor(iso, capsys) assert "[FAIL] config.json valid" in out assert code == 1 def test_low_disk_space(self, iso, capsys): iso["config_file"].write_text('{"bot":{}}') out, code = self._run_doctor(iso, capsys, disk_bavail=10, disk_frsize=4096) assert "[FAIL]" in out assert "Disk space" in out assert code == 1 # --------------------------------------------------------------------------- # cmd_restart # --------------------------------------------------------------------------- class TestRestart: def test_no_pid_file(self, iso, capsys): with pytest.raises(SystemExit): cli.cmd_restart(_args()) assert "no PID file" in capsys.readouterr().out def test_invalid_pid(self, iso, capsys): iso["pid_file"].write_text("nope") with pytest.raises(SystemExit): cli.cmd_restart(_args()) assert "invalid PID" in capsys.readouterr().out def test_dead_process(self, iso, capsys): iso["pid_file"].write_text("99999") with patch("os.kill", side_effect=OSError): with pytest.raises(SystemExit): cli.cmd_restart(_args()) assert "not running" in capsys.readouterr().out def test_sends_sigterm(self, iso, capsys): iso["pid_file"].write_text("42") calls = [] with patch("os.kill", side_effect=lambda p, s: calls.append((p, s))): cli.cmd_restart(_args()) assert (42, 0) in calls assert (42, signal.SIGTERM) in calls assert "SIGTERM" in capsys.readouterr().out assert "42" in capsys.readouterr().out or True # already consumed above # --------------------------------------------------------------------------- # cmd_logs # --------------------------------------------------------------------------- class TestLogs: def test_missing_file(self, iso, capsys): cli.cmd_logs(_args(lines=20)) assert "No log file" in capsys.readouterr().out def test_empty_file(self, iso, capsys): iso["log_file"].write_text("") cli.cmd_logs(_args(lines=20)) assert capsys.readouterr().out == "" def test_last_n_lines(self, iso, capsys): iso["log_file"].write_text("\n".join(f"L{i}" for i in range(50))) cli.cmd_logs(_args(lines=5)) lines = capsys.readouterr().out.strip().splitlines() assert len(lines) == 5 assert lines[0] == "L45" assert lines[-1] == "L49" def test_fewer_than_n(self, iso, capsys): iso["log_file"].write_text("a\nb\nc") cli.cmd_logs(_args(lines=20)) lines = capsys.readouterr().out.strip().splitlines() assert len(lines) == 3 assert lines == ["a", "b", "c"] def test_exact_n(self, iso, capsys): iso["log_file"].write_text("\n".join(f"x{i}" for i in range(5))) cli.cmd_logs(_args(lines=5)) lines = capsys.readouterr().out.strip().splitlines() assert len(lines) == 5 # --------------------------------------------------------------------------- # sessions list # --------------------------------------------------------------------------- class TestSessionsList: def test_empty(self, iso, capsys): cli._sessions_list() assert "No active sessions" in capsys.readouterr().out def test_missing_file(self, iso, capsys): # sessions_file doesn't exist yet iso["sessions_file"].parent.joinpath("active.json") # no write cli._sessions_list() assert "No active sessions" in capsys.readouterr().out def test_shows_sessions(self, iso, capsys): data = { "ch1": {"model": "sonnet", "message_count": 5, "last_message_at": "2025-01-15T10:30:00.000+00:00"}, "ch2": {"model": "opus", "message_count": 12, "last_message_at": "2025-01-15T11:00:00.000+00:00"}, } iso["sessions_file"].write_text(json.dumps(data)) cli._sessions_list() out = capsys.readouterr().out assert "ch1" in out assert "sonnet" in out assert "ch2" in out assert "opus" in out # --------------------------------------------------------------------------- # sessions clear # --------------------------------------------------------------------------- class TestSessionsClear: def test_clear_specific(self, iso, capsys): with patch("src.claude_session.clear_session", return_value=True) as m: cli._sessions_clear("ch1") m.assert_called_once_with("ch1") assert "cleared" in capsys.readouterr().out.lower() def test_clear_specific_not_found(self, iso, capsys): with patch("src.claude_session.clear_session", return_value=False): cli._sessions_clear("missing") assert "No session found" in capsys.readouterr().out def test_clear_all(self, iso, capsys): with patch("src.claude_session.list_sessions", return_value={"a": {}, "b": {}}), \ patch("src.claude_session.clear_session") as mc: cli._sessions_clear(None) assert mc.call_count == 2 assert "2 session(s)" in capsys.readouterr().out def test_clear_all_empty(self, iso, capsys): with patch("src.claude_session.list_sessions", return_value={}): cli._sessions_clear(None) assert "No active sessions" in capsys.readouterr().out # --------------------------------------------------------------------------- # channel add # --------------------------------------------------------------------------- class TestChannelAdd: def test_adds_channel(self, iso, capsys): mock_cfg = MagicMock() mock_cfg.get.return_value = {} with patch("src.config.Config", return_value=mock_cfg): cli._channel_add("123456", "general") mock_cfg.set.assert_called_once_with( "channels", {"general": {"id": "123456"}} ) mock_cfg.save.assert_called_once() assert "added" in capsys.readouterr().out.lower() def test_duplicate_alias(self, iso, capsys): mock_cfg = MagicMock() mock_cfg.get.return_value = {"general": {"id": "111"}} with patch("src.config.Config", return_value=mock_cfg): with pytest.raises(SystemExit): cli._channel_add("999", "general") assert "already exists" in capsys.readouterr().out # --------------------------------------------------------------------------- # channel list # --------------------------------------------------------------------------- class TestChannelList: def test_empty(self, iso, capsys): mock_cfg = MagicMock() mock_cfg.get.return_value = {} with patch("src.config.Config", return_value=mock_cfg): cli._channel_list() assert "No channels registered" in capsys.readouterr().out def test_shows_channels(self, iso, capsys): mock_cfg = MagicMock() mock_cfg.get.return_value = { "general": {"id": "111"}, "dev": {"id": "222"}, } with patch("src.config.Config", return_value=mock_cfg): cli._channel_list() out = capsys.readouterr().out assert "general" in out assert "111" in out assert "dev" in out assert "222" in out # --------------------------------------------------------------------------- # cmd_send # --------------------------------------------------------------------------- class TestSend: def test_send_message(self, iso, capsys): mock_cfg = MagicMock() mock_cfg.get.return_value = {"test": {"id": "chan1"}} with patch("src.config.Config", return_value=mock_cfg), \ patch("src.router.route_message", return_value=("Hello!", False)) as mock_route: cli.cmd_send(_args(alias="test", message=["hi", "there"])) mock_route.assert_called_once_with("chan1", "cli-user", "hi there") out = capsys.readouterr().out assert "Hello!" in out def test_unknown_alias(self, iso, capsys): mock_cfg = MagicMock() mock_cfg.get.return_value = {} with patch("src.config.Config", return_value=mock_cfg): with pytest.raises(SystemExit): cli.cmd_send(_args(alias="nope", message=["hi"])) assert "unknown channel" in capsys.readouterr().out.lower()