feat(cli): add cvs man command

cvs/cli_plugins/man_plugin.py

@@ -0,0 +1,95 @@
+import sys
+
+from .list_plugin import ListPlugin
+from cvs.lib.man_lib import find_parameters, iter_parameters, render_json, render_text
+from cvs.parsers.config_registry import TEST_CONFIG_DOCS, documented_tests, get_config_doc, resolve_sample_path
+
+
+class ManPlugin(ListPlugin):
+    def get_name(self):
+        return "man"
+
+    def get_parser(self, subparsers):
+        parser = subparsers.add_parser("man", help="Explain the config parameters for a test")
+        parser.add_argument("test", nargs="?", help="Test to explain. Omit to list tests that have a man page.")
+        parser.add_argument("parameter", nargs="?", help="Optional: show only parameters matching this name")
+        parser.add_argument("--json", action="store_true", dest="as_json", help="Emit the reference as JSON")
+        parser.set_defaults(_plugin=self)
+        return parser
+
+    def get_epilog(self):
+        return """
+Man Commands:
+  cvs man                              List tests that have a config parameter reference
+  cvs man rccl_perf                    Explain every config parameter for rccl_perf
+  cvs man rccl_perf nic_model          Explain a single parameter
+  cvs man rccl_perf --json             Emit the reference as JSON"""
+
+    @staticmethod
+    def _parameters_for(doc):
+        """Flatten every documented section of a test into one parameter list."""
+        params = []
+        for section in doc.sections:
+            params.extend(iter_parameters(section.model, prefix=section.key))
+        return params
+
+    def _list_documented(self):
+        print("\nConfig parameter references")
+        print("=" * 80)
+        for test_name in documented_tests():
+            print(f"\n  • {test_name}")
+            print(f"      {TEST_CONFIG_DOCS[test_name].summary}")
+
+        print(f"\n{'=' * 80}")
+        print(f"Total: {len(TEST_CONFIG_DOCS)} tests with a config parameter reference")
+
+        undocumented = set()
+        for tests in self.test_map.values():
+            undocumented.update(set(tests) - set(TEST_CONFIG_DOCS))
+        if undocumented:
+            print(f"{len(undocumented)} other test suites have no reference yet.")
+        print("\nUse 'cvs man <test>' to explain a test's config parameters.\n")
+
+    def run(self, args):
+        # main() parses with parse_known_args, so an unrecognised flag lands here
+        # silently instead of erroring. Reject it rather than ignore it.
+        unknown = getattr(args, "extra_pytest_args", None)
+        if unknown:
+            self._emit_error(f"Error: unrecognized arguments: {' '.join(unknown)}", args.as_json)
+            self._emit_error("Use 'cvs man --help' to see the available options.", args.as_json)
+            sys.exit(1)
+
+        if not args.test:
+            self._list_documented()
+            return
+
+        doc = get_config_doc(args.test)
+        if not doc:
+            self._emit_error(f"Error: no config parameter reference for '{args.test}'", args.as_json)
+            if self._find_test(args.test) is not None:
+                msg = "This test exists but is not documented yet. Use 'cvs man' to see what is."
+                self._emit_error(msg, args.as_json)
+            else:
+                self._emit_error("Use 'cvs list' to see available tests.", args.as_json)
+            sys.exit(1)
+
+        params = self._parameters_for(doc)
+        title = f"cvs man {args.test}"
+
+        if args.parameter:
+            matches = find_parameters(params, args.parameter)
+            if not matches:
+                self._emit_error(f"Error: no parameter matching '{args.parameter}' in {args.test}", args.as_json)
+                self._emit_error(f"Use 'cvs man {args.test}' to see every parameter.", args.as_json)
+                sys.exit(1)
+            params = matches
+            title = f"{title} {args.parameter}"
+
+        sample_paths = [resolve_sample_path(sample) for sample in doc.samples]
+
+        if args.as_json:
+            print(render_json(params, test=args.test, config_files=sample_paths))
+            return
+
+        print(render_text(params, title=title, summary=doc.summary))
+        print(f"Sample config: {', '.join(sample_paths)}\n")

cvs/cli_plugins/unittests/test_man_plugin.py

@@ -0,0 +1,141 @@
+import argparse
+import io
+import json
+import os
+import unittest
+from contextlib import redirect_stderr, redirect_stdout
+
+from cvs.cli_plugins.man_plugin import ManPlugin
+from cvs.parsers.config_registry import resolve_sample_path
+
+
+def make_args(test=None, parameter=None, as_json=False, extra_pytest_args=None):
+    args = argparse.Namespace()
+    args.test = test
+    args.parameter = parameter
+    args.as_json = as_json
+    args.extra_pytest_args = extra_pytest_args if extra_pytest_args is not None else []
+    return args
+
+
+def run_plugin(plugin, args):
+    buf = io.StringIO()
+    with redirect_stdout(buf):
+        plugin.run(args)
+    return buf.getvalue()
+
+
+class TestManPlugin(unittest.TestCase):
+    def setUp(self):
+        self.plugin = ManPlugin()
+
+    def test_get_name(self):
+        self.assertEqual("man", self.plugin.get_name())
+
+    def test_registers_itself_for_dispatch(self):
+        parser = argparse.ArgumentParser()
+        subparsers = parser.add_subparsers(dest="command")
+        self.plugin.get_parser(subparsers)
+
+        parsed = parser.parse_args(["man", "rccl_perf", "nic_model", "--json"])
+        self.assertIs(self.plugin, parsed._plugin)
+        self.assertEqual("rccl_perf", parsed.test)
+        self.assertEqual("nic_model", parsed.parameter)
+        self.assertTrue(parsed.as_json)
+
+    def test_test_argument_is_optional(self):
+        parser = argparse.ArgumentParser()
+        subparsers = parser.add_subparsers(dest="command")
+        self.plugin.get_parser(subparsers)
+
+        parsed = parser.parse_args(["man"])
+        self.assertIsNone(parsed.test)
+
+    def test_lists_documented_tests(self):
+        output = run_plugin(self.plugin, make_args())
+        self.assertIn("rccl_perf", output)
+        self.assertIn("preflight_checks", output)
+        self.assertIn("Total:", output)
+
+    def test_explains_a_test(self):
+        output = run_plugin(self.plugin, make_args(test="rccl_perf"))
+        self.assertIn("rccl.mpi_params", output)
+        self.assertIn("no_of_nodes", output)
+
+        sample_path = resolve_sample_path("input/config_file/rccl/rccl_config.json")
+        self.assertIn(sample_path, output)
+        self.assertTrue(os.path.isfile(sample_path), f"printed sample path {sample_path} does not exist")
+
+    def test_documents_every_registered_section(self):
+        output = run_plugin(self.plugin, make_args(test="megatron_llama3_1_8b_single"))
+        self.assertIn("config", output)
+        self.assertIn("model_params", output)
+
+    def test_explains_a_single_parameter(self):
+        output = run_plugin(self.plugin, make_args(test="rccl_perf", parameter="nic_model"))
+        self.assertIn("nic_model", output)
+        self.assertNotIn("no_of_local_ranks", output)
+
+    def test_documents_the_code_default_not_the_sample_value(self):
+        # rccl_config.json ships nic_model "thor"; rccl_lib.py defaults to "ainic".
+        output = run_plugin(self.plugin, make_args(test="rccl_perf", parameter="nic_model"))
+        self.assertIn("ainic", output)
+
+    def test_json_output_is_parseable(self):
+        output = run_plugin(self.plugin, make_args(test="rccl_perf", parameter="nic_model", as_json=True))
+        payload = json.loads(output)
+        self.assertEqual("rccl_perf", payload["test"])
+        self.assertEqual(1, len(payload["parameters"]))
+        self.assertEqual("rccl.cvs_params.nic_model", payload["parameters"][0]["path"])
+
+    def test_json_config_files_are_real_paths(self):
+        output = run_plugin(self.plugin, make_args(test="rccl_perf", as_json=True))
+        payload = json.loads(output)
+        self.assertEqual([resolve_sample_path("input/config_file/rccl/rccl_config.json")], payload["config_files"])
+        for config_file in payload["config_files"]:
+            self.assertTrue(os.path.isfile(config_file), f"{config_file} does not exist")
+
+    def test_unknown_test_exits_nonzero(self):
+        with self.assertRaises(SystemExit) as ctx:
+            run_plugin(self.plugin, make_args(test="no_such_test_anywhere"))
+        self.assertEqual(1, ctx.exception.code)
+
+    def test_undocumented_but_real_test_is_distinguished(self):
+        buf = io.StringIO()
+        with self.assertRaises(SystemExit), redirect_stdout(buf):
+            self.plugin.run(make_args(test="ib_perf_bw_test"))
+        self.assertIn("not documented yet", buf.getvalue())
+
+    def test_unknown_parameter_exits_nonzero(self):
+        with self.assertRaises(SystemExit) as ctx:
+            run_plugin(self.plugin, make_args(test="rccl_perf", parameter="no_such_parameter"))
+        self.assertEqual(1, ctx.exception.code)
+
+    def test_unknown_test_error_goes_to_stderr_in_json_mode(self):
+        # In --json mode, error text must not land on stdout, or piping the
+        # output to a JSON parser would fail on the human-readable message.
+        out, err = io.StringIO(), io.StringIO()
+        with self.assertRaises(SystemExit), redirect_stdout(out), redirect_stderr(err):
+            self.plugin.run(make_args(test="no_such_test_anywhere", as_json=True))
+        self.assertEqual("", out.getvalue())
+        self.assertIn("no config parameter reference", err.getvalue())
+
+    def test_unknown_parameter_error_goes_to_stderr_in_json_mode(self):
+        out, err = io.StringIO(), io.StringIO()
+        with self.assertRaises(SystemExit), redirect_stdout(out), redirect_stderr(err):
+            self.plugin.run(make_args(test="rccl_perf", parameter="no_such_parameter", as_json=True))
+        self.assertEqual("", out.getvalue())
+        self.assertIn("no parameter matching", err.getvalue())
+
+    def test_unrecognized_flag_is_rejected(self):
+        # main() parses with parse_known_args, so stray flags arrive here
+        # silently rather than being caught by argparse.
+        buf = io.StringIO()
+        with self.assertRaises(SystemExit) as ctx, redirect_stdout(buf):
+            self.plugin.run(make_args(test="rccl_perf", extra_pytest_args=["--bogus"]))
+        self.assertEqual(1, ctx.exception.code)
+        self.assertIn("unrecognized arguments", buf.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()

cvs/unittests/test_main.py

@@ -15,7 +15,7 @@ class TestMain(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         """Set up shared test data"""
-        cls.expected_ordered_plugins = ["copy-config", "generate", "list", "run", "scp", "monitor", "exec"]
+        cls.expected_ordered_plugins = ["copy-config", "generate", "list", "man", "run", "scp", "monitor", "exec"]
 
     def test_get_version_success(self):
         """Test successful version retrieval"""