228 lines
8 KiB
Python
228 lines
8 KiB
Python
import io
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SRC = ROOT / "src"
|
|
if str(SRC) not in sys.path:
|
|
sys.path.insert(0, str(SRC))
|
|
|
|
import aman
|
|
from config import Config
|
|
from diagnostics import DiagnosticCheck, DiagnosticReport
|
|
from onboarding_ui import OnboardingResult
|
|
|
|
|
|
class _FakeDesktop:
|
|
def __init__(self):
|
|
self.hotkey = None
|
|
self.hotkey_callback = None
|
|
|
|
def start_hotkey_listener(self, hotkey, callback):
|
|
self.hotkey = hotkey
|
|
self.hotkey_callback = callback
|
|
|
|
def stop_hotkey_listener(self):
|
|
return
|
|
|
|
def start_cancel_listener(self, callback):
|
|
_ = callback
|
|
return
|
|
|
|
def stop_cancel_listener(self):
|
|
return
|
|
|
|
def validate_hotkey(self, hotkey):
|
|
_ = hotkey
|
|
return
|
|
|
|
def inject_text(self, text, backend, *, remove_transcription_from_clipboard=False):
|
|
_ = (text, backend, remove_transcription_from_clipboard)
|
|
return
|
|
|
|
def run_tray(self, _state_getter, on_quit, **_kwargs):
|
|
on_quit()
|
|
|
|
def request_quit(self):
|
|
return
|
|
|
|
|
|
class _FakeDaemon:
|
|
def __init__(self, cfg, _desktop, *, verbose=False):
|
|
self.cfg = cfg
|
|
self.verbose = verbose
|
|
self._paused = False
|
|
|
|
def get_state(self):
|
|
return "idle"
|
|
|
|
def is_paused(self):
|
|
return self._paused
|
|
|
|
def toggle_paused(self):
|
|
self._paused = not self._paused
|
|
return self._paused
|
|
|
|
def apply_config(self, cfg):
|
|
self.cfg = cfg
|
|
|
|
def toggle(self):
|
|
return
|
|
|
|
def shutdown(self, timeout=1.0):
|
|
_ = timeout
|
|
return True
|
|
|
|
|
|
class _RetrySetupDesktop(_FakeDesktop):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.setup_invocations = 0
|
|
|
|
def run_tray(self, _state_getter, on_quit, **kwargs):
|
|
setup_cb = kwargs.get("on_setup_wizard")
|
|
if setup_cb is not None and self.setup_invocations == 0:
|
|
self.setup_invocations += 1
|
|
setup_cb()
|
|
return
|
|
on_quit()
|
|
|
|
|
|
class AmanCliTests(unittest.TestCase):
|
|
def test_parse_cli_args_defaults_to_run_command(self):
|
|
args = aman._parse_cli_args(["--dry-run"])
|
|
|
|
self.assertEqual(args.command, "run")
|
|
self.assertTrue(args.dry_run)
|
|
|
|
def test_parse_cli_args_doctor_command(self):
|
|
args = aman._parse_cli_args(["doctor", "--json"])
|
|
|
|
self.assertEqual(args.command, "doctor")
|
|
self.assertTrue(args.json)
|
|
|
|
def test_doctor_command_json_output_and_exit_code(self):
|
|
report = DiagnosticReport(
|
|
checks=[DiagnosticCheck(id="config.load", ok=True, message="ok", hint="")]
|
|
)
|
|
args = aman._parse_cli_args(["doctor", "--json"])
|
|
out = io.StringIO()
|
|
with patch("aman.run_diagnostics", return_value=report), patch("sys.stdout", out):
|
|
exit_code = aman._doctor_command(args)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
payload = json.loads(out.getvalue())
|
|
self.assertTrue(payload["ok"])
|
|
self.assertEqual(payload["checks"][0]["id"], "config.load")
|
|
|
|
def test_doctor_command_failed_report_returns_exit_code_2(self):
|
|
report = DiagnosticReport(
|
|
checks=[DiagnosticCheck(id="config.load", ok=False, message="broken", hint="fix")]
|
|
)
|
|
args = aman._parse_cli_args(["doctor"])
|
|
out = io.StringIO()
|
|
with patch("aman.run_diagnostics", return_value=report), patch("sys.stdout", out):
|
|
exit_code = aman._doctor_command(args)
|
|
|
|
self.assertEqual(exit_code, 2)
|
|
self.assertIn("[FAIL] config.load", out.getvalue())
|
|
|
|
def test_init_command_creates_default_config(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / "config.json"
|
|
args = aman._parse_cli_args(["init", "--config", str(path)])
|
|
|
|
exit_code = aman._init_command(args)
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertTrue(path.exists())
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
self.assertIn("daemon", payload)
|
|
|
|
def test_init_command_refuses_overwrite_without_force(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / "config.json"
|
|
path.write_text('{"daemon":{"hotkey":"Super+m"}}\n', encoding="utf-8")
|
|
args = aman._parse_cli_args(["init", "--config", str(path)])
|
|
|
|
exit_code = aman._init_command(args)
|
|
self.assertEqual(exit_code, 1)
|
|
self.assertIn("Super+m", path.read_text(encoding="utf-8"))
|
|
|
|
def test_init_command_force_overwrites_existing_config(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / "config.json"
|
|
path.write_text('{"daemon":{"hotkey":"Super+m"}}\n', encoding="utf-8")
|
|
args = aman._parse_cli_args(["init", "--config", str(path), "--force"])
|
|
|
|
exit_code = aman._init_command(args)
|
|
self.assertEqual(exit_code, 0)
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
self.assertEqual(payload["daemon"]["hotkey"], "Cmd+m")
|
|
|
|
def test_run_command_missing_config_uses_onboarding_and_writes_file(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / "config.json"
|
|
args = aman._parse_cli_args(["run", "--config", str(path)])
|
|
desktop = _FakeDesktop()
|
|
onboard_cfg = Config()
|
|
onboard_cfg.daemon.hotkey = "Super+m"
|
|
with patch("aman._lock_single_instance", return_value=object()), patch(
|
|
"aman.get_desktop_adapter", return_value=desktop
|
|
), patch(
|
|
"aman.run_onboarding_wizard",
|
|
return_value=OnboardingResult(completed=True, config=onboard_cfg, aborted_reason=None),
|
|
) as onboarding_mock, patch("aman.Daemon", _FakeDaemon):
|
|
exit_code = aman._run_command(args)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertTrue(path.exists())
|
|
self.assertEqual(desktop.hotkey, "Super+m")
|
|
onboarding_mock.assert_called_once()
|
|
|
|
def test_run_command_missing_config_cancel_returns_without_starting_daemon(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / "config.json"
|
|
args = aman._parse_cli_args(["run", "--config", str(path)])
|
|
desktop = _FakeDesktop()
|
|
with patch("aman._lock_single_instance", return_value=object()), patch(
|
|
"aman.get_desktop_adapter", return_value=desktop
|
|
), patch(
|
|
"aman.run_onboarding_wizard",
|
|
return_value=OnboardingResult(completed=False, config=None, aborted_reason="cancelled"),
|
|
), patch("aman.Daemon") as daemon_cls:
|
|
exit_code = aman._run_command(args)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertFalse(path.exists())
|
|
daemon_cls.assert_not_called()
|
|
|
|
def test_run_command_missing_config_cancel_then_retry_setup(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / "config.json"
|
|
args = aman._parse_cli_args(["run", "--config", str(path)])
|
|
desktop = _RetrySetupDesktop()
|
|
onboard_cfg = Config()
|
|
onboarding_results = [
|
|
OnboardingResult(completed=False, config=None, aborted_reason="cancelled"),
|
|
OnboardingResult(completed=True, config=onboard_cfg, aborted_reason=None),
|
|
]
|
|
with patch("aman._lock_single_instance", return_value=object()), patch(
|
|
"aman.get_desktop_adapter", return_value=desktop
|
|
), patch(
|
|
"aman.run_onboarding_wizard",
|
|
side_effect=onboarding_results,
|
|
), patch("aman.Daemon", _FakeDaemon):
|
|
exit_code = aman._run_command(args)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertTrue(path.exists())
|
|
self.assertEqual(desktop.setup_invocations, 1)
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|