Thales Maciel
4d0081d1d0
Break the old god module into flat siblings for CLI parsing, run lifecycle, daemon state, shared processing helpers, benchmark tooling, and maintainer-only model sync so changes stop sharing one giant import graph. Keep aman as a thin shim over aman_cli, move sync-default-model behind the hidden aman-maint entrypoint plus Make wrappers, and update packaging metadata plus maintainer docs to reflect the new surface. Retarget the tests to the new seams with dedicated runtime, run, benchmark, maintainer, and entrypoint suites, and verify with python3 -m unittest discover -s tests -p "test_*.py", python3 -m py_compile src/*.py tests/*.py, PYTHONPATH=src python3 -m aman --help, PYTHONPATH=src python3 -m aman version, and PYTHONPATH=src python3 -m aman_maint --help.
240 lines
9.5 KiB
Python
240 lines
9.5 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_cli
|
|
from diagnostics import DiagnosticCheck, DiagnosticReport
|
|
|
|
|
|
class AmanCliTests(unittest.TestCase):
|
|
def test_parse_cli_args_help_flag_uses_top_level_parser(self):
|
|
out = io.StringIO()
|
|
|
|
with patch("sys.stdout", out), self.assertRaises(SystemExit) as exc:
|
|
aman_cli.parse_cli_args(["--help"])
|
|
|
|
self.assertEqual(exc.exception.code, 0)
|
|
rendered = out.getvalue()
|
|
self.assertIn("run", rendered)
|
|
self.assertIn("doctor", rendered)
|
|
self.assertIn("self-check", rendered)
|
|
self.assertIn("systemd --user service", rendered)
|
|
|
|
def test_parse_cli_args_short_help_flag_uses_top_level_parser(self):
|
|
out = io.StringIO()
|
|
|
|
with patch("sys.stdout", out), self.assertRaises(SystemExit) as exc:
|
|
aman_cli.parse_cli_args(["-h"])
|
|
|
|
self.assertEqual(exc.exception.code, 0)
|
|
self.assertIn("self-check", out.getvalue())
|
|
|
|
def test_parse_cli_args_defaults_to_run_command(self):
|
|
args = aman_cli.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_cli.parse_cli_args(["doctor", "--json"])
|
|
|
|
self.assertEqual(args.command, "doctor")
|
|
self.assertTrue(args.json)
|
|
|
|
def test_parse_cli_args_self_check_command(self):
|
|
args = aman_cli.parse_cli_args(["self-check", "--json"])
|
|
|
|
self.assertEqual(args.command, "self-check")
|
|
self.assertTrue(args.json)
|
|
|
|
def test_parse_cli_args_bench_command(self):
|
|
args = aman_cli.parse_cli_args(
|
|
["bench", "--text", "hello", "--repeat", "2", "--warmup", "0", "--json"]
|
|
)
|
|
|
|
self.assertEqual(args.command, "bench")
|
|
self.assertEqual(args.text, "hello")
|
|
self.assertEqual(args.repeat, 2)
|
|
self.assertEqual(args.warmup, 0)
|
|
self.assertTrue(args.json)
|
|
|
|
def test_parse_cli_args_bench_requires_input(self):
|
|
with self.assertRaises(SystemExit):
|
|
aman_cli.parse_cli_args(["bench"])
|
|
|
|
def test_parse_cli_args_eval_models_command(self):
|
|
args = aman_cli.parse_cli_args(
|
|
[
|
|
"eval-models",
|
|
"--dataset",
|
|
"benchmarks/cleanup_dataset.jsonl",
|
|
"--matrix",
|
|
"benchmarks/model_matrix.small_first.json",
|
|
]
|
|
)
|
|
self.assertEqual(args.command, "eval-models")
|
|
self.assertEqual(args.dataset, "benchmarks/cleanup_dataset.jsonl")
|
|
self.assertEqual(args.matrix, "benchmarks/model_matrix.small_first.json")
|
|
self.assertEqual(args.heuristic_dataset, "")
|
|
self.assertEqual(args.heuristic_weight, 0.25)
|
|
self.assertEqual(args.report_version, 2)
|
|
|
|
def test_parse_cli_args_eval_models_with_heuristic_options(self):
|
|
args = aman_cli.parse_cli_args(
|
|
[
|
|
"eval-models",
|
|
"--dataset",
|
|
"benchmarks/cleanup_dataset.jsonl",
|
|
"--matrix",
|
|
"benchmarks/model_matrix.small_first.json",
|
|
"--heuristic-dataset",
|
|
"benchmarks/heuristics_dataset.jsonl",
|
|
"--heuristic-weight",
|
|
"0.4",
|
|
"--report-version",
|
|
"2",
|
|
]
|
|
)
|
|
self.assertEqual(args.heuristic_dataset, "benchmarks/heuristics_dataset.jsonl")
|
|
self.assertEqual(args.heuristic_weight, 0.4)
|
|
self.assertEqual(args.report_version, 2)
|
|
|
|
def test_parse_cli_args_build_heuristic_dataset_command(self):
|
|
args = aman_cli.parse_cli_args(
|
|
[
|
|
"build-heuristic-dataset",
|
|
"--input",
|
|
"benchmarks/heuristics_dataset.raw.jsonl",
|
|
"--output",
|
|
"benchmarks/heuristics_dataset.jsonl",
|
|
]
|
|
)
|
|
self.assertEqual(args.command, "build-heuristic-dataset")
|
|
self.assertEqual(args.input, "benchmarks/heuristics_dataset.raw.jsonl")
|
|
self.assertEqual(args.output, "benchmarks/heuristics_dataset.jsonl")
|
|
|
|
def test_parse_cli_args_legacy_maint_command_errors_with_migration_hint(self):
|
|
err = io.StringIO()
|
|
|
|
with patch("sys.stderr", err), self.assertRaises(SystemExit) as exc:
|
|
aman_cli.parse_cli_args(["sync-default-model"])
|
|
|
|
self.assertEqual(exc.exception.code, 2)
|
|
self.assertIn("aman-maint sync-default-model", err.getvalue())
|
|
self.assertIn("make sync-default-model", err.getvalue())
|
|
|
|
def test_version_command_prints_version(self):
|
|
out = io.StringIO()
|
|
args = aman_cli.parse_cli_args(["version"])
|
|
with patch("aman_cli.app_version", return_value="1.2.3"), patch("sys.stdout", out):
|
|
exit_code = aman_cli.version_command(args)
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertEqual(out.getvalue().strip(), "1.2.3")
|
|
|
|
def test_app_version_prefers_local_pyproject_version(self):
|
|
pyproject_text = '[project]\nversion = "9.9.9"\n'
|
|
|
|
with patch.object(aman_cli.Path, "exists", return_value=True), patch.object(
|
|
aman_cli.Path, "read_text", return_value=pyproject_text
|
|
), patch("aman_cli.importlib.metadata.version", return_value="1.0.0"):
|
|
self.assertEqual(aman_cli.app_version(), "9.9.9")
|
|
|
|
def test_doctor_command_json_output_and_exit_code(self):
|
|
report = DiagnosticReport(
|
|
checks=[DiagnosticCheck(id="config.load", status="ok", message="ok", next_step="")]
|
|
)
|
|
args = aman_cli.parse_cli_args(["doctor", "--json"])
|
|
out = io.StringIO()
|
|
with patch("aman_cli.run_doctor", return_value=report), patch("sys.stdout", out):
|
|
exit_code = aman_cli.doctor_command(args)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
payload = json.loads(out.getvalue())
|
|
self.assertTrue(payload["ok"])
|
|
self.assertEqual(payload["status"], "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", status="fail", message="broken", next_step="fix")]
|
|
)
|
|
args = aman_cli.parse_cli_args(["doctor"])
|
|
out = io.StringIO()
|
|
with patch("aman_cli.run_doctor", return_value=report), patch("sys.stdout", out):
|
|
exit_code = aman_cli.doctor_command(args)
|
|
|
|
self.assertEqual(exit_code, 2)
|
|
self.assertIn("[FAIL] config.load", out.getvalue())
|
|
self.assertIn("overall: fail", out.getvalue())
|
|
|
|
def test_doctor_command_warning_report_returns_exit_code_0(self):
|
|
report = DiagnosticReport(
|
|
checks=[DiagnosticCheck(id="model.cache", status="warn", message="missing", next_step="run aman once")]
|
|
)
|
|
args = aman_cli.parse_cli_args(["doctor"])
|
|
out = io.StringIO()
|
|
with patch("aman_cli.run_doctor", return_value=report), patch("sys.stdout", out):
|
|
exit_code = aman_cli.doctor_command(args)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertIn("[WARN] model.cache", out.getvalue())
|
|
self.assertIn("overall: warn", out.getvalue())
|
|
|
|
def test_self_check_command_uses_self_check_runner(self):
|
|
report = DiagnosticReport(
|
|
checks=[DiagnosticCheck(id="startup.readiness", status="ok", message="ready", next_step="")]
|
|
)
|
|
args = aman_cli.parse_cli_args(["self-check", "--json"])
|
|
out = io.StringIO()
|
|
with patch("aman_cli.run_self_check", return_value=report) as runner, patch("sys.stdout", out):
|
|
exit_code = aman_cli.self_check_command(args)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
runner.assert_called_once_with("")
|
|
payload = json.loads(out.getvalue())
|
|
self.assertEqual(payload["status"], "ok")
|
|
|
|
def test_init_command_creates_default_config(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
path = Path(td) / "config.json"
|
|
args = aman_cli.parse_cli_args(["init", "--config", str(path)])
|
|
|
|
exit_code = aman_cli.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_cli.parse_cli_args(["init", "--config", str(path)])
|
|
|
|
exit_code = aman_cli.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_cli.parse_cli_args(["init", "--config", str(path), "--force"])
|
|
|
|
exit_code = aman_cli.init_command(args)
|
|
self.assertEqual(exit_code, 0)
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
self.assertEqual(payload["daemon"]["hotkey"], "Cmd+m")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|