diff --git a/cvs/lib/man_lib.py b/cvs/lib/man_lib.py new file mode 100644 index 000000000..780e19be1 --- /dev/null +++ b/cvs/lib/man_lib.py @@ -0,0 +1,290 @@ +"""Introspection and rendering of test config parameters for ``cvs man``. + +Config parameters are documented on the pydantic models in ``cvs/parsers/``; +this module turns those models into flat, printable parameter references so +the sample config files can stay free of ``_comment_*`` documentation keys. +""" + +import json +import typing + +from pydantic import BaseModel +from pydantic_core import PydanticUndefined + +TYPE_ALIASES = { + "str": "string", + "int": "integer", + "float": "float", + "bool": "boolean", + "NoneType": "null", +} + +# Rendered when a field is a free-form mapping whose keys the user chooses. +OPEN_MAPPING = "mapping" + + +class ParamDoc: + """A single documented config parameter, flattened to a dotted path. + + ``is_section`` marks a field that only exists to group other parameters + (a nested model). Its own default is the whole sub-config, which is noise + in a man page, so renderers show it as a heading rather than a value. + """ + + def __init__(self, path, type_name, default, required, description, examples, constraints, is_section=False): + self.path = path + self.type_name = type_name + self.default = default + self.required = required + self.description = description + self.examples = examples or [] + self.constraints = constraints or [] + self.is_section = is_section + + @property + def section(self): + """Dotted path of the enclosing section, or "" for a top-level parameter.""" + return self.path.rsplit(".", 1)[0] if "." in self.path else "" + + @property + def name(self): + return self.path.rsplit(".", 1)[-1] + + def to_dict(self): + entry = { + "path": self.path, + "type": self.type_name, + "required": self.required, + "description": self.description, + } + if self.is_section: + entry["section"] = True + elif not self.required: + entry["default"] = self.default + if self.examples: + entry["examples"] = self.examples + if self.constraints: + entry["constraints"] = self.constraints + return entry + + +def _unwrap_optional(annotation): + """Return (inner_annotation, is_optional) for Optional[X] / Union[X, None].""" + if typing.get_origin(annotation) is typing.Union: + args = [a for a in typing.get_args(annotation) if a is not type(None)] + if len(args) == 1: + return args[0], True + return annotation, False + + +def _model_of(annotation): + """Return (model, path_suffix, is_section) for the BaseModel behind an annotation. + + A directly nested model is a section: it groups parameters and has no value + of its own worth printing. A model reached through a container is a real + parameter, and ``path_suffix`` records the levels crossed to get there so a + reader can see that ``model_params.single_node...precision`` sits + two mappings deep rather than being a direct child. + """ + inner, _ = _unwrap_optional(annotation) + if isinstance(inner, type) and issubclass(inner, BaseModel): + return inner, "", True + + origin = typing.get_origin(inner) + args = typing.get_args(inner) + + if origin in (dict, typing.Dict) and len(args) == 2: + model, suffix, _ = _model_of(args[1]) + if model is not None: + return model, f".{suffix}", False + if origin in (list, typing.List) and args: + model, suffix, _ = _model_of(args[0]) + if model is not None: + return model, f"[]{suffix}", False + + return None, "", False + + +def _type_name(annotation): + inner, optional = _unwrap_optional(annotation) + origin = typing.get_origin(inner) + + if origin in (list, typing.List): + args = typing.get_args(inner) + rendered = f"list[{_type_name(args[0])}]" if args else "list" + elif origin in (dict, typing.Dict): + args = typing.get_args(inner) + rendered = f"mapping[{_type_name(args[0])} -> {_type_name(args[1])}]" if args else OPEN_MAPPING + elif typing.get_origin(inner) is typing.Literal: + rendered = " | ".join(repr(a) for a in typing.get_args(inner)) + else: + raw = getattr(inner, "__name__", str(inner)) + rendered = TYPE_ALIASES.get(raw, raw) + + return f"{rendered} (optional)" if optional else rendered + + +def _constraints(field): + """Render pydantic constraint metadata (Ge, Le, MinLen, ...) as readable strings.""" + labels = { + "ge": ">=", + "gt": ">", + "le": "<=", + "lt": "<", + "min_length": "min length", + "max_length": "max length", + } + rendered = [] + for meta in field.metadata: + for attr, label in labels.items(): + value = getattr(meta, attr, None) + if value is not None: + rendered.append(f"{label} {value}") + return rendered + + +def _default_of(field): + if field.default is not PydanticUndefined: + return field.default + if field.default_factory is not None: + return field.default_factory() + return None + + +def iter_parameters(model, prefix=""): + """Flatten a pydantic config model into ParamDoc entries, depth first. + + Nested models are recursed into so the caller gets one entry per leaf + parameter, keyed by the dotted path a user would edit in their config file. + """ + params = [] + for name, field in model.model_fields.items(): + # Config keys such as "32_cu_local_read" are not valid Python + # identifiers, so the schema aliases them. Document the on-disk key. + key = field.alias or name + path = f"{prefix}.{key}" if prefix else key + nested, suffix, is_section = _model_of(field.annotation) + + params.append( + ParamDoc( + path=path, + type_name=_type_name(field.annotation), + default=None if is_section else _default_of(field), + required=field.is_required(), + description=field.description or "", + examples=list(field.examples) if field.examples else [], + constraints=_constraints(field), + is_section=is_section, + ) + ) + + if nested is not None: + params.extend(iter_parameters(nested, prefix=f"{path}{suffix}")) + + return params + + +def find_parameters(params, query): + """Match parameters by exact name or path first, falling back to substring. + + An exact match that has documented descendants -- a section, or a field + whose value is itself a nested model, list, or dict of models -- includes + those descendants too, since the parameter's own default is not the whole + answer to "what does configure". + """ + exact = [p for p in params if query in (p.name, p.path)] + if exact: + exact_paths = {p.path for p in exact} + descendants = [ + p + for p in params + if p.path not in exact_paths + and any(p.path.startswith(f"{ep}.") or p.path.startswith(f"{ep}[") for ep in exact_paths) + ] + return exact + descendants + return [p for p in params if query.lower() in p.path.lower()] + + +def _format_value(value): + if value is None: + return "none" + if isinstance(value, bool): + return str(value).lower() + if isinstance(value, (list, dict)): + return json.dumps(value) + return repr(value) if isinstance(value, str) else str(value) + + +def _wrap(text, width, indent): + """Wrap description prose without importing textwrap's paragraph handling.""" + lines = [] + current = "" + for word in text.split(): + candidate = f"{current} {word}".strip() + if len(candidate) > width and current: + lines.append(f"{indent}{current}") + current = word + else: + current = candidate + if current: + lines.append(f"{indent}{current}") + return lines + + +def render_text(params, title=None, summary=None): + """Render a parameter reference in the plain-print house style used by every plugin.""" + lines = [] + if title: + lines.append("") + lines.append(title) + lines.append("=" * 80) + if summary: + lines.append("") + lines.extend(_wrap(summary, 78, "")) + + section_notes = {p.path: p.description for p in params if p.is_section} + leaves = [p for p in params if not p.is_section] + + # Group rather than emit on change: a section's own fields and its nested + # sections interleave in declaration order, which would repeat headings. + grouped = {} + for param in leaves: + grouped.setdefault(param.section, []).append(param) + + for section, members in grouped.items(): + lines.append("") + lines.append(f" {section or '(top level)'}") + lines.append(" " + "-" * 78) + note = section_notes.get(section) + if note: + lines.extend(_wrap(note, 76, " ")) + + for param in members: + marker = "required" if param.required else f"default {_format_value(param.default)}" + lines.append("") + lines.append(f" • {param.name} [{param.type_name}, {marker}]") + + if param.description: + lines.extend(_wrap(param.description, 74, " ")) + else: + lines.append(" (undocumented)") + + if param.constraints: + lines.append(f" constraints: {', '.join(param.constraints)}") + if param.examples: + lines.append(f" example: {', '.join(_format_value(e) for e in param.examples)}") + + lines.append("") + lines.append("=" * 80) + lines.append(f"Total: {len(leaves)} parameter{'' if len(leaves) == 1 else 's'}") + lines.append("") + return "\n".join(lines) + + +def render_json(params, test=None, config_files=None): + payload = {"parameters": [p.to_dict() for p in params]} + if test: + payload["test"] = test + if config_files: + payload["config_files"] = list(config_files) + return json.dumps(payload, indent=2, default=str) diff --git a/cvs/lib/unittests/test_man_lib.py b/cvs/lib/unittests/test_man_lib.py new file mode 100644 index 000000000..2e7a66a52 --- /dev/null +++ b/cvs/lib/unittests/test_man_lib.py @@ -0,0 +1,179 @@ +import unittest +from typing import Dict, List, Optional + +from pydantic import BaseModel, ConfigDict, Field + +from cvs.lib.man_lib import find_parameters, iter_parameters, render_json, render_text + + +class Leaf(BaseModel): + model_config = ConfigDict(extra="allow") + + plain: str = Field(default="hello", description="A plain string.") + bounded: int = Field(default=4, ge=1, le=8, description="A bounded integer.") + needed: str = Field(description="A required value.") + aliased: str = Field(alias="32_cu_local_read", default="1650", description="A key that is not an identifier.") + undescribed: str = Field(default="x") + + +class Section(BaseModel): + model_config = ConfigDict(extra="allow") + + nested: Leaf = Field(default_factory=lambda: Leaf(needed="n"), description="A grouping section.") + listed: List[Leaf] = Field(default_factory=list, description="A list of models.") + mapped: Optional[Dict[str, Dict[str, Leaf]]] = Field(default=None, description="Two mappings deep.") + toggle: bool = Field(default=False, description="A boolean.", examples=[True]) + + +class TestIterParameters(unittest.TestCase): + def setUp(self): + self.params = iter_parameters(Leaf) + self.by_path = {p.path: p for p in self.params} + + def test_reports_type_default_and_description(self): + plain = self.by_path["plain"] + self.assertEqual("string", plain.type_name) + self.assertEqual("hello", plain.default) + self.assertFalse(plain.required) + self.assertEqual("A plain string.", plain.description) + + def test_reports_constraints(self): + self.assertEqual([">= 1", "<= 8"], self.by_path["bounded"].constraints) + + def test_marks_required_fields(self): + self.assertTrue(self.by_path["needed"].required) + + def test_uses_alias_as_the_documented_key(self): + self.assertIn("32_cu_local_read", self.by_path) + self.assertNotIn("aliased", self.by_path) + + def test_missing_description_is_empty_not_absent(self): + self.assertEqual("", self.by_path["undescribed"].description) + + def test_prefix_is_applied(self): + prefixed = iter_parameters(Leaf, prefix="rccl.cvs_params") + self.assertIn("rccl.cvs_params.plain", {p.path for p in prefixed}) + + +class TestNesting(unittest.TestCase): + def setUp(self): + self.params = iter_parameters(Section) + self.paths = {p.path for p in self.params} + + def test_nested_model_is_a_section_and_recursed_into(self): + section = next(p for p in self.params if p.path == "nested") + self.assertTrue(section.is_section) + self.assertIn("nested.plain", self.paths) + + def test_list_of_models_records_the_element_level(self): + self.assertIn("listed[].plain", self.paths) + + def test_dict_of_dict_of_models_records_both_levels(self): + self.assertIn("mapped...plain", self.paths) + + def test_container_fields_are_not_sections(self): + listed = next(p for p in self.params if p.path == "listed") + self.assertFalse(listed.is_section) + + def test_section_default_is_suppressed(self): + section = next(p for p in self.params if p.path == "nested") + self.assertIsNone(section.default) + + +class TestFindParameters(unittest.TestCase): + def setUp(self): + self.params = iter_parameters(Section) + + def test_exact_leaf_name_wins_over_substring(self): + matches = find_parameters(self.params, "plain") + self.assertTrue(matches) + self.assertTrue(all(p.name == "plain" for p in matches)) + + def test_falls_back_to_substring(self): + matches = find_parameters(self.params, "bound") + self.assertTrue(matches) + self.assertTrue(all("bound" in p.path for p in matches)) + + def test_no_match_returns_empty(self): + self.assertEqual([], find_parameters(self.params, "nothing_matches_this")) + + def test_query_matching_a_section_name_resolves_to_its_children(self): + # "nested" is a section (a grouping model with no value of its own), so + # matching it exactly must not return *only* the empty section stub -- + # it should fall through to substring matching and find its leaf + # children too, the same way an unfiltered "cvs man " would. + matches = find_parameters(self.params, "nested") + leaf_matches = [p for p in matches if not p.is_section] + self.assertTrue(leaf_matches, "expected leaf children of the 'nested' section, got only the section stub") + self.assertIn("nested.plain", {p.path for p in leaf_matches}) + + def test_query_matching_a_list_of_models_field_includes_its_elements(self): + # "listed" is not a section (its own default, [], is meaningful), but + # it is a List[Leaf] -- an exact match on it should still pull in the + # documented element fields, not just the empty-list stub. + matches = find_parameters(self.params, "listed") + paths = {p.path for p in matches} + self.assertIn("listed", paths) + self.assertIn("listed[].plain", paths) + + def test_query_matching_a_dict_of_dict_of_models_field_includes_its_leaves(self): + matches = find_parameters(self.params, "mapped") + paths = {p.path for p in matches} + self.assertIn("mapped", paths) + self.assertIn("mapped...plain", paths) + + +class TestRenderText(unittest.TestCase): + def setUp(self): + self.text = render_text(iter_parameters(Section), title="cvs man demo", summary="A demo.") + + def test_includes_title_and_summary(self): + self.assertIn("cvs man demo", self.text) + self.assertIn("A demo.", self.text) + + def test_sections_are_not_repeated(self): + # Declaration order interleaves a section's own fields with its + # children, which previously emitted the same heading twice. + heading = "\n nested\n" + self.assertEqual(1, self.text.count(heading)) + + def test_counts_only_leaf_parameters(self): + leaves = [p for p in iter_parameters(Section) if not p.is_section] + self.assertIn(f"Total: {len(leaves)} parameters", self.text) + + def test_marks_required_and_defaults(self): + self.assertIn("required", self.text) + self.assertIn("default", self.text) + + def test_flags_undocumented_parameters(self): + self.assertIn("(undocumented)", self.text) + + def test_singular_parameter_count(self): + one = [p for p in iter_parameters(Leaf) if p.path == "plain"] + self.assertIn("Total: 1 parameter\n", render_text(one)) + + +class TestRenderJson(unittest.TestCase): + def test_emits_parseable_payload(self): + import json + + payload = json.loads(render_json(iter_parameters(Leaf), test="demo", config_files=["a.json"])) + self.assertEqual("demo", payload["test"]) + self.assertEqual(["a.json"], payload["config_files"]) + + by_path = {entry["path"]: entry for entry in payload["parameters"]} + self.assertEqual("hello", by_path["plain"]["default"]) + self.assertTrue(by_path["needed"]["required"]) + self.assertNotIn("default", by_path["needed"]) + + def test_sections_are_flagged_and_carry_no_default(self): + import json + + payload = json.loads(render_json(iter_parameters(Section))) + nested = next(e for e in payload["parameters"] if e["path"] == "nested") + self.assertTrue(nested["section"]) + self.assertNotIn("default", nested) + + +if __name__ == "__main__": + unittest.main()