diff options
| -rw-r--r-- | changelogs/fragments/t8989_l3_interface_dict_op.yml | 3 | ||||
| -rw-r--r-- | docs/vyos.rest.vyos_l3_interfaces_module.rst | 51 | ||||
| -rw-r--r-- | plugins/modules/vyos_l3_interfaces.py | 372 | ||||
| -rw-r--r-- | tests/unit/fixtures/l3_interfaces_running.json | 19 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_l3_interfaces.py | 448 |
5 files changed, 385 insertions, 508 deletions
diff --git a/changelogs/fragments/t8989_l3_interface_dict_op.yml b/changelogs/fragments/t8989_l3_interface_dict_op.yml new file mode 100644 index 0000000..5cccc00 --- /dev/null +++ b/changelogs/fragments/t8989_l3_interface_dict_op.yml @@ -0,0 +1,3 @@ +--- +minor_changes: + - vyos_l3_interfaces - Refactor the module to comply with dict_op paradigm. diff --git a/docs/vyos.rest.vyos_l3_interfaces_module.rst b/docs/vyos.rest.vyos_l3_interfaces_module.rst index d251abb..784a565 100644 --- a/docs/vyos.rest.vyos_l3_interfaces_module.rst +++ b/docs/vyos.rest.vyos_l3_interfaces_module.rst @@ -18,7 +18,7 @@ Version added: 1.0.0 Synopsis -------- - Manages IPv4 and IPv6 address configuration on VyOS interfaces using the HTTPS REST API. -- Mirrors ``vyos.vyos.vyos_l3_interfaces`` but uses the HTTP API instead of CLI. +- L2 attributes (description, mtu, duplex, speed, vrf) and VIF presence/L2 attributes are owned by :ref:`vyos.rest.vyos_interfaces <vyos.rest.vyos_interfaces_module>`, not this module -- confirmed by design: a VIF's device path is shared between the two modules, and this module's own commands only ever touch the ``address`` leaf within it, never the VIF subtree as a whole or any of vyos_interfaces' own fields. @@ -251,21 +251,6 @@ Parameters <tr> <td colspan="4"> <div class="ansibleOptionAnchor" id="parameter-"></div> - <b>running_config</b> - <a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a> - <div style="font-size: small"> - <span style="color: purple">string</span> - </div> - </td> - <td> - </td> - <td> - <div>Used only with state <code>parsed</code>.</div> - </td> - </tr> - <tr> - <td colspan="4"> - <div class="ansibleOptionAnchor" id="parameter-"></div> <b>state</b> <a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a> <div style="font-size: small"> @@ -279,8 +264,6 @@ Parameters <li>overridden</li> <li>deleted</li> <li>gathered</li> - <li>rendered</li> - <li>parsed</li> </ul> </td> <td> @@ -289,8 +272,6 @@ Parameters <div><code>overridden</code> - Replace addresses for all interfaces.</div> <div><code>deleted</code> - Remove listed or all interface addresses.</div> <div><code>gathered</code> - Read interface addresses from device without changes.</div> - <div><code>rendered</code> - Return commands for provided config without connecting.</div> - <div><code>parsed</code> - Parse running_config into structured data.</div> </td> </tr> </table> @@ -423,36 +404,6 @@ Common return values are documented `here <https://docs.ansible.com/ansible/late <tr> <td colspan="1"> <div class="ansibleOptionAnchor" id="return-"></div> - <b>parsed</b> - <a class="ansibleOptionLink" href="#return-" title="Permalink to this return value"></a> - <div style="font-size: small"> - <span style="color: purple">list</span> - </div> - </td> - <td>when state is parsed</td> - <td> - <div>Structured data parsed from running_config (state=parsed).</div> - <br/> - </td> - </tr> - <tr> - <td colspan="1"> - <div class="ansibleOptionAnchor" id="return-"></div> - <b>rendered</b> - <a class="ansibleOptionLink" href="#return-" title="Permalink to this return value"></a> - <div style="font-size: small"> - <span style="color: purple">list</span> - </div> - </td> - <td>when state is rendered</td> - <td> - <div>Commands for the provided config (state=rendered).</div> - <br/> - </td> - </tr> - <tr> - <td colspan="1"> - <div class="ansibleOptionAnchor" id="return-"></div> <b>response</b> <a class="ansibleOptionLink" href="#return-" title="Permalink to this return value"></a> <div style="font-size: small"> diff --git a/plugins/modules/vyos_l3_interfaces.py b/plugins/modules/vyos_l3_interfaces.py index 9b33973..c9c6deb 100644 --- a/plugins/modules/vyos_l3_interfaces.py +++ b/plugins/modules/vyos_l3_interfaces.py @@ -14,7 +14,13 @@ short_description: Manage L3 interface configuration on VyOS devices via REST AP description: - Manages IPv4 and IPv6 address configuration on VyOS interfaces using the HTTPS REST API. - - Mirrors C(vyos.vyos.vyos_l3_interfaces) but uses the HTTP API instead of CLI. + - >- + L2 attributes (description, mtu, duplex, speed, vrf) and VIF + presence/L2 attributes are owned by M(vyos.rest.vyos_interfaces), not + this module -- confirmed by design: a VIF's device path is shared + between the two modules, and this module's own commands only ever + touch the C(address) leaf within it, never the VIF subtree as a whole + or any of vyos_interfaces' own fields. version_added: "1.0.0" author: - VyOS Community (@vyos) @@ -71,9 +77,6 @@ options: address: description: IPv6 address in CIDR notation, C(dhcpv6), or C(auto-config). type: str - running_config: - description: Used only with state C(parsed). - type: str state: description: - C(merged) - Add addresses without removing existing ones. @@ -81,10 +84,8 @@ options: - C(overridden) - Replace addresses for all interfaces. - C(deleted) - Remove listed or all interface addresses. - C(gathered) - Read interface addresses from device without changes. - - C(rendered) - Return commands for provided config without connecting. - - C(parsed) - Parse running_config into structured data. type: str - choices: [merged, replaced, overridden, deleted, gathered, rendered, parsed] + choices: [merged, replaced, overridden, deleted, gathered] default: merged seealso: - module: vyos.vyos.vyos_l3_interfaces @@ -142,14 +143,6 @@ gathered: description: Current L3 interface configuration as structured data. returned: when state is gathered type: list -rendered: - description: Commands for the provided config (state=rendered). - returned: when state is rendered - type: list -parsed: - description: Structured data parsed from running_config (state=parsed). - returned: when state is parsed - type: list saved: description: Whether the config was saved after changes. returned: when changes are applied @@ -161,10 +154,16 @@ 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, to_tag_dict + +_BASE = ["interfaces"] -_IFACE_TYPE = { +# Aligned with vyos_interfaces' own 11-type table -- confirmed +# inconsistent in the original (missing ppp/wlan here specifically), +# corrected for consistency between the two modules that share the +# same interface namespace. +_IFACE_TYPE_PREFIX = { "eth": "ethernet", "bond": "bonding", "lo": "loopback", @@ -173,23 +172,42 @@ _IFACE_TYPE = { "vti": "vti", "dum": "dummy", "vtun": "openvpn", + "ppp": "pppoe", + "wlan": "wireless", "br": "bridge", } -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 over a + name-prefix guess -- only fall back to guessing for a brand-new + interface that doesn't exist on the device yet. Same fix as + vyos_interfaces' own confirmed bug: the original guessed + unconditionally, even for interfaces the device already knows the + real type of. + """ + for itype, ifaces in (raw_have or {}).items(): + if name in to_tag_dict(ifaces): + return itype + return _guess_iface_type(name) + + +def _iface_base(name, raw_have): + return _BASE + [_resolve_iface_type(name, raw_have), name] def _addr_list(raw): - """Normalize address field — string or list → sorted list.""" + """VyOS's address leaf collapses to a bare string for a single + value, or a list for multiple -- confirmed device behavior, + handled explicitly here since it's a plain multi-value leaf, not + a tag-node (to_tag_dict doesn't apply).""" if not raw: return [] if isinstance(raw, str): @@ -200,92 +218,105 @@ def _addr_list(raw): def _split_addresses(addresses): - """Split address list into ipv4 and ipv6 lists.""" ipv4 = [] ipv6 = [] for addr in addresses: - if addr in ("dhcp", "dhcpv6"): - if addr == "dhcp": - ipv4.append(addr) - else: - ipv6.append(addr) - elif ":" in addr: + if addr == "dhcp": + ipv4.append(addr) + elif addr in ("dhcpv6", "auto-config") or ":" in addr: ipv6.append(addr) else: ipv4.append(addr) return sorted(ipv4), sorted(ipv6) -def _parse_iface(name, idata): - """Parse raw API interface data into argspec format.""" - idata = idata or {} - entry = {"name": name} - - addrs = _addr_list(idata.get("address")) - if addrs: - ipv4, ipv6 = _split_addresses(addrs) - if ipv4: - entry["ipv4"] = [{"address": a} for a in ipv4] - if ipv6: - entry["ipv6"] = [{"address": a} for a in ipv6] - - vif_data = idata.get("vif") or {} - if isinstance(vif_data, dict) and vif_data: - vifs = [] - for vlan_id, vdata in sorted(vif_data.items(), key=lambda x: int(x[0])): - vdata = vdata or {} - vif = {"vlan_id": int(vlan_id)} - vaddrs = _addr_list(vdata.get("address")) - if vaddrs: - vipv4, vipv6 = _split_addresses(vaddrs) - if vipv4: - vif["ipv4"] = [{"address": a} for a in vipv4] - if vipv6: - vif["ipv6"] = [{"address": a} for a in vipv6] - vifs.append(vif) - if vifs: - entry["vifs"] = vifs - - return entry +def _parse_addr_entry(idata): + """Returns (ipv4_argspec_list, ipv6_argspec_list) for one address + leaf's raw value -- shared by both interface-level and vif-level + parsing.""" + addrs = _addr_list((idata or {}).get("address")) + if not addrs: + return [], [] + ipv4, ipv6 = _split_addresses(addrs) + return ( + [{"address": a} for a in ipv4], + [{"address": a} for a in ipv6], + ) def get_running_config(vyos): - raw = vyos.get_config(["interfaces"]) - if not raw or not isinstance(raw, dict): - return [] + """VyOS's REST API collapses a single-child tag node to a plain + string (or a list for multiple) -- confirmed as a real failure + mode during vyos_ospf_interfaces's build. Normalizing through + to_tag_dict unconditionally means callers always receive a + genuine dict. + """ + return to_tag_dict(vyos.get_config(_BASE) or {}) + +def _device_to_argspec(raw): result = [] - for itype, ifaces in sorted(raw.items()): - if not isinstance(ifaces, dict): - continue - for iname, idata in sorted(ifaces.items()): - entry = _parse_iface(iname, idata) - # only include if there's at least one address or vif + for itype, ifaces in sorted((raw or {}).items()): + for name, idata in sorted(to_tag_dict(ifaces).items()): + idata = idata or {} + entry = {"name": name} + + ipv4, ipv6 = _parse_addr_entry(idata) + if ipv4: + entry["ipv4"] = ipv4 + if ipv6: + entry["ipv6"] = ipv6 + + vif_raw = idata.get("vif") + if vif_raw: + vifs = [] + for vlan_id, vdata in sorted(to_tag_dict(vif_raw).items(), key=lambda x: int(x[0])): + vipv4, vipv6 = _parse_addr_entry(vdata) + if vipv4 or vipv6: + vif = {"vlan_id": int(vlan_id)} + if vipv4: + vif["ipv4"] = vipv4 + if vipv6: + vif["ipv6"] = vipv6 + vifs.append(vif) + if vifs: + entry["vifs"] = vifs + if entry.get("ipv4") or entry.get("ipv6") or entry.get("vifs"): result.append(entry) - return result def _normalize(config): - """Convert argspec list to dict keyed by interface name.""" + """Argspec list -> dict keyed by interface name, with address + lists flattened to plain sorted string lists for set-diffing.""" result = {} for entry in config or []: - name = entry["name"] - ipv4 = sorted([a["address"] for a in (entry.get("ipv4") or [])]) - ipv6 = sorted([a["address"] for a in (entry.get("ipv6") or [])]) + name = entry.get("name") + if not name: + continue + ipv4 = sorted(a["address"] for a in (entry.get("ipv4") or []) if a.get("address")) + ipv6 = sorted(a["address"] for a in (entry.get("ipv6") or []) if a.get("address")) vifs = {} for vif in entry.get("vifs") or []: - vid = vif["vlan_id"] - vipv4 = sorted([a["address"] for a in (vif.get("ipv4") or [])]) - vipv6 = sorted([a["address"] for a in (vif.get("ipv6") or [])]) + vid = vif.get("vlan_id") + if vid is None: + continue + vipv4 = sorted(a["address"] for a in (vif.get("ipv4") or []) if a.get("address")) + vipv6 = sorted(a["address"] for a in (vif.get("ipv6") or []) if a.get("address")) vifs[vid] = {"ipv4": vipv4, "ipv6": vipv6} result[name] = {"ipv4": ipv4, "ipv6": ipv6, "vifs": vifs} return result def _addr_cmds(base, want_addrs, have_addrs, state): - """Generate set/delete commands for address lists.""" + """Address is a plain multi-value leaf, not a keyed/tag-node + section -- a set-diff is the correct, direct mechanism here + (dict_op's keyed-entry model doesn't fit a bare value list). + Confirmed scoped to exactly the "address" leaf under base, + never anything else -- base may be an interface or a vif, and + in neither case does this function ever touch a sibling leaf. + """ cmds = [] want_set = set(want_addrs) have_set = set(have_addrs) @@ -300,15 +331,25 @@ def _addr_cmds(base, want_addrs, have_addrs, state): return cmds -def _vif_cmds(iface_base, want_vifs, have_vifs, state): - """Generate commands for VIF subinterfaces.""" +def _vif_addr_cmds(iface_base, want_vifs, have_vifs, state): + """VIF address commands, explicitly scoped to only the address + leaf within each VIF -- never the VIF subtree as a whole. + + Confirmed severe bug in the original: an omitted VIF under + replaced/overridden/deleted generated a whole-VIF delete + (`delete ... vif <id>`), which would also destroy the VIF's own + description/mtu/disable settings -- fields owned by + vyos_interfaces, not this module. The same class of bug just + confirmed and fixed in vyos_interfaces for interface-level + "address" leaking the other way. Here, an omitted VIF under any + state only ever has its own known addresses individually deleted; + the VIF's own presence/other fields are never touched. + """ cmds = [] - if state in ("replaced", "overridden"): - for vid in set(have_vifs) - set(want_vifs): - cmds.append(("delete", iface_base + ["vif", str(vid)])) - - for vid, want_vif in want_vifs.items(): + all_vids = set(have_vifs) | set(want_vifs) + for vid in all_vids: + want_vif = want_vifs.get(vid, {"ipv4": [], "ipv6": []}) have_vif = have_vifs.get(vid, {"ipv4": [], "ipv6": []}) vif_base = iface_base + ["vif", str(vid)] cmds += _addr_cmds(vif_base, want_vif["ipv4"], have_vif["ipv4"], state) @@ -317,144 +358,119 @@ def _vif_cmds(iface_base, want_vifs, have_vifs, state): return cmds -def build_commands(config, have_raw, state): - cmds = [] - have_map = _normalize(have_raw) +def build_commands(config, raw_have, state): + raw_have = raw_have or {} + have_list = _device_to_argspec(raw_have) + have_map = _normalize(have_list) if state == "deleted": + cmds = [] if not config: for name, have in have_map.items(): - base = _iface_base(name) - for addr in have["ipv4"] + have["ipv6"]: - cmds.append(("delete", base + ["address", addr])) - for vid in have["vifs"]: - cmds.append(("delete", base + ["vif", str(vid)])) - else: - want_map = _normalize(config) - for name, want in want_map.items(): - have = have_map.get(name, {"ipv4": [], "ipv6": [], "vifs": {}}) - base = _iface_base(name) - if not want["ipv4"] and not want["ipv6"] and not want["vifs"]: - # delete all addresses for this interface - for addr in have["ipv4"] + have["ipv6"]: + base = _iface_base(name, raw_have) + cmds += _addr_cmds(base, [], have["ipv4"], "deleted") + cmds += _addr_cmds(base, [], have["ipv6"], "deleted") + cmds += _vif_addr_cmds(base, {}, have["vifs"], "deleted") + return cmds + for entry in config or []: + name = entry.get("name") + if not name: + continue + have = have_map.get(name, {"ipv4": [], "ipv6": [], "vifs": {}}) + base = _iface_base(name, raw_have) + named_addrs = entry.get("ipv4") or entry.get("ipv6") or entry.get("vifs") + if not named_addrs: + cmds += _addr_cmds(base, [], have["ipv4"], "deleted") + cmds += _addr_cmds(base, [], have["ipv6"], "deleted") + cmds += _vif_addr_cmds(base, {}, have["vifs"], "deleted") + else: + want = _normalize([entry])[name] + for addr in want["ipv4"] + want["ipv6"]: + if addr in have["ipv4"] + have["ipv6"]: cmds.append(("delete", base + ["address", addr])) - for vid in have["vifs"]: - cmds.append(("delete", base + ["vif", str(vid)])) - else: - for addr in want["ipv4"] + want["ipv6"]: - if addr in have["ipv4"] + have["ipv6"]: - cmds.append(("delete", base + ["address", addr])) - for vid in want["vifs"]: - if vid in have["vifs"]: - cmds.append(("delete", base + ["vif", str(vid)])) + for vid, want_vif in want["vifs"].items(): + have_vif = have["vifs"].get(vid, {"ipv4": [], "ipv6": []}) + vif_base = base + ["vif", str(vid)] + want_addrs = want_vif["ipv4"] + want_vif["ipv6"] + addrs = want_addrs or (have_vif["ipv4"] + have_vif["ipv6"]) + for addr in addrs: + if addr in have_vif["ipv4"] + have_vif["ipv6"]: + cmds.append(("delete", vif_base + ["address", addr])) return cmds want_map = _normalize(config) + commands = [] if state == "overridden": for name in set(have_map) - set(want_map): have = have_map[name] - base = _iface_base(name) - for addr in have["ipv4"] + have["ipv6"]: - cmds.append(("delete", base + ["address", addr])) - for vid in have["vifs"]: - cmds.append(("delete", base + ["vif", str(vid)])) + base = _iface_base(name, raw_have) + commands += _addr_cmds(base, [], have["ipv4"], "overridden") + commands += _addr_cmds(base, [], have["ipv6"], "overridden") + commands += _vif_addr_cmds(base, {}, have["vifs"], "overridden") for name, want in want_map.items(): have = have_map.get(name, {"ipv4": [], "ipv6": [], "vifs": {}}) - base = _iface_base(name) - cmds += _addr_cmds(base, want["ipv4"], have["ipv4"], state) - cmds += _addr_cmds(base, want["ipv6"], have["ipv6"], state) - cmds += _vif_cmds(base, want["vifs"], have["vifs"], state) + base = _iface_base(name, raw_have) + commands += _addr_cmds(base, want["ipv4"], have["ipv4"], state) + commands += _addr_cmds(base, want["ipv6"], have["ipv6"], state) + commands += _vif_addr_cmds(base, want["vifs"], have["vifs"], state) - return cmds + return commands +_ADDR_OPTIONS = dict(address=dict(type="str")) + +_VIF_OPTIONS = dict( + vlan_id=dict(type="int", required=True), + ipv4=dict(type="list", elements="dict", options=_ADDR_OPTIONS), + ipv6=dict(type="list", elements="dict", options=_ADDR_OPTIONS), +) + +_ENTRY_OPTIONS = dict( + name=dict(type="str", required=True), + ipv4=dict(type="list", elements="dict", options=_ADDR_OPTIONS), + ipv6=dict(type="list", elements="dict", options=_ADDR_OPTIONS), + vifs=dict(type="list", elements="dict", options=_VIF_OPTIONS), +) + ARGUMENT_SPEC = dict( - config=dict( - type="list", - elements="dict", - options=dict( - name=dict(type="str", required=True), - ipv4=dict( - type="list", - elements="dict", - options=dict(address=dict(type="str")), - ), - ipv6=dict( - type="list", - elements="dict", - options=dict(address=dict(type="str")), - ), - vifs=dict( - type="list", - elements="dict", - options=dict( - vlan_id=dict(type="int", required=True), - ipv4=dict( - type="list", - elements="dict", - options=dict(address=dict(type="str")), - ), - ipv6=dict( - type="list", - elements="dict", - options=dict(address=dict(type="str")), - ), - ), - ), - ), - ), - running_config=dict(type="str"), + config=dict(type="list", elements="dict", options=_ENTRY_OPTIONS), state=dict( type="str", default="merged", - choices=["merged", "replaced", "overridden", "deleted", "gathered", "rendered", "parsed"], + choices=["merged", "replaced", "overridden", "deleted", "gathered"], ), ) def main(): - module = AnsibleModule( - argument_spec=ARGUMENT_SPEC, - mutually_exclusive=[["config", "running_config"]], - required_if=[ - ("state", "rendered", ["config"]), - ("state", "parsed", ["running_config"]), - ], - supports_check_mode=True, - ) + module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True) vyos = VyOSModule(module) state = module.params["state"] config = module.params.get("config") or [] - if state == "parsed": - # parsed is offline — just return empty for now - module.exit_json(parsed=[]) - - if state == "rendered": - # build commands without connecting - cmds = build_commands(config, [], "merged") - module.exit_json(rendered=cmds, commands=cmds) - - have = get_running_config(vyos) + raw_have = get_running_config(vyos) + have = _device_to_argspec(raw_have) 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) module.exit_json( changed=True, before=have, - after=get_running_config(vyos), + after=after, commands=commands, saved=saved, response=response, diff --git a/tests/unit/fixtures/l3_interfaces_running.json b/tests/unit/fixtures/l3_interfaces_running.json index 676055b..3115861 100644 --- a/tests/unit/fixtures/l3_interfaces_running.json +++ b/tests/unit/fixtures/l3_interfaces_running.json @@ -1,21 +1,12 @@ { "ethernet": { "eth0": { - "address": "dhcp", - "hw-id": "52:54:00:65:5a:24", - "vif": { - "100": { - "address": "192.0.2.100/24" - } - } - } + "address": ["192.0.2.1/24", "2001:db8::1/64"], + "vif": { "100": { "address": ["192.0.2.100/24"] } } + }, + "eth1": {} }, "loopback": { - "lo": { - "address": ["10.0.0.1/32", "10.0.0.2/32"] - } - }, - "dummy": { - "dummy0": {} + "lo": { "address": ["10.0.0.1/32"] } } } diff --git a/tests/unit/modules/test_vyos_l3_interfaces.py b/tests/unit/modules/test_vyos_l3_interfaces.py index 66cd1fc..79fc8cf 100644 --- a/tests/unit/modules/test_vyos_l3_interfaces.py +++ b/tests/unit/modules/test_vyos_l3_interfaces.py @@ -4,300 +4,216 @@ from __future__ import absolute_import, division, print_function __metaclass__ = type -import json -import os import unittest from unittest.mock import MagicMock from ansible_collections.vyos.rest.plugins.modules.vyos_l3_interfaces import ( _addr_cmds, - _addr_list, - _normalize, - _parse_iface, + _device_to_argspec, + _guess_iface_type, + _iface_base, + _resolve_iface_type, _split_addresses, + _vif_addr_cmds, build_commands, get_running_config, ) +from .base import load_fixture -def load_fixture(filename): - fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") - path = os.path.join(fixtures_dir, filename) - with open(path) as f: - return json.load(f) + +_BASE = ["interfaces"] class VyOSModuleTestCase(unittest.TestCase): def setUp(self): self.mock_vyos = MagicMock() - self.mock_vyos.get_config = MagicMock(return_value={}) - - def set_running_config(self, data): - self.mock_vyos.get_config.return_value = data - - -class TestVyOSL3InterfacesAddrList(unittest.TestCase): - - def test_string_returns_list(self): - self.assertEqual(_addr_list("dhcp"), ["dhcp"]) - - def test_list_returned_sorted(self): - result = _addr_list(["10.0.0.2/32", "10.0.0.1/32"]) - self.assertEqual(result, ["10.0.0.1/32", "10.0.0.2/32"]) - - def test_none_returns_empty(self): - self.assertEqual(_addr_list(None), []) - - def test_empty_list_returns_empty(self): - self.assertEqual(_addr_list([]), []) - - -class TestVyOSL3InterfacesSplitAddresses(unittest.TestCase): - - def test_dhcp_goes_to_ipv4(self): - ipv4, ipv6 = _split_addresses(["dhcp"]) - self.assertIn("dhcp", ipv4) - self.assertEqual(ipv6, []) - - def test_dhcpv6_goes_to_ipv6(self): - ipv4, ipv6 = _split_addresses(["dhcpv6"]) - self.assertEqual(ipv4, []) - self.assertIn("dhcpv6", ipv6) - - def test_ipv4_cidr(self): - ipv4, ipv6 = _split_addresses(["192.0.2.1/24"]) - self.assertIn("192.0.2.1/24", ipv4) - self.assertEqual(ipv6, []) - - def test_ipv6_cidr(self): - ipv4, ipv6 = _split_addresses(["2001:db8::1/128"]) - self.assertEqual(ipv4, []) - self.assertIn("2001:db8::1/128", ipv6) - - def test_mixed(self): - ipv4, ipv6 = _split_addresses(["dhcp", "192.0.2.1/24", "2001:db8::1/128"]) - self.assertIn("dhcp", ipv4) - self.assertIn("192.0.2.1/24", ipv4) - self.assertIn("2001:db8::1/128", ipv6) - - -class TestVyOSL3InterfacesParseIface(unittest.TestCase): - - def test_parse_dhcp(self): - result = _parse_iface("eth0", {"address": "dhcp"}) - self.assertEqual(result["name"], "eth0") - self.assertEqual(result["ipv4"], [{"address": "dhcp"}]) - - def test_parse_multiple_addresses(self): - result = _parse_iface("lo", {"address": ["10.0.0.1/32", "10.0.0.2/32"]}) - addrs = [a["address"] for a in result["ipv4"]] - self.assertIn("10.0.0.1/32", addrs) - self.assertIn("10.0.0.2/32", addrs) - - def test_parse_vif(self): - result = _parse_iface( - "eth0", - { - "address": "dhcp", - "vif": {"100": {"address": "192.0.2.100/24"}}, - }, - ) - self.assertIn("vifs", result) - self.assertEqual(result["vifs"][0]["vlan_id"], 100) - self.assertEqual(result["vifs"][0]["ipv4"][0]["address"], "192.0.2.100/24") - - def test_parse_no_address(self): - result = _parse_iface("lo", {}) - self.assertNotIn("ipv4", result) - self.assertNotIn("ipv6", result) - - def test_hw_id_ignored(self): - result = _parse_iface("eth0", {"hw-id": "52:54:00:65:5a:24"}) - self.assertNotIn("hw_id", result) - self.assertNotIn("hw-id", result) - - -class TestVyOSL3InterfacesGetRunningFixture(VyOSModuleTestCase): - - def setUp(self): - super().setUp() self.fixture = load_fixture("l3_interfaces_running.json") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) - def test_fixture_parses_eth0_dhcp(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - eth0 = next((e for e in result if e["name"] == "eth0"), None) - self.assertIsNotNone(eth0) - addrs = [a["address"] for a in eth0.get("ipv4", [])] - self.assertIn("dhcp", addrs) + def gather(self): + return _device_to_argspec(self.fixture) - def test_fixture_parses_loopback_addresses(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - lo = next((e for e in result if e["name"] == "lo"), None) - self.assertIsNotNone(lo) - addrs = [a["address"] for a in lo.get("ipv4", [])] - self.assertIn("10.0.0.1/32", addrs) - self.assertIn("10.0.0.2/32", addrs) - - def test_fixture_parses_vif(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - eth0 = next(e for e in result if e["name"] == "eth0") - self.assertIn("vifs", eth0) - self.assertEqual(eth0["vifs"][0]["vlan_id"], 100) - - def test_empty_interface_not_included(self): - self.set_running_config(self.fixture) - result = get_running_config(self.mock_vyos) - names = [e["name"] for e in result] - # loopback with no addresses should not appear - self.assertNotIn("dummy0", names) - - -class TestVyOSL3InterfacesGetRunning(VyOSModuleTestCase): - def test_empty_returns_empty_list(self): - self.set_running_config({}) +class TestGetRunningConfig(VyOSModuleTestCase): + def test_returns_config_directly(self): result = get_running_config(self.mock_vyos) - self.assertEqual(result, []) - - def test_interface_without_address_excluded(self): - self.set_running_config( - { - "ethernet": {"eth0": {"hw-id": "52:54:00:65:5a:24"}}, - }, - ) - result = get_running_config(self.mock_vyos) - self.assertEqual(result, []) - - -class TestVyOSL3InterfacesNormalize(unittest.TestCase): - - def test_normalize_ipv4(self): - config = [ - { - "name": "lo", - "ipv4": [{"address": "10.0.0.1/32"}], - }, - ] - result = _normalize(config) - self.assertIn("lo", result) - self.assertIn("10.0.0.1/32", result["lo"]["ipv4"]) - - def test_normalize_vif(self): - config = [ - { - "name": "eth0", - "vifs": [{"vlan_id": 100, "ipv4": [{"address": "192.0.2.100/24"}]}], - }, - ] - result = _normalize(config) - self.assertIn(100, result["eth0"]["vifs"]) - self.assertIn("192.0.2.100/24", result["eth0"]["vifs"][100]["ipv4"]) - - def test_normalize_empty(self): - result = _normalize([]) - self.assertEqual(result, {}) - - -class TestVyOSL3InterfacesAddrCmds(unittest.TestCase): - - def test_adds_new_address(self): - base = ["interfaces", "loopback", "lo"] - cmds = _addr_cmds(base, ["10.0.0.1/32"], [], "merged") - self.assertIn(("set", base + ["address", "10.0.0.1/32"]), cmds) - - def test_idempotent(self): - base = ["interfaces", "loopback", "lo"] - cmds = _addr_cmds(base, ["10.0.0.1/32"], ["10.0.0.1/32"], "merged") - self.assertEqual(cmds, []) - - def test_merged_does_not_delete_extra(self): - base = ["interfaces", "loopback", "lo"] - cmds = _addr_cmds(base, ["10.0.0.1/32"], ["10.0.0.1/32", "10.0.0.2/32"], "merged") - self.assertEqual(cmds, []) - - def test_replaced_deletes_extra(self): - base = ["interfaces", "loopback", "lo"] - cmds = _addr_cmds(base, ["10.0.0.1/32"], ["10.0.0.1/32", "10.0.0.2/32"], "replaced") - self.assertIn(("delete", base + ["address", "10.0.0.2/32"]), cmds) - - -class TestVyOSL3InterfacesBuildCommands(unittest.TestCase): - - def _have_lo(self): - return [ - { - "name": "lo", - "ipv4": [ - {"address": "10.0.0.1/32"}, - {"address": "10.0.0.2/32"}, - ], - }, - ] - - def test_merged_adds_address(self): - config = [{"name": "lo", "ipv4": [{"address": "10.0.0.3/32"}]}] - cmds = build_commands(config, [], "merged") - self.assertIn( - ("set", ["interfaces", "loopback", "lo", "address", "10.0.0.3/32"]), - cmds, - ) - - def test_merged_idempotent(self): - cmds = build_commands(self._have_lo(), self._have_lo(), "merged") - self.assertEqual(cmds, []) - - def test_deleted_no_config_removes_all(self): - cmds = build_commands([], self._have_lo(), "deleted") - paths = [c[1] for c in cmds] - self.assertIn(["interfaces", "loopback", "lo", "address", "10.0.0.1/32"], paths) - self.assertIn(["interfaces", "loopback", "lo", "address", "10.0.0.2/32"], paths) - - def test_deleted_with_config_removes_interface_addresses(self): - config = [{"name": "lo"}] - cmds = build_commands(config, self._have_lo(), "deleted") - paths = [c[1] for c in cmds] - self.assertIn(["interfaces", "loopback", "lo", "address", "10.0.0.1/32"], paths) - - def test_deleted_idempotent_when_empty(self): - cmds = build_commands([], [], "deleted") + self.assertIn("ethernet", result) + + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + self.assertEqual(get_running_config(self.mock_vyos), {}) + + def test_collapsed_response_normalized(self): + self.mock_vyos.get_config = MagicMock(return_value="eth0") + self.assertEqual(get_running_config(self.mock_vyos), {"eth0": {}}) + + +class TestTypeResolution(unittest.TestCase): + def test_resolves_from_have_when_present(self): + raw_have = {"loopback": {"lo": {}}} + self.assertEqual(_resolve_iface_type("lo", raw_have), "loopback") + + def test_falls_back_to_guess_for_new_interface(self): + self.assertEqual(_resolve_iface_type("eth5", {}), "ethernet") + + def test_guess_covers_all_11_types_aligned_with_vyos_interfaces(self): + """Confirmed original inconsistency: this module's own type + table was missing ppp/wlan compared to vyos_interfaces' table, + despite both modules operating on the same interface + namespace. Aligned here.""" + expected = { + "eth0": "ethernet", + "bond0": "bonding", + "lo": "loopback", + "tun0": "tunnel", + "wg0": "wireguard", + "vti0": "vti", + "dum0": "dummy", + "vtun0": "openvpn", + "ppp0": "pppoe", + "wlan0": "wireless", + "br0": "bridge", + } + for name, itype in expected.items(): + self.assertEqual(_guess_iface_type(name), itype) + + def test_iface_base_uses_resolved_type(self): + raw_have = {"bonding": {"bond0": {}}} + self.assertEqual(_iface_base("bond0", raw_have), ["interfaces", "bonding", "bond0"]) + + +class TestSplitAddresses(unittest.TestCase): + def test_ipv4_ipv6_dhcp_dhcpv6(self): + ipv4, ipv6 = _split_addresses(["192.0.2.1/24", "2001:db8::1/64", "dhcp", "dhcpv6"]) + self.assertEqual(ipv4, ["192.0.2.1/24", "dhcp"]) + self.assertEqual(ipv6, ["2001:db8::1/64", "dhcpv6"]) + + def test_auto_config(self): + ipv4, ipv6 = _split_addresses(["auto-config"]) + self.assertEqual(ipv6, ["auto-config"]) + + +class TestAddrCmds(unittest.TestCase): + """Confirmed scoped correctly: address commands only ever touch + the "address" leaf under the given base, regardless of whether + that base is an interface or a vif.""" + + def test_set_new_address(self): + cmds = _addr_cmds(["interfaces", "ethernet", "eth0"], ["192.0.2.1/24"], [], "merged") + expected = ("set", ["interfaces", "ethernet", "eth0", "address", "192.0.2.1/24"]) + self.assertEqual(cmds, [expected]) + + def test_merged_never_deletes(self): + cmds = _addr_cmds(["interfaces", "ethernet", "eth0"], [], ["192.0.2.1/24"], "merged") self.assertEqual(cmds, []) - def test_replaced_removes_extra_address(self): - config = [{"name": "lo", "ipv4": [{"address": "10.0.0.1/32"}]}] - cmds = build_commands(config, self._have_lo(), "replaced") - self.assertIn( - ("delete", ["interfaces", "loopback", "lo", "address", "10.0.0.2/32"]), - cmds, + def test_replaced_deletes_omitted(self): + cmds = _addr_cmds(["interfaces", "ethernet", "eth0"], [], ["192.0.2.1/24"], "replaced") + expected = ("delete", ["interfaces", "ethernet", "eth0", "address", "192.0.2.1/24"]) + self.assertEqual(cmds, [expected]) + + +class TestVifAddrCmdsNeverWholeSubtree(unittest.TestCase): + """Primary confirmed severe bug: the original generated a whole- + VIF delete (`delete ... vif <id>`) for an omitted VIF under + replaced/overridden/deleted. Since VIFs are shared with + vyos_interfaces (which owns description/mtu/disable on them), + this would destroy that module's own fields, not just this + module's addresses. Confirmed 4 separate occurrences in the + original build_commands. Fixed: every VIF operation here is + scoped to exactly the address leaf, never the VIF subtree.""" + + def test_omitted_vif_only_deletes_its_addresses(self): + have_vifs = {100: {"ipv4": ["192.0.2.100/24"], "ipv6": []}} + cmds = _vif_addr_cmds(["interfaces", "ethernet", "eth0"], {}, have_vifs, "replaced") + whole_vif_delete = ("delete", ["interfaces", "ethernet", "eth0", "vif", "100"]) + self.assertNotIn(whole_vif_delete, cmds) + vif_addr_delete = ( + "delete", + ["interfaces", "ethernet", "eth0", "vif", "100", "address", "192.0.2.100/24"], ) - - def test_overridden_removes_unlisted_interface(self): - config = [{"name": "lo", "ipv4": [{"address": "10.0.0.1/32"}]}] - have = self._have_lo() + [ - { - "name": "eth0", - "ipv4": [{"address": "192.0.2.1/24"}], - }, - ] - cmds = build_commands(config, have, "overridden") - self.assertIn( - ("delete", ["interfaces", "ethernet", "eth0", "address", "192.0.2.1/24"]), - cmds, - ) - - def test_vif_added(self): + self.assertIn(vif_addr_delete, cmds) + + def test_deleted_state_also_never_whole_subtree(self): + have_vifs = {100: {"ipv4": ["192.0.2.100/24"], "ipv6": []}} + cmds = _vif_addr_cmds(["interfaces", "ethernet", "eth0"], {}, have_vifs, "deleted") + whole_vif_delete = ("delete", ["interfaces", "ethernet", "eth0", "vif", "100"]) + self.assertNotIn(whole_vif_delete, cmds) + + def test_overridden_also_never_whole_subtree(self): + have_vifs = {100: {"ipv4": ["192.0.2.100/24"], "ipv6": []}} + cmds = _vif_addr_cmds(["interfaces", "ethernet", "eth0"], {}, have_vifs, "overridden") + whole_vif_delete = ("delete", ["interfaces", "ethernet", "eth0", "vif", "100"]) + self.assertNotIn(whole_vif_delete, cmds) + + +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_all_interfaces_with_addresses_present(self): + have = self.gather() + names = {e["name"] for e in have} + # eth1 has no addresses at all -- must not appear + self.assertEqual(names, {"eth0", "lo"}) + + def test_eth0_addresses_and_vif_parsed(self): + have = self.gather() + eth0 = next(e for e in have if e["name"] == "eth0") + self.assertEqual(eth0["ipv4"][0]["address"], "192.0.2.1/24") + self.assertEqual(eth0["ipv6"][0]["address"], "2001:db8::1/64") + self.assertEqual(eth0["vifs"][0]["vlan_id"], 100) + self.assertEqual(eth0["vifs"][0]["ipv4"][0]["address"], "192.0.2.100/24") + + def test_collapsed_single_address_string(self): + entry = _device_to_argspec({"ethernet": {"eth0": {"address": "192.0.2.1/24"}}}) + self.assertEqual(entry[0]["ipv4"][0]["address"], "192.0.2.1/24") + + +class TestBuildCommands(VyOSModuleTestCase): + def test_merged_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "merged"), []) + + def test_replaced_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "replaced"), []) + + def test_deleted_all_never_whole_interface_subtree(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + cmds = build_commands([], raw_have, "deleted") + self.assertNotIn(("delete", _BASE + ["ethernet", "eth0"]), cmds) + self.assertIn(("delete", _BASE + ["ethernet", "eth0", "address", "192.0.2.1/24"]), cmds) + + def test_deleted_named_never_whole_interface_subtree(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + cmds = build_commands([{"name": "eth0"}], raw_have, "deleted") + self.assertNotIn(("delete", _BASE + ["ethernet", "eth0"]), cmds) + + def test_overridden_omitted_interface_never_whole_subtree(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + cmds = build_commands([], raw_have, "overridden") + self.assertNotIn(("delete", _BASE + ["ethernet", "eth0"]), cmds) + + def test_merged_adds_without_removing(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + config = [{"name": "eth0", "ipv4": [{"address": "192.0.2.2/24"}]}] + cmds = build_commands(config, raw_have, "merged") + self.assertIn(("set", _BASE + ["ethernet", "eth0", "address", "192.0.2.2/24"]), cmds) + self.assertFalse(any(c[0] == "delete" for c in cmds)) + + def test_replaced_removes_unlisted_address(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24", "192.0.2.2/24"]}}} + config = [{"name": "eth0", "ipv4": [{"address": "192.0.2.1/24"}]}] + cmds = build_commands(config, raw_have, "replaced") + self.assertIn(("delete", _BASE + ["ethernet", "eth0", "address", "192.0.2.2/24"]), cmds) + + def test_merged_new_vif_address(self): config = [ - { - "name": "eth0", - "vifs": [{"vlan_id": 100, "ipv4": [{"address": "192.0.2.100/24"}]}], - }, + {"name": "eth0", "vifs": [{"vlan_id": 200, "ipv4": [{"address": "192.0.2.200/24"}]}]}, ] - cmds = build_commands(config, [], "merged") + cmds = build_commands(config, {}, "merged") self.assertIn( - ("set", ["interfaces", "ethernet", "eth0", "vif", "100", "address", "192.0.2.100/24"]), + ("set", _BASE + ["ethernet", "eth0", "vif", "200", "address", "192.0.2.200/24"]), cmds, ) |