Fix support for deeply nested oneOfs; group nested oneOf options on help pages (#785)

Author: lgarber-akamai

Makefile

@@ -64,7 +64,7 @@ test-unit:
 	@mkdir -p /tmp/linode/.config
 	@orig_xdg_config_home=$${XDG_CONFIG_HOME:-}; \
 	export LINODE_CLI_TEST_MODE=1 XDG_CONFIG_HOME=/tmp/linode/.config; \
-	pytest -v tests/unit; \
+	pytest -vv tests/unit; \
 	exit_code=$$?; \
 	export XDG_CONFIG_HOME=$$orig_xdg_config_home; \
 	exit $$exit_code

linodecli/baked/request.py

@@ -48,6 +48,8 @@ def __init__(  # pylint: disable=too-many-arguments
         :type parent: Optional[str]
         :param depth: The depth of this argument, or how many parent arguments this argument has.
         :type depth: int
+        :param option_variants: A mapping of options, defined using oneOf in the to spec,
+                                to a variant of this argument.
         """
         #: The name of this argument, mostly used for display and docs
         self.name = name
@@ -147,6 +149,7 @@ def _parse_request_model(
     :returns: The flattened request model, as a list
     :rtype: list[OpenAPIRequestArg]
     """
+
     args = []
 
     properties, required = _aggregate_schema_properties(schema)

linodecli/baked/util.py

@@ -24,28 +24,39 @@ def _aggregate_schema_properties(
     properties = {}
     required = defaultdict(lambda: 0)
 
-    def _handle_schema(_schema: Schema):
-        if _schema.properties is None:
+    def __inner(
+        path: List[str],
+        entry: Schema,
+    ):
+        if isinstance(entry, dict):
+            # TODO: Figure out why this happens (openapi3 package bug?)
+            # pylint: disable=protected-access
+            entry = Schema(path, entry, schema._root)
+
+        if entry.properties is None:
+            # If there aren't any properties, this might be a
+            # composite schema
+            for composition_field in ["oneOf", "allOf", "anyOf"]:
+                for i, nested_entry in enumerate(
+                    getattr(entry, composition_field) or []
+                ):
+                    __inner(
+                        schema.path + [composition_field, str(i)],
+                        nested_entry,
+                    )
+
             return
 
+        # This is a valid option
+        properties.update(entry.properties)
+
         nonlocal schema_count
         schema_count += 1
 
-        properties.update(dict(_schema.properties))
-
-        # Aggregate required keys and their number of usages.
-        if _schema.required is not None:
-            for key in _schema.required:
-                required[key] += 1
-
-    _handle_schema(schema)
-
-    one_of = schema.oneOf or []
-    any_of = schema.anyOf or []
+        for key in entry.required or []:
+            required[key] += 1
 
-    for entry in one_of + any_of:
-        # pylint: disable=protected-access
-        _handle_schema(Schema(schema.path, entry, schema._root))
+    __inner(schema.path, schema)
 
     return (
         properties,

linodecli/help_pages.py

@@ -298,50 +298,58 @@ def _help_group_arguments(
     """
     Returns help page groupings for a list of POST/PUT arguments.
     """
+    args = [arg for arg in args if not arg.read_only]
     args_sorted = sorted(args, key=lambda a: a.path)
 
-    groups_tmp = defaultdict(list)
+    paths = {tuple(arg.path.split(".")) for arg in args_sorted}
+    path_to_args = defaultdict(list)
 
-    # Initial grouping by root parent
     for arg in args_sorted:
-        if arg.read_only:
-            continue
+        arg_path = tuple(arg.path.split("."))
+
+        if not arg.is_parent:
+            # Parent arguments are grouped in with their children
+            arg_path = arg_path[:-1]
+
+        # Find first common parent
+        while len(arg_path) > 1 and arg_path not in paths:
+            arg_path = arg_path[:-1]
 
-        groups_tmp[arg.path.split(".", 1)[0]].append(arg)
+        path_to_args[arg_path].append(arg)
 
     group_required = []
     groups = []
     ungrouped = []
 
-    for group in groups_tmp.values():
-        # If the group has more than one element,
-        # leave it as is in the result
-        if len(group) > 1:
+    for k, group in sorted(
+        path_to_args.items(), key=lambda a: (len(a[0]), a[0], len(a[1]))
+    ):
+        if len(k) > 0 and len(group) > 1:
+            # This is a named subgroup
             groups.append(
                 # Args should be ordered by least depth -> required -> path
                 sorted(group, key=lambda v: (v.depth, not v.required, v.path)),
             )
             continue
 
-        target_arg = group[0]
-
         # If the group's argument is required,
-        # add it to the required group
-        if target_arg.required:
-            group_required.append(target_arg)
-            continue
+        # add it to the top-level required group
+        for arg in group:
+            if arg.required:
+                group_required.append(arg)
+                continue
 
-        # Add ungrouped arguments (single value groups) to the
-        # "ungrouped" group.
-        ungrouped.append(target_arg)
+            # Add ungrouped arguments (single value groups) to the
+            # "ungrouped" group.
+            ungrouped.append(arg)
 
     result = []
 
     if len(group_required) > 0:
-        result.append(group_required)
+        result.append(sorted(group_required, key=lambda v: v.path))
 
     if len(ungrouped) > 0:
-        result.append(ungrouped)
+        result.append(sorted(ungrouped, key=lambda v: v.path))
 
     result += groups
 

tests/integration/linodes/fixtures.py

@@ -311,6 +311,60 @@ def linode_with_vpc_interface_as_json(linode_cloud_firewall):
     delete_target_id(target="vpcs", id=vpc_id)
 
 
+@pytest.fixture
+def linode_with_vpc_interface_as_args(linode_cloud_firewall):
+    """
+    NOTE: This is fixture exists to accommodate a regression test.
+          For new tests, use linode_with_vpc_interface_as_json.
+    """
+
+    vpc_json = create_vpc_w_subnet()
+
+    vpc_region = vpc_json["region"]
+    vpc_id = str(vpc_json["id"])
+    subnet_id = str(vpc_json["subnets"][0]["id"])
+
+    linode_json = json.loads(
+        exec_test_command(
+            BASE_CMDS["linodes"]
+            + [
+                "create",
+                "--type",
+                "g6-nanode-1",
+                "--region",
+                vpc_region,
+                "--image",
+                DEFAULT_TEST_IMAGE,
+                "--root_pass",
+                DEFAULT_RANDOM_PASS,
+                "--firewall_id",
+                linode_cloud_firewall,
+                "--interfaces.purpose",
+                "vpc",
+                "--interfaces.primary",
+                "true",
+                "--interfaces.subnet_id",
+                subnet_id,
+                "--interfaces.ipv4.nat_1_1",
+                "any",
+                "--interfaces.ipv4.vpc",
+                "10.0.0.5",
+                "--interfaces.ip_ranges",
+                json.dumps(["10.0.0.6/32"]),
+                "--interfaces.purpose",
+                "public",
+                "--json",
+                "--suppress-warnings",
+            ]
+        )
+    )[0]
+
+    yield linode_json, vpc_json
+
+    delete_target_id(target="linodes", id=str(linode_json["id"]))
+    delete_target_id(target="vpcs", id=vpc_id)
+
+
 # Interfaces new
 @pytest.fixture(scope="module")
 def linode_interface_public(linode_cloud_firewall):

tests/integration/linodes/test_interfaces.py

@@ -6,6 +6,7 @@
     exec_test_command,
 )
 from tests.integration.linodes.fixtures import (  # noqa: F401
+    linode_with_vpc_interface_as_args,
     linode_with_vpc_interface_as_json,
 )
 
@@ -40,5 +41,9 @@ def assert_interface_configuration(
     assert public_interface["purpose"] == "public"
 
 
+def test_with_vpc_interface_as_args(linode_with_vpc_interface_as_args):
+    assert_interface_configuration(*linode_with_vpc_interface_as_args)
+
+
 def test_with_vpc_interface_as_json(linode_with_vpc_interface_as_json):
     assert_interface_configuration(*linode_with_vpc_interface_as_json)