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()