323 lines
13 KiB
Python
323 lines
13 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import io
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
from unittest import mock
|
||
|
|
|
||
|
|
import mmo_cli_output
|
||
|
|
import mmoctl
|
||
|
|
|
||
|
|
import scripts.install as install_script
|
||
|
|
|
||
|
|
|
||
|
|
class TTYBuffer(io.StringIO):
|
||
|
|
def isatty(self) -> bool:
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
class UnreadableTTY(TTYBuffer):
|
||
|
|
def read(self, *_args: object, **_kwargs: object) -> str:
|
||
|
|
raise AssertionError("interactive stdin must not be read implicitly")
|
||
|
|
|
||
|
|
|
||
|
|
def parser_tree(
|
||
|
|
root: argparse.ArgumentParser,
|
||
|
|
) -> list[tuple[tuple[str, ...], argparse.ArgumentParser]]:
|
||
|
|
found: list[tuple[tuple[str, ...], argparse.ArgumentParser]] = []
|
||
|
|
pending: list[tuple[tuple[str, ...], argparse.ArgumentParser]] = [((), root)]
|
||
|
|
while pending:
|
||
|
|
path, parser = pending.pop()
|
||
|
|
found.append((path, parser))
|
||
|
|
for action in parser._actions:
|
||
|
|
if isinstance(action, argparse._SubParsersAction):
|
||
|
|
pending.extend(((*path, name), child) for name, child in action.choices.items())
|
||
|
|
return found
|
||
|
|
|
||
|
|
|
||
|
|
class CLIUXTests(unittest.TestCase):
|
||
|
|
def test_complete_command_surface_has_descriptions_and_option_help(self) -> None:
|
||
|
|
parsers = parser_tree(mmoctl._parser())
|
||
|
|
self.assertEqual(len(parsers), 69)
|
||
|
|
for path, parser in parsers:
|
||
|
|
with self.subTest(command=" ".join(path) or "root"):
|
||
|
|
self.assertTrue(parser.description)
|
||
|
|
help_text = parser.format_help()
|
||
|
|
self.assertIn("usage:", help_text)
|
||
|
|
self.assertNotIn("\x1b[", help_text)
|
||
|
|
for action in parser._actions:
|
||
|
|
if isinstance(action, (argparse._HelpAction, argparse._SubParsersAction)):
|
||
|
|
continue
|
||
|
|
self.assertNotIn(action.help, (None, "", argparse.SUPPRESS))
|
||
|
|
|
||
|
|
def test_entrypoint_routing_has_one_source_of_truth(self) -> None:
|
||
|
|
parser = mmoctl._parser(prog="codex-mmo")
|
||
|
|
self.assertEqual(mmoctl._route_argv(parser, [], implicit_run=True), ["run"])
|
||
|
|
self.assertEqual(
|
||
|
|
mmoctl._route_argv(parser, ["profile", "list"], implicit_run=True),
|
||
|
|
["profile", "list"],
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
mmoctl._route_argv(parser, ["profile", "list", "--json"], implicit_run=True),
|
||
|
|
["--json", "profile", "list"],
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
mmoctl._route_argv(parser, ["--quiet", "--", "--search"], implicit_run=True),
|
||
|
|
["--quiet", "run", "--", "--search"],
|
||
|
|
)
|
||
|
|
self.assertEqual(
|
||
|
|
mmoctl._route_argv(parser, ["run", "--", "--json"], implicit_run=True),
|
||
|
|
["run", "--", "--json"],
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_usage_errors_are_local_actionable_and_machine_readable(self) -> None:
|
||
|
|
stdout = io.StringIO()
|
||
|
|
stderr = io.StringIO()
|
||
|
|
with mock.patch("sys.stdout", stdout), mock.patch("sys.stderr", stderr):
|
||
|
|
status = mmoctl.main(["session", "list", "--limt", "1"])
|
||
|
|
self.assertEqual(status, 2)
|
||
|
|
self.assertEqual(stdout.getvalue(), "")
|
||
|
|
self.assertIn("usage: codex-mmoctl session list", stderr.getvalue())
|
||
|
|
self.assertIn("Did you mean '--limit'?", stderr.getvalue())
|
||
|
|
|
||
|
|
stdout = io.StringIO()
|
||
|
|
stderr = io.StringIO()
|
||
|
|
with mock.patch("sys.stdout", stdout), mock.patch("sys.stderr", stderr):
|
||
|
|
status = mmoctl.main(["session", "list", "--limt", "1", "--json"])
|
||
|
|
self.assertEqual(status, 2)
|
||
|
|
self.assertEqual(stdout.getvalue(), "")
|
||
|
|
error = json.loads(stderr.getvalue())
|
||
|
|
self.assertEqual(error["error_type"], "usage")
|
||
|
|
self.assertIn("--limit", error["hint"])
|
||
|
|
|
||
|
|
def test_cross_option_errors_fail_before_runtime_initialization(self) -> None:
|
||
|
|
cases = (
|
||
|
|
(["resume"], "exactly one"),
|
||
|
|
(["resume", "session-id", "--last"], "exactly one"),
|
||
|
|
(["resume", "session-id", "--all"], "only with --last"),
|
||
|
|
(["doctor", "--probe"], "requires --live"),
|
||
|
|
(["eval", "compare", "one-run"], "at least two"),
|
||
|
|
(
|
||
|
|
["catalog", "refresh", "--no-codex", "--install-codex-overlay"],
|
||
|
|
"cannot be combined",
|
||
|
|
),
|
||
|
|
(
|
||
|
|
[
|
||
|
|
"catalog",
|
||
|
|
"refresh",
|
||
|
|
"--no-remote",
|
||
|
|
"--openrouter-url",
|
||
|
|
"https://example.invalid/models",
|
||
|
|
],
|
||
|
|
"cannot be combined",
|
||
|
|
),
|
||
|
|
(
|
||
|
|
["catalog", "refresh", "--no-codex", "--codex-bin", "/bin/false"],
|
||
|
|
"cannot be combined",
|
||
|
|
),
|
||
|
|
(["catalog", "discover", "codex", "--timeout", "2"], "not valid"),
|
||
|
|
(
|
||
|
|
[
|
||
|
|
"catalog",
|
||
|
|
"discover",
|
||
|
|
"opencode-go",
|
||
|
|
"--url",
|
||
|
|
"https://one.invalid/models",
|
||
|
|
"--opencode-url",
|
||
|
|
"https://two.invalid/models",
|
||
|
|
],
|
||
|
|
"not both",
|
||
|
|
),
|
||
|
|
(
|
||
|
|
["catalog", "discover", "openrouter", "--codex-bin", "/bin/false"],
|
||
|
|
"not valid",
|
||
|
|
),
|
||
|
|
(["run", "--json"], "not supported"),
|
||
|
|
)
|
||
|
|
for argv, expected in cases:
|
||
|
|
with self.subTest(argv=argv):
|
||
|
|
stderr = io.StringIO()
|
||
|
|
with mock.patch("sys.stdout", io.StringIO()), mock.patch("sys.stderr", stderr):
|
||
|
|
status = mmoctl.main(argv)
|
||
|
|
self.assertEqual(status, 2)
|
||
|
|
self.assertIn(expected, stderr.getvalue())
|
||
|
|
|
||
|
|
def test_exec_never_blocks_for_an_implicit_interactive_prompt(self) -> None:
|
||
|
|
stderr = io.StringIO()
|
||
|
|
with (
|
||
|
|
mock.patch("sys.stdin", UnreadableTTY()),
|
||
|
|
mock.patch("sys.stdout", io.StringIO()),
|
||
|
|
mock.patch("sys.stderr", stderr),
|
||
|
|
):
|
||
|
|
status = mmoctl.main(["exec"])
|
||
|
|
self.assertEqual(status, 2)
|
||
|
|
self.assertIn("requires PROMPT or piped stdin", stderr.getvalue())
|
||
|
|
|
||
|
|
stdout = io.StringIO()
|
||
|
|
with (
|
||
|
|
mock.patch("sys.stdin", io.StringIO("inspect this\n")),
|
||
|
|
mock.patch("sys.stdout", stdout),
|
||
|
|
mock.patch("sys.stderr", io.StringIO()),
|
||
|
|
mock.patch(
|
||
|
|
"mmoctl.run_root_exec",
|
||
|
|
return_value={"exit_code": 0, "result": "complete"},
|
||
|
|
) as execute,
|
||
|
|
):
|
||
|
|
status = mmoctl.main(["exec", "--profile", "fixture"])
|
||
|
|
self.assertEqual(status, 0)
|
||
|
|
self.assertEqual(stdout.getvalue(), "complete\n")
|
||
|
|
self.assertEqual(execute.call_args.kwargs["prompt"], "inspect this\n")
|
||
|
|
|
||
|
|
def test_structured_output_is_human_on_tty_and_json_in_pipelines(self) -> None:
|
||
|
|
session_id = "20260823-120000-reverse-engineering-0123456789"
|
||
|
|
sessions = [
|
||
|
|
{
|
||
|
|
"session_id": session_id,
|
||
|
|
"status": "paused",
|
||
|
|
"profile_id": "reverse-engineering",
|
||
|
|
"session_kind": "interactive",
|
||
|
|
"last_active_at": "2026-08-23T12:00:00+00:00",
|
||
|
|
"resumable": True,
|
||
|
|
}
|
||
|
|
]
|
||
|
|
human = TTYBuffer()
|
||
|
|
with mock.patch("sys.stdout", human), mock.patch.dict(os.environ, {"COLUMNS": "50"}):
|
||
|
|
mmo_cli_output.emit_structured("session.list", sessions)
|
||
|
|
self.assertIn(session_id, human.getvalue())
|
||
|
|
self.assertIn("Status: paused", human.getvalue())
|
||
|
|
self.assertNotIn("\x1b[", human.getvalue())
|
||
|
|
|
||
|
|
machine = io.StringIO()
|
||
|
|
with mock.patch("sys.stdout", machine):
|
||
|
|
mmo_cli_output.emit_structured("session.list", sessions)
|
||
|
|
self.assertEqual(json.loads(machine.getvalue()), sessions)
|
||
|
|
|
||
|
|
forced = TTYBuffer()
|
||
|
|
with mock.patch("sys.stdout", forced):
|
||
|
|
mmo_cli_output.emit_structured("session.list", sessions, force_json=True)
|
||
|
|
self.assertEqual(json.loads(forced.getvalue()), sessions)
|
||
|
|
|
||
|
|
def test_human_empty_states_are_unambiguous(self) -> None:
|
||
|
|
output = TTYBuffer()
|
||
|
|
with mock.patch("sys.stdout", output):
|
||
|
|
mmo_cli_output.emit_structured("jobs.list", [])
|
||
|
|
self.assertEqual(output.getvalue(), "No jobs found.\n")
|
||
|
|
|
||
|
|
def test_catalog_tables_show_domain_fields_instead_of_empty_generic_columns(self) -> None:
|
||
|
|
resources = {
|
||
|
|
"chatgpt_subscription": {
|
||
|
|
"max_active": 4,
|
||
|
|
"lock_key": "provider:chatgpt",
|
||
|
|
"description": "Built-in ChatGPT account concurrency",
|
||
|
|
}
|
||
|
|
}
|
||
|
|
output = TTYBuffer()
|
||
|
|
with mock.patch("sys.stdout", output), mock.patch.dict(os.environ, {"COLUMNS": "120"}):
|
||
|
|
mmo_cli_output.emit_structured("catalog.resources", resources)
|
||
|
|
rendered = output.getvalue()
|
||
|
|
self.assertIn("CAPACITY", rendered)
|
||
|
|
self.assertIn("provider:chatgpt", rendered)
|
||
|
|
self.assertIn("Built-in ChatGPT", rendered)
|
||
|
|
|
||
|
|
def test_progress_uses_only_interactive_stderr_and_honors_quiet(self) -> None:
|
||
|
|
stdout = TTYBuffer()
|
||
|
|
stderr = TTYBuffer()
|
||
|
|
with mock.patch("sys.stdout", stdout), mock.patch("sys.stderr", stderr):
|
||
|
|
mmo_cli_output.progress("Checking profile...")
|
||
|
|
self.assertEqual(stdout.getvalue(), "")
|
||
|
|
self.assertEqual(stderr.getvalue(), "Checking profile...\n")
|
||
|
|
|
||
|
|
quiet = TTYBuffer()
|
||
|
|
with mock.patch("sys.stderr", quiet):
|
||
|
|
mmo_cli_output.progress("Checking profile...", quiet=True)
|
||
|
|
self.assertEqual(quiet.getvalue(), "")
|
||
|
|
|
||
|
|
redirected = io.StringIO()
|
||
|
|
with mock.patch("sys.stderr", redirected):
|
||
|
|
mmo_cli_output.progress("Checking profile...")
|
||
|
|
self.assertEqual(redirected.getvalue(), "")
|
||
|
|
|
||
|
|
def test_runtime_errors_never_pollute_structured_stdout(self) -> None:
|
||
|
|
stdout = io.StringIO()
|
||
|
|
stderr = io.StringIO()
|
||
|
|
with (
|
||
|
|
mock.patch("sys.stdout", stdout),
|
||
|
|
mock.patch("sys.stderr", stderr),
|
||
|
|
mock.patch("mmoctl._handle_catalog", side_effect=RuntimeError("provider failed")),
|
||
|
|
):
|
||
|
|
status = mmoctl.main(["catalog", "summary", "--json"])
|
||
|
|
self.assertEqual(status, 1)
|
||
|
|
self.assertEqual(stdout.getvalue(), "")
|
||
|
|
error = json.loads(stderr.getvalue())
|
||
|
|
self.assertEqual(error["error_type"], "runtime")
|
||
|
|
self.assertEqual(error["error"], "provider failed")
|
||
|
|
|
||
|
|
def test_scalar_commands_offer_explicit_json_without_changing_text_mode(self) -> None:
|
||
|
|
stdout = io.StringIO()
|
||
|
|
with (
|
||
|
|
mock.patch("sys.stdout", stdout),
|
||
|
|
mock.patch("mmoctl.active_profile_id", return_value="reverse-engineering"),
|
||
|
|
):
|
||
|
|
self.assertEqual(mmoctl.main(["profile", "current"]), 0)
|
||
|
|
self.assertEqual(stdout.getvalue(), "reverse-engineering\n")
|
||
|
|
|
||
|
|
stdout = io.StringIO()
|
||
|
|
with (
|
||
|
|
mock.patch("sys.stdout", stdout),
|
||
|
|
mock.patch("mmoctl.active_profile_id", return_value="reverse-engineering"),
|
||
|
|
):
|
||
|
|
self.assertEqual(mmoctl.main(["profile", "current", "--json"]), 0)
|
||
|
|
self.assertEqual(json.loads(stdout.getvalue()), {"profile_id": "reverse-engineering"})
|
||
|
|
|
||
|
|
def test_clean_reports_scope_and_preserves_dry_run_semantics(self) -> None:
|
||
|
|
stdout = io.StringIO()
|
||
|
|
with (
|
||
|
|
mock.patch("sys.stdout", stdout),
|
||
|
|
mock.patch(
|
||
|
|
"mmoctl.load_settings",
|
||
|
|
return_value={"job_retention_days": 30, "session_retention_days": 90},
|
||
|
|
),
|
||
|
|
mock.patch(
|
||
|
|
"mmoctl.clean_state", return_value={"jobs_removed": 4, "sessions_removed": 2}
|
||
|
|
) as clean,
|
||
|
|
):
|
||
|
|
self.assertEqual(mmoctl.main(["clean", "--dry-run"]), 0)
|
||
|
|
clean.assert_called_once_with(job_days=30, session_days=90, dry_run=True)
|
||
|
|
self.assertEqual(
|
||
|
|
json.loads(stdout.getvalue()),
|
||
|
|
{
|
||
|
|
"dry_run": True,
|
||
|
|
"job_days": 30,
|
||
|
|
"jobs_removed": 4,
|
||
|
|
"session_days": 90,
|
||
|
|
"sessions_removed": 2,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_generated_wrappers_delegate_all_routing_to_the_python_cli(self) -> None:
|
||
|
|
entrypoint = Path("/tmp/codex mmo/mmoctl.py")
|
||
|
|
primary = install_script.main_wrapper(Path("/usr/bin/python3"), entrypoint)
|
||
|
|
control = install_script.wrapper(
|
||
|
|
Path("/usr/bin/python3"),
|
||
|
|
entrypoint,
|
||
|
|
environment={"MMO_CLI_ENTRYPOINT": "codex-mmoctl"},
|
||
|
|
)
|
||
|
|
self.assertIn("export MMO_CLI_ENTRYPOINT=codex-mmo", primary)
|
||
|
|
self.assertIn("export MMO_CLI_ENTRYPOINT=codex-mmoctl", control)
|
||
|
|
self.assertNotIn("case ", primary)
|
||
|
|
self.assertIn("'$@'".replace("'", '"'), primary)
|
||
|
|
|
||
|
|
def test_installer_help_is_descriptive_and_color_free(self) -> None:
|
||
|
|
help_text = install_script.parser().format_help()
|
||
|
|
self.assertIn("atomically replace", help_text)
|
||
|
|
self.assertIn("--install-root", help_text)
|
||
|
|
self.assertIn("--debug", help_text)
|
||
|
|
self.assertNotIn("\x1b[", help_text)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|