T8989: vyos_interfaces dict_op refactor (#33)HEADmain

* T8989: vyos_interfaces dict_op refactor


---------

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Daniil Baturin <daniil@vyos.io>

1 files changed, 324 insertions, 115 deletions

diff --git a/plugins/modules/vyos_interfaces.py b/plugins/modules/vyos_interfaces.py
index 6ed45d5..f5733ac 100644
--- a/plugins/modules/vyos_interfaces.py
+++ b/plugins/modules/vyos_interfaces.py
@@ -12,9 +12,17 @@ DOCUMENTATION = r"""
 module: vyos_interfaces
 short_description: Manage interface configuration on VyOS devices via REST API.
 description:
-  - Manages L2 interface configuration (description, MTU, speed, duplex, enabled)
-    on VyOS devices using the HTTPS REST API.
+  - Manages L2 interface configuration (description, MTU, speed, duplex,
+    enabled, VRF assignment, VLAN sub-interfaces) on VyOS devices using the
+    HTTPS REST API.
   - IP address configuration is handled by M(vyos.rest.vyos_l3_interfaces).
+  - >-
+    Covers 11 interface types (ethernet, bonding, loopback, tunnel,
+    wireguard, vti, dummy, openvpn, pppoe, wireless, bridge), resolved
+    from the device response when present, otherwise guessed from the
+    interface name. The current CLI collection module documents a
+    narrower scope of 5 types (ethernet, bonding, vxlan, loopback,
+    vti).
 version_added: "1.0.0"
 author:
   - VyOS Community (@vyos)
@@ -46,6 +54,28 @@ options:
         description: Interface speed setting.
         type: str
         choices: [auto, "10", "100", "1000", "2500", "10000"]
+      vrf:
+        description: VRF instance to bind this interface to.
+        type: str
+      vifs:
+        description: 802.1Q VLAN sub-interfaces.
+        type: list
+        elements: dict
+        suboptions:
+          vlan_id:
+            description: VLAN ID for this sub-interface.
+            type: int
+            required: true
+          description:
+            description: Sub-interface description.
+            type: str
+          enabled:
+            description: Whether the sub-interface is enabled.
+            type: bool
+            default: true
+          mtu:
+            description: Sub-interface MTU.
+            type: int
   state:
     description:
       - C(merged) - Merge config with existing interface settings.
@@ -69,6 +99,10 @@ EXAMPLES = r"""
         description: Management interface
         mtu: 1500
         enabled: true
+        vrf: mgmt
+        vifs:
+          - vlan_id: 200
+            description: VIF 200
     state: merged
 
 - name: Disable an interface
@@ -78,7 +112,7 @@ EXAMPLES = r"""
         enabled: false
     state: merged
 
-- name: Delete interface description
+- name: Delete interface config
   vyos.rest.vyos_interfaces:
     config:
       - name: eth0
@@ -117,11 +151,18 @@ response:
 """
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.vyos.rest.plugins.module_utils.vyos import VyOSModule
+from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
+    VyOSModule,
+    cast_by_spec,
+    dict_op,
+    to_tag_dict,
+)
+
 
+_BASE = ["interfaces"]
 
-# Interface name prefix → API type key
-_IFACE_TYPE = {
+
+_IFACE_TYPE_PREFIX = {
     "eth": "ethernet",
     "bond": "bonding",
     "lo": "loopback",
@@ -135,146 +176,293 @@ _IFACE_TYPE = {
     "br": "bridge",
 }
 
-# L2 fields managed by this module — excludes address, hw-id etc.
-_L2_FIELDS = ["description", "mtu", "duplex", "speed"]
-
 
-def _iface_type(name):
-    for prefix, itype in _IFACE_TYPE.items():
+def _guess_iface_type(name):
+    for prefix, itype in _IFACE_TYPE_PREFIX.items():
         if name.startswith(prefix):
             return itype
     return "ethernet"
 
 
-def _iface_base(name):
-    return ["interfaces", _iface_type(name), name]
+def _resolve_iface_type(name, raw_have):
+    """Prefer the real type from the device's own raw response
+    (organized by type at the top level) over a name-prefix guess --
+    only fall back to guessing for a brand-new interface that doesn't
+    exist on the device yet.
+    """
+    for itype, ifaces in (raw_have or {}).items():
+        if name in to_tag_dict(ifaces):
+            return itype
+    return _guess_iface_type(name)
 
 
-def get_running_config(vyos):
-    raw = vyos.get_config(["interfaces"])
-    if not raw or not isinstance(raw, dict):
-        return []
+def _iface_base(name, raw_have):
+    return _BASE + [_resolve_iface_type(name, raw_have), name]
 
-    result = []
-    for itype, ifaces in sorted(raw.items()):
-        if not isinstance(ifaces, dict):
-            continue
-        for iname, idata in sorted(ifaces.items()):
-            idata = idata or {}
-            entry = {"name": iname}
-            if idata.get("description"):
-                entry["description"] = idata["description"]
-            if "mtu" in idata:
-                entry["mtu"] = int(idata["mtu"])
-            if "duplex" in idata:
-                entry["duplex"] = idata["duplex"]
-            if "speed" in idata:
-                entry["speed"] = idata["speed"]
-            entry["enabled"] = "disable" not in idata
-            result.append(entry)
 
+_DEVICE_RENAMES = {
+    "vifs": "vif",
+}
+
+
+_ENABLED_FIELD = "enabled"
+_DISABLE_DEVICE_KEY = "disable"
+
+
+def _derive_key_field(options_spec):
+    """The field identifying each entry in a keyed-list section is
+    never inferable from a generic walk alone -- but it doesn't need
+    to be hand-declared either: every such section in this argspec
+    already marks exactly one suboption required=True (you can't
+    create a VIF without a vlan_id). Deriving it here means the key
+    field is asserted to exist by the argspec itself, not duplicated
+    in a place that could drift out of sync with it.
+    """
+    required = [k for k, spec in options_spec.items() if spec.get("required")]
+    if len(required) != 1:
+        raise ValueError(
+            "expected exactly one required suboption to serve as the key field, "
+            "found: {0}".format(required),
+        )
+    return required[0]
+
+
+def _keyed_list_to_device(items, key_field, entry_transform):
+    result = {}
+    for item in items or []:
+        if item.get(key_field) is None:
+            continue
+        rest = {k: v for k, v in item.items() if k != key_field}
+        result[str(item[key_field])] = entry_transform(rest)
     return result
 
 
-def _normalize(config):
-    """Convert argspec list to dict keyed by interface name."""
-    return {entry["name"]: entry for entry in (config or [])}
+def _keyed_list_from_device(raw, key_field, entry_transform, key_cast=None):
+    key_cast = key_cast or (lambda k: k)
+    return [
+        {key_field: key_cast(key), **entry_transform(data or {})}
+        for key, data in sorted(to_tag_dict(raw).items())
+    ]
 
 
-def _iface_cmds(name, want, have):
-    """Generate set/delete commands to bring have → want for one interface."""
-    cmds = []
-    base = _iface_base(name)
-    have = have or {}
+def _spec_to_device(value, options_spec):
+    if not isinstance(value, dict):
+        return value
+    result = {}
+    for arg_key, sub_spec in options_spec.items():
+        if arg_key == _ENABLED_FIELD:
+            if value.get(arg_key) is False:
+                result[_DISABLE_DEVICE_KEY] = {}
+            continue
 
-    # description
-    want_desc = want.get("description")
-    have_desc = have.get("description")
-    if want_desc is not None and want_desc != have_desc:
-        cmds.append(("set", base + ["description", want_desc]))
-    elif want_desc is None and have_desc is not None:
-        cmds.append(("delete", base + ["description"]))
+        val = value.get(arg_key)
+        if val is None or val is False:
+            continue
 
-    # mtu
-    want_mtu = want.get("mtu")
-    have_mtu = have.get("mtu")
-    if want_mtu is not None and want_mtu != have_mtu:
-        cmds.append(("set", base + ["mtu", str(want_mtu)]))
+        device_key = _DEVICE_RENAMES.get(arg_key, arg_key.replace("_", "-"))
+        sub_type = sub_spec.get("type")
+        sub_options = sub_spec.get("options")
+
+        if sub_type == "dict" and sub_options:
+            converted = _spec_to_device(val, sub_options)
+            if converted:
+                result[device_key] = converted
+        elif sub_type == "list" and sub_options:
+            key_field = _derive_key_field(sub_options)
+            result[device_key] = _keyed_list_to_device(
+                val,
+                key_field,
+                lambda rest, spec=sub_options: _spec_to_device(rest, spec),
+            )
+        elif val is True:
+            result[device_key] = {}
+        elif sub_type == "list":
+            result[device_key] = list(val)
+        else:
+            result[device_key] = val
+    return result
+
+
+def _device_to_spec(raw, options_spec):
+    if not raw or not isinstance(raw, dict):
+        return {}
+    have_idx = {k.replace("-", "_"): k for k in raw}
+    result = {}
+
+    if _ENABLED_FIELD in options_spec and _DISABLE_DEVICE_KEY in raw:
+        result[_ENABLED_FIELD] = False
 
-    # duplex
-    want_duplex = want.get("duplex")
-    have_duplex = have.get("duplex")
-    if want_duplex is not None and want_duplex != have_duplex:
-        cmds.append(("set", base + ["duplex", want_duplex]))
+    for arg_key, sub_spec in options_spec.items():
+        if arg_key == _ENABLED_FIELD:
+            continue
+        device_key = _DEVICE_RENAMES.get(arg_key, arg_key.replace("_", "-"))
+        orig_key = device_key if device_key in raw else have_idx.get(arg_key)
+        if orig_key is None:
+            continue
+        raw_val = raw[orig_key]
+        sub_type = sub_spec.get("type")
+        sub_options = sub_spec.get("options")
+
+        if sub_type == "dict" and sub_options:
+            converted = _device_to_spec(raw_val, sub_options)
+            if converted:
+                result[arg_key] = converted
+        elif sub_type == "list" and sub_options:
+            key_field = _derive_key_field(sub_options)
+            key_cast = int if sub_options[key_field].get("type") == "int" else None
+            entries = _keyed_list_from_device(
+                raw_val,
+                key_field,
+                lambda d, spec=sub_options: _device_to_spec(d, spec),
+                key_cast=key_cast,
+            )
+            if entries:
+                result[arg_key] = entries
+        elif sub_type == "list":
+            if raw_val:
+                result[arg_key] = sorted(to_tag_dict(raw_val).keys())
+        elif isinstance(raw_val, dict) and not raw_val:
+            result[arg_key] = True
+        else:
+            result[arg_key] = raw_val
+    return result
 
-    # speed
-    want_speed = want.get("speed")
-    have_speed = have.get("speed")
-    if want_speed is not None and want_speed != have_speed:
-        cmds.append(("set", base + ["speed", want_speed]))
 
-    # enabled / disable flag
-    want_enabled = want.get("enabled", True)
-    have_enabled = have.get("enabled", True)
-    if not want_enabled and have_enabled:
-        cmds.append(("set", base + ["disable"]))
-    elif want_enabled and not have_enabled:
-        cmds.append(("delete", base + ["disable"]))
+def get_running_config(vyos):
+    """VyOS's REST API collapses a single-child tag node to a plain
+    string (or a list for multiple) -- normalizing through to_tag_dict
+    unconditionally means callers always receive a genuine dict.
+    """
+    return to_tag_dict(vyos.get_config(_BASE) or {})
 
-    return cmds
 
+def _device_to_argspec(raw):
+    result = []
+    for itype, ifaces in sorted((raw or {}).items()):
+        for name, data in sorted(to_tag_dict(ifaces).items()):
+            entry = {"name": name}
+            entry.update(_device_to_spec(data or {}, _ENTRY_OPTIONS))
+            result.append(entry)
+    return result
 
-def _delete_iface_config(name, have):
-    """Generate delete commands to remove L2 config from an interface."""
-    cmds = []
-    base = _iface_base(name)
-    have = have or {}
 
-    for field in _L2_FIELDS:
-        if field in have:
-            cmds.append(("delete", base + [field]))
-    if not have.get("enabled", True):
-        cmds.append(("delete", base + ["disable"]))
+def _shadow_vif_entries(want_device, have_device):
+    """Ensure want_device has a (possibly empty) placeholder for every
+    VLAN ID present in have_device's own "vif" dict, so a dict_op
+    purge recurses into each VIF individually rather than treating the
+    whole "vif" key, or any single VLAN entry, as one unit.
+    """
+    have_vifs = have_device.get("vif")
+    if not have_vifs:
+        return want_device
+    shadowed = dict(want_device)
+    want_vifs = dict(shadowed.get("vif") or {})
+    for vlan_id in have_vifs:
+        want_vifs.setdefault(vlan_id, {})
+    shadowed["vif"] = want_vifs
+    return shadowed
+
+
+def _purge_commands(want_device, have_device, base):
+    return dict_op(_shadow_vif_entries(want_device, have_device), have_device, base, op="purge")
+
+
+def _entry_to_device(entry, options_spec):
+    """to-device conversion for a keyed entry whose own key field
+    (e.g. "name" for an interface, same role "vlan_id" plays for a
+    VIF) is present in the input but must never be treated as a
+    regular child leaf -- it identifies the entry itself and is
+    already expressed in the API path (_iface_base), not a field to
+    set/purge under it. _keyed_list_to_device already strips a VIF's
+    "vlan_id" the same way before conversion; interface entries need
+    the same treatment here since build_commands handles the top
+    level manually rather than through that helper.
+    """
+    key_field = _derive_key_field(options_spec)
+    rest = {k: v for k, v in (entry or {}).items() if k != key_field}
+    return _spec_to_device(rest, options_spec)
+
+
+def _scoped_purge_commands(name, have_entry, raw_have):
+    """Remove every field this module manages for one interface --
+    scoped to this module's own fields only, never a whole-subtree
+    delete for the VIF container shared with vyos_l3_interfaces.
+    """
+    have_device = _entry_to_device(have_entry, _ENTRY_OPTIONS)
+    base = _iface_base(name, raw_have)
+    return _purge_commands({}, have_device, base)
+
+
+def _enabled_leaves(device, allowed_vlan_ids=None):
+    """allowed_vlan_ids restricts the VIF portion to VLAN IDs the task
+    actually listed. The interface itself needs no such restriction --
+    it's always "listed" by virtue of appearing in want at all -- but
+    an unlisted VIF was never mentioned by the task, and merged must
+    never touch it. Confirmed real bug otherwise: a VIF the task
+    doesn't reference at all (or a listed interface whose vifs simply
+    omits it) would still have its stale "disable" leaf removed,
+    silently re-enabling a VLAN the administrator deliberately
+    disabled. Pass None (the want side, where every VIF present
+    already is one the task listed) to skip this restriction.
+    """
+    result = {}
+    if _DISABLE_DEVICE_KEY in device:
+        result[_DISABLE_DEVICE_KEY] = device[_DISABLE_DEVICE_KEY]
+    vif = device.get("vif")
+    if vif:
+        vif_result = {
+            vlan_id: {_DISABLE_DEVICE_KEY: v[_DISABLE_DEVICE_KEY]}
+            for vlan_id, v in vif.items()
+            if _DISABLE_DEVICE_KEY in v
+            and (allowed_vlan_ids is None or vlan_id in allowed_vlan_ids)
+        }
+        if vif_result:
+            result["vif"] = vif_result
+    return result
 
-    return cmds
 
+def build_commands(config, raw_have, state):
+    raw_have = raw_have or {}
+    config = config or []
 
-def build_commands(config, have_raw, state):
-    cmds = []
-    have_map = _normalize(have_raw)
+    have_list = _device_to_argspec(raw_have)
+    have_by_name = {e["name"]: e for e in have_list}
+    want_by_name = {e["name"]: e for e in config if e.get("name")}
 
     if state == "deleted":
-        if not config:
-            for name, have in have_map.items():
-                cmds += _delete_iface_config(name, have)
-        else:
-            for entry in config:
-                name = entry["name"]
-                cmds += _delete_iface_config(name, have_map.get(name, {}))
+        cmds = []
+        targets = (
+            have_by_name
+            if not config
+            else {n: have_by_name[n] for n in want_by_name if n in have_by_name}
+        )
+        for name, have_entry in targets.items():
+            cmds += _scoped_purge_commands(name, have_entry, raw_have)
         return cmds
 
-    want_map = _normalize(config)
-
+    commands = []
     if state == "overridden":
-        # delete L2 config from interfaces not in want
-        for name in set(have_map) - set(want_map):
-            cmds += _delete_iface_config(name, have_map[name])
-
-    for name, want in want_map.items():
-        have = have_map.get(name, {})
+        for name in set(have_by_name) - set(want_by_name):
+            commands += _scoped_purge_commands(name, have_by_name[name], raw_have)
 
-        if state == "replaced":
-            # pre-check — only act if something differs
-            test_cmds = _iface_cmds(name, want, have)
-            if not test_cmds:
-                continue
-            # delete L2 fields then rebuild
-            cmds += _delete_iface_config(name, have)
-            have = {}
+    for name, want_entry in want_by_name.items():
+        have_entry = have_by_name.get(name, {})
+        want_device = _entry_to_device(want_entry, _ENTRY_OPTIONS)
+        have_device = _entry_to_device(have_entry, _ENTRY_OPTIONS)
+        base = _iface_base(name, raw_have)
 
-        cmds += _iface_cmds(name, want, have if state != "replaced" else {})
+        if state in ("replaced", "overridden"):
+            commands += _purge_commands(want_device, have_device, base)
+        else:
+            want_enabled = _enabled_leaves(want_device)
+            have_enabled = _enabled_leaves(
+                have_device,
+                allowed_vlan_ids=(want_device.get("vif") or {}).keys(),
+            )
+            commands += _purge_commands(want_enabled, have_enabled, base)
+        commands += dict_op(want_device, have_device, base, op="set")
 
-    return cmds
+    return commands
 
 
 ARGUMENT_SPEC = dict(
@@ -288,6 +476,17 @@ ARGUMENT_SPEC = dict(
             mtu=dict(type="int"),
             duplex=dict(type="str", choices=["auto", "full", "half"]),
             speed=dict(type="str", choices=["auto", "10", "100", "1000", "2500", "10000"]),
+            vrf=dict(type="str"),
+            vifs=dict(
+                type="list",
+                elements="dict",
+                options=dict(
+                    vlan_id=dict(type="int", required=True),
+                    description=dict(type="str"),
+                    enabled=dict(type="bool", default=True),
+                    mtu=dict(type="int"),
+                ),
+            ),
         ),
     ),
     state=dict(
@@ -297,6 +496,9 @@ ARGUMENT_SPEC = dict(
     ),
 )
 
+_ENTRY_OPTIONS = ARGUMENT_SPEC["config"]["options"]
+_VIF_OPTIONS = _ENTRY_OPTIONS["vifs"]["options"]
+
 
 def main():
     module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
@@ -305,23 +507,30 @@ def main():
     state = module.params["state"]
     config = module.params.get("config") or []
 
-    have = get_running_config(vyos)
+    raw_have = get_running_config(vyos)
+    have = _device_to_argspec(raw_have)
+    for entry in have:
+        cast_by_spec(entry, _ENTRY_OPTIONS)
 
     if state == "gathered":
         module.exit_json(changed=False, gathered=have)
 
-    commands = build_commands(config, have, state)
+    commands = build_commands(config, raw_have, state)
 
     if module.check_mode:
-        module.exit_json(changed=bool(commands), commands=commands, before=have)
+        module.exit_json(changed=bool(commands), commands=commands, before=have, after=have)
 
     if commands:
         response = vyos.apply_commands(commands)
         saved = vyos.save_config()
+        after_raw = get_running_config(vyos)
+        after = _device_to_argspec(after_raw)
+        for entry in after:
+            cast_by_spec(entry, _ENTRY_OPTIONS)
         module.exit_json(
             changed=True,
             before=have,
-            after=get_running_config(vyos),
+            after=after,
             commands=commands,
             saved=saved,
             response=response,