summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorDaniil Baturin <daniil@vyos.io>2026-09-28 11:40:29 +0100
committerGitHub <noreply@github.com>2026-09-28 11:40:29 +0100
commitac5b6f72bc9360fce3522fdabb6fa7621f696898 (patch)
treede8a57a24e8202c8499ec3cb33ffbe9917e89f89
parent0c671d0f72ff5222e009024f532a9ecbafe543df (diff)
parent35f2016a0d5c40c3e4b6374ebe459571be544b1f (diff)
downloadrest.vyos-T8989_update_get_vaule.tar.gz
rest.vyos-T8989_update_get_vaule.zip
Merge branch 'main' into T8989_update_get_vauleT8989_update_get_vaule
-rw-r--r--changelogs/fragments/t8989_l3_interface_dict_op.yml3
-rw-r--r--changelogs/fragments/t8989_ospfv3_dict_op.yml3
-rw-r--r--docs/vyos.rest.vyos_l3_interfaces_module.rst51
-rw-r--r--docs/vyos.rest.vyos_ospfv3_module.rst16
-rw-r--r--plugins/modules/vyos_l3_interfaces.py372
-rw-r--r--plugins/modules/vyos_ospfv3.py367
-rw-r--r--tests/unit/fixtures/l3_interfaces_running.json19
-rw-r--r--tests/unit/fixtures/ospfv3_running.json11
-rw-r--r--tests/unit/modules/test_vyos_l3_interfaces.py448
-rw-r--r--tests/unit/modules/test_vyos_ospfv3.py260
10 files changed, 785 insertions, 765 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/changelogs/fragments/t8989_ospfv3_dict_op.yml b/changelogs/fragments/t8989_ospfv3_dict_op.yml
new file mode 100644
index 0000000..b151ac3
--- /dev/null
+++ b/changelogs/fragments/t8989_ospfv3_dict_op.yml
@@ -0,0 +1,3 @@
+---
+minor_changes:
+ - vyos_ospfv3 - Refactor module to dict_op style.
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/docs/vyos.rest.vyos_ospfv3_module.rst b/docs/vyos.rest.vyos_ospfv3_module.rst
index 18bb48e..2b63cf5 100644
--- a/docs/vyos.rest.vyos_ospfv3_module.rst
+++ b/docs/vyos.rest.vyos_ospfv3_module.rst
@@ -19,6 +19,9 @@ Synopsis
--------
- Manages OSPFv3 configuration on VyOS devices via the REST API.
- Uses REST API (``connection=httpapi``) instead of CLI.
+- Scope matches the current vyos.vyos.vyos_ospfv3 (CLI collection) module, confirmed against VyOS's official documentation (1.4+/1.5 LTS/rolling).
+- ``areas.interface`` (an area-to-interface assignment list) exists in the CLI module's argspec but was deliberately NOT carried over here -- confirmed via VyOS's official docs across multiple versions that ``set protocols ospfv3 area <id> interface <name>`` is the superseded, 1.3-era syntax. The current mechanism, ``set protocols ospfv3 interface <name> area <id>``, is a per-interface setting and is modeled in :ref:`vyos.rest.vyos_ospf_interfaces <vyos.rest.vyos_ospf_interfaces_module>`'s ``area`` field instead (and is not affected by this module's ``state=replaced``).
+- ``distance`` and ``graceful-restart`` are real, confirmed OSPFv3 features not modeled here, matching a genuine gap in the CLI module's own scope rather than an oversight.
@@ -133,7 +136,7 @@ Parameters
<td>
</td>
<td>
- <div>Summarize routes matching prefix.</div>
+ <div>Summarize routes matching prefix (border routers only).</div>
</td>
</tr>
<tr>
@@ -278,6 +281,7 @@ Parameters
<a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a>
<div style="font-size: small">
<span style="color: purple">string</span>
+ / <span style="color: red">required</span>
</div>
</td>
<td>
@@ -332,6 +336,16 @@ Notes
- ``ansible_network_os`` must be set to ``vyos.rest.vyos``.
+See Also
+--------
+
+.. seealso::
+
+ :ref:`vyos.vyos.vyos_ospfv3_module`
+ The official documentation on the **vyos.vyos.vyos_ospfv3** module.
+ :ref:`vyos.rest.vyos_ospf_interfaces_module`
+ The official documentation on the **vyos.rest.vyos_ospf_interfaces** module.
+
Examples
--------
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/plugins/modules/vyos_ospfv3.py b/plugins/modules/vyos_ospfv3.py
index fcf404e..5923a61 100644
--- a/plugins/modules/vyos_ospfv3.py
+++ b/plugins/modules/vyos_ospfv3.py
@@ -1,6 +1,7 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
-# GNU General Public License v3.0+
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
from __future__ import absolute_import, division, print_function
@@ -13,6 +14,22 @@ short_description: Manage OSPFv3 configuration on VyOS devices using REST API
description:
- Manages OSPFv3 configuration on VyOS devices via the REST API.
- Uses REST API (C(connection=httpapi)) instead of CLI.
+ - >-
+ Scope matches the current vyos.vyos.vyos_ospfv3 (CLI collection) module,
+ confirmed against VyOS's official documentation (1.4+/1.5 LTS/rolling).
+ - >-
+ C(areas.interface) (an area-to-interface assignment list) exists in the
+ CLI module's argspec but was deliberately NOT carried over here --
+ confirmed via VyOS's official docs across multiple versions that
+ C(set protocols ospfv3 area <id> interface <name>) is the superseded,
+ 1.3-era syntax. The current mechanism, C(set protocols ospfv3 interface
+ <name> area <id>), is a per-interface setting and is modeled in
+ M(vyos.rest.vyos_ospf_interfaces)'s C(area) field instead (and is not
+ affected by this module's C(state=replaced)).
+ - >-
+ C(distance) and C(graceful-restart) are real, confirmed OSPFv3 features
+ not modeled here, matching a genuine gap in the CLI module's own scope
+ rather than an oversight.
version_added: "1.0.0"
author:
- VyOS Community (@vyos)
@@ -37,7 +54,7 @@ options:
description: Name of import-list.
type: str
range:
- description: Summarize routes matching prefix.
+ description: Summarize routes matching prefix (border routers only).
type: list
elements: dict
suboptions:
@@ -65,6 +82,7 @@ options:
suboptions:
route_type:
description: Protocol to redistribute.
+ required: true
type: str
choices: [bgp, connected, kernel, ripng, static]
route_map:
@@ -83,6 +101,9 @@ options:
notes:
- Requires C(ansible_connection=httpapi) with the VyOS httpapi plugin.
- C(ansible_network_os) must be set to C(vyos.rest.vyos).
+seealso:
+ - module: vyos.vyos.vyos_ospfv3
+ - module: vyos.rest.vyos_ospf_interfaces
"""
EXAMPLES = r"""
@@ -139,184 +160,201 @@ 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,
+ autoclean,
+ cast_by_spec,
+ dict_op,
+ from_device,
+ to_tag_dict,
+)
_BASE = ["protocols", "ospfv3"]
+def _kebab_fields(d):
+ """autoclean, then kebab-convert the resulting keys.
+
+ Needed because dict_op requires have's keys to already be genuine
+ device kebab-case -- it only normalizes underscores to dashes for
+ its own lookup index, but uses have's key verbatim for the output
+ path. autoclean deliberately leaves keys exactly as given (dict_op
+ is meant to convert during its own want-vs-have comparison), which
+ only works when have comes straight from the device. Here, have is
+ reconstructed by round-tripping through this module's own entry-
+ transforms, so any field passed through unconverted would stay
+ snake_case and dict_op would have no way to recover the real
+ device key -- confirmed as a real bug during vyos_ospfv2's build.
+ Safe here since every call site is a leaf-level dict of schema
+ field names, never an opaque tag-node value like an area ID used
+ as a dict key.
+ """
+ cleaned = autoclean(d)
+ return {k.replace("_", "-"): v for k, v in cleaned.items()}
+
+
+def _derive_key_field(options_spec):
+ 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=None):
+ entry_transform = entry_transform or _kebab_fields
+ result = {}
+ for item in items or []:
+ if not item.get(key_field):
+ 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 _keyed_list_from_device(raw, key_field, entry_transform=None):
+ entry_transform = entry_transform or from_device
+ return [
+ {key_field: key, **entry_transform(data or {})}
+ for key, data in sorted(to_tag_dict(raw).items())
+ ]
+
+
+# ---------------------------------------------------------------------------
+# range -- confirmed against CLI: advertise/not_advertise are presence-
+# only opposite flags (not both meaningful at once, though the argspec
+# doesn't enforce mutual exclusivity -- matching the CLI's own scope).
+# Fully generic once keyed by address.
+# ---------------------------------------------------------------------------
+
+_RANGE_KEY = "address"
+
+
+def _area_entry_to_device(rest):
+ exclude = {"range"}
+ device = _kebab_fields({k: v for k, v in rest.items() if k not in exclude})
+ ranges = rest.get("range") or []
+ if ranges:
+ device["range"] = _keyed_list_to_device(ranges, _RANGE_KEY)
+ return device
+
+
+def _area_entry_from_device(data):
+ exclude = {"range"}
+ entry = from_device({k: v for k, v in data.items() if k not in exclude})
+ range_raw = data.get("range")
+ if range_raw:
+ entry["range"] = _keyed_list_from_device(range_raw, _RANGE_KEY)
+ return entry
+
+
+def _want_to_device(config):
+ config = config or {}
+ device = {}
+
+ areas = config.get("areas") or []
+ if areas:
+ device["area"] = _keyed_list_to_device(areas, _AREA_KEY, _area_entry_to_device)
+
+ params_device = _kebab_fields(config.get("parameters") or {})
+ if params_device:
+ device["parameters"] = params_device
+
+ redist = config.get("redistribute") or []
+ if redist:
+ device["redistribute"] = _keyed_list_to_device(redist, _REDISTRIBUTE_KEY)
+
+ return device
+
+
def get_running_config(vyos):
- raw = vyos.get_config(_BASE)
+ """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 (an unguarded response
+ iterated character-by-character). 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):
if not raw or not isinstance(raw, dict):
return {}
- return _parse_ospfv3(raw)
+ entry = {}
+ area_raw = raw.get("area")
+ if area_raw:
+ areas = _keyed_list_from_device(area_raw, _AREA_KEY, _area_entry_from_device)
+ if areas:
+ entry["areas"] = areas
-def _parse_ospfv3(raw):
- result = {}
+ params_raw = raw.get("parameters")
+ if params_raw:
+ entry["parameters"] = from_device(params_raw)
- # parameters
- params = raw.get("parameters", {})
- if params:
- result["parameters"] = {}
- if "router-id" in params:
- result["parameters"]["router_id"] = params["router-id"]
-
- # redistribute
- redist_raw = raw.get("redistribute", {})
- if redist_raw and isinstance(redist_raw, dict):
- redist = []
- for route_type, data in sorted(redist_raw.items()):
- entry = {"route_type": route_type}
- if isinstance(data, dict) and data.get("route-map"):
- entry["route_map"] = data["route-map"]
- redist.append(entry)
- if redist:
- result["redistribute"] = redist
-
- # areas
- area_raw = raw.get("area", {})
- if area_raw and isinstance(area_raw, dict):
- areas = []
- for area_id, area_data in sorted(area_raw.items()):
- area = {"area_id": area_id}
- area_data = area_data or {}
- if area_data.get("export-list"):
- area["export_list"] = area_data["export-list"]
- if area_data.get("import-list"):
- area["import_list"] = area_data["import-list"]
- range_raw = area_data.get("range", {})
- if range_raw and isinstance(range_raw, dict):
- ranges = []
- for prefix, rdata in sorted(range_raw.items()):
- r = {"address": prefix}
- rdata = rdata or {}
- if "advertise" in rdata:
- r["advertise"] = True
- if "not-advertise" in rdata:
- r["not_advertise"] = True
- ranges.append(r)
- if ranges:
- area["range"] = ranges
- areas.append(area)
- if areas:
- result["areas"] = areas
+ redist_raw = raw.get("redistribute")
+ if redist_raw:
+ entry["redistribute"] = _keyed_list_from_device(redist_raw, _REDISTRIBUTE_KEY)
- return result
+ return entry
-def build_commands(config, have, state):
- cmds = []
+def build_commands(config, raw_have, state):
+ raw_have = raw_have or {}
if state == "deleted":
- if have:
- cmds.append(("delete", _BASE))
- return cmds
+ return [("delete", _BASE)] if raw_have else []
+
+ want = _want_to_device(config)
+ norm_have = _want_to_device(_device_to_argspec(raw_have))
+ commands = []
if state == "replaced":
- # Build what we would set from scratch and compare to have
- would_set = build_commands(config, {}, "merged")
- have_set = build_commands(have, {}, "merged")
- if would_set == have_set:
- return []
- if have:
- cmds.append(("delete", _BASE))
- have = {}
-
- # parameters
- want_params = (config or {}).get("parameters") or {}
- have_params = have.get("parameters") or {}
- if want_params.get("router_id") and want_params["router_id"] != have_params.get("router_id"):
- cmds.append(("set", _BASE + ["parameters", "router-id", want_params["router_id"]]))
-
- # redistribute
- want_redist = {r["route_type"]: r for r in ((config or {}).get("redistribute") or [])}
- have_redist = {r["route_type"]: r for r in (have.get("redistribute") or [])}
-
- for rt in set(have_redist) - set(want_redist):
- if state == "merged":
- pass # merged doesn't remove
- for rt, entry in want_redist.items():
- if rt not in have_redist:
- cmds.append(("set", _BASE + ["redistribute", rt]))
- if entry.get("route_map"):
- have_rm = have_redist.get(rt, {}).get("route_map")
- if entry["route_map"] != have_rm:
- cmds.append(("set", _BASE + ["redistribute", rt, "route-map", entry["route_map"]]))
-
- # areas
- want_areas = {a["area_id"]: a for a in ((config or {}).get("areas") or [])}
- have_areas = {a["area_id"]: a for a in (have.get("areas") or [])}
-
- for area_id, want_area in want_areas.items():
- have_area = have_areas.get(area_id, {})
- abase = _BASE + ["area", area_id]
-
- if want_area.get("export_list") and want_area["export_list"] != have_area.get(
- "export_list",
- ):
- cmds.append(("set", abase + ["export-list", want_area["export_list"]]))
- if want_area.get("import_list") and want_area["import_list"] != have_area.get(
- "import_list",
- ):
- cmds.append(("set", abase + ["import-list", want_area["import_list"]]))
-
- want_ranges = {r["address"]: r for r in (want_area.get("range") or [])}
- have_ranges = {r["address"]: r for r in (have_area.get("range") or [])}
-
- for addr in want_ranges:
- if addr not in have_ranges:
- cmds.append(("set", abase + ["range", addr]))
- r = want_ranges[addr]
- if r.get("not_advertise"):
- cmds.append(("set", abase + ["range", addr, "not-advertise"]))
- elif r.get("advertise"):
- cmds.append(("set", abase + ["range", addr, "advertise"]))
-
- return cmds
+ commands += dict_op(want, norm_have, _BASE, op="purge")
+ commands += dict_op(want, norm_have, _BASE, op="set")
+ return commands
-ARGUMENT_SPEC = dict(
- config=dict(
- type="dict",
- options=dict(
- areas=dict(
- type="list",
- elements="dict",
- options=dict(
- area_id=dict(type="str", required=True),
- export_list=dict(type="str"),
- import_list=dict(type="str"),
- range=dict(
- type="list",
- elements="dict",
- options=dict(
- address=dict(type="str", required=True),
- advertise=dict(type="bool"),
- not_advertise=dict(type="bool"),
- ),
- ),
- ),
- ),
- parameters=dict(
- type="dict",
- options=dict(
- router_id=dict(type="str"),
- ),
- ),
- redistribute=dict(
- type="list",
- elements="dict",
- options=dict(
- route_type=dict(
- type="str",
- choices=["bgp", "connected", "kernel", "ripng", "static"],
- ),
- route_map=dict(type="str"),
- ),
- ),
- ),
+_RANGE_OPTIONS = dict(
+ address=dict(type="str", required=True),
+ advertise=dict(type="bool"),
+ not_advertise=dict(type="bool"),
+)
+
+_AREA_OPTIONS = dict(
+ area_id=dict(type="str", required=True),
+ export_list=dict(type="str"),
+ import_list=dict(type="str"),
+ range=dict(type="list", elements="dict", options=_RANGE_OPTIONS),
+)
+
+_PARAMETERS_OPTIONS = dict(
+ router_id=dict(type="str"),
+)
+
+_REDISTRIBUTE_OPTIONS = dict(
+ route_type=dict(
+ type="str",
+ required=True,
+ choices=["bgp", "connected", "kernel", "ripng", "static"],
),
+ route_map=dict(type="str"),
+)
+
+_AREA_KEY = _derive_key_field(_AREA_OPTIONS)
+_REDISTRIBUTE_KEY = _derive_key_field(_REDISTRIBUTE_OPTIONS)
+
+_CONFIG_OPTIONS = dict(
+ areas=dict(type="list", elements="dict", options=_AREA_OPTIONS),
+ parameters=dict(type="dict", options=_PARAMETERS_OPTIONS),
+ redistribute=dict(type="list", elements="dict", options=_REDISTRIBUTE_OPTIONS),
+)
+
+ARGUMENT_SPEC = dict(
+ config=dict(type="dict", options=_CONFIG_OPTIONS),
state=dict(
default="merged",
choices=["merged", "replaced", "deleted", "gathered"],
@@ -331,12 +369,14 @@ 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)
+ cast_by_spec(have, _CONFIG_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)
@@ -344,10 +384,13 @@ def main():
if commands:
response = vyos.apply_commands(commands)
saved = vyos.save_config()
+ after_raw = get_running_config(vyos)
+ after = _device_to_argspec(after_raw)
+ cast_by_spec(after, _CONFIG_OPTIONS)
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/fixtures/ospfv3_running.json b/tests/unit/fixtures/ospfv3_running.json
index 44a1bcd..6588a1d 100644
--- a/tests/unit/fixtures/ospfv3_running.json
+++ b/tests/unit/fixtures/ospfv3_running.json
@@ -1,6 +1,4 @@
{
- "parameters": { "router-id": "192.0.2.10" },
- "redistribute": { "bgp": {}, "connected": { "route-map": "RM1" } },
"area": {
"2": {
"export-list": "export1",
@@ -11,9 +9,12 @@
}
},
"3": {
- "range": {
- "2001:db40::/32": {}
- }
+ "range": { "2001:db40::/32": { "advertise": {} } }
}
+ },
+ "parameters": { "router-id": "192.0.2.10" },
+ "redistribute": {
+ "bgp": { "route-map": "redist-map" },
+ "static": {}
}
}
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,
)
diff --git a/tests/unit/modules/test_vyos_ospfv3.py b/tests/unit/modules/test_vyos_ospfv3.py
index a84041b..a19b5fa 100644
--- a/tests/unit/modules/test_vyos_ospfv3.py
+++ b/tests/unit/modules/test_vyos_ospfv3.py
@@ -4,136 +4,218 @@ 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_ospfv3 import (
+ _CONFIG_OPTIONS,
+ ARGUMENT_SPEC,
+ _area_entry_from_device,
+ _area_entry_to_device,
+ _derive_key_field,
+ _device_to_argspec,
+ _kebab_fields,
build_commands,
+ cast_by_spec,
get_running_config,
)
+from .base import load_fixture
-def load_fixture(filename):
- fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures")
- with open(os.path.join(fixtures_dir, filename)) as f:
- return json.load(f)
+
+_BASE = ["protocols", "ospfv3"]
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 TestVyOSOspfv3Parse(VyOSModuleTestCase):
-
- def setUp(self):
- super().setUp()
self.fixture = load_fixture("ospfv3_running.json")
+ self.mock_vyos.get_config = MagicMock(return_value=self.fixture)
- def test_parses_parameters(self):
- self.set_running_config(self.fixture)
- result = get_running_config(self.mock_vyos)
- self.assertEqual(result["parameters"]["router_id"], "192.0.2.10")
+ def gather(self):
+ have = _device_to_argspec(self.fixture)
+ cast_by_spec(have, _CONFIG_OPTIONS)
+ return have
- def test_parses_redistribute(self):
- self.set_running_config(self.fixture)
+
+class TestGetRunningConfig(VyOSModuleTestCase):
+ def test_returns_config_directly(self):
result = get_running_config(self.mock_vyos)
- route_types = [r["route_type"] for r in result["redistribute"]]
- self.assertIn("bgp", route_types)
- self.assertIn("connected", route_types)
- connected = next(r for r in result["redistribute"] if r["route_type"] == "connected")
- self.assertEqual(connected["route_map"], "RM1")
-
- def test_parses_areas(self):
- self.set_running_config(self.fixture)
+ self.assertIn("area", 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):
+ """Regression coverage for the confirmed failure mode from
+ vyos_ospf_interfaces's build: an unguarded collapsed response
+ (VyOS's single-value tag-node quirk) would otherwise be
+ iterated character-by-character downstream."""
+ self.mock_vyos.get_config = MagicMock(return_value="eth1")
result = get_running_config(self.mock_vyos)
- self.assertEqual(len(result["areas"]), 2)
- area2 = next(a for a in result["areas"] if a["area_id"] == "2")
- self.assertEqual(area2["export_list"], "export1")
- self.assertEqual(area2["import_list"], "import1")
- self.assertEqual(len(area2["range"]), 2)
- not_adv = next(r for r in area2["range"] if r["address"] == "2001:db20::/32")
- self.assertTrue(not_adv["not_advertise"])
+ self.assertEqual(result, {"eth1": {}})
- def test_empty_config_returns_empty(self):
- self.set_running_config({})
- result = get_running_config(self.mock_vyos)
- self.assertEqual(result, {})
+class TestDeriveKeyField(unittest.TestCase):
+ def test_derives_area_id(self):
+ area_opts = ARGUMENT_SPEC["config"]["options"]["areas"]["options"]
+ self.assertEqual(_derive_key_field(area_opts), "area_id")
-class TestVyOSOspfv3BuildCommands(unittest.TestCase):
+ def test_raises_if_none_required(self):
+ with self.assertRaises(ValueError):
+ _derive_key_field({"a": {"type": "str"}})
- def test_deleted_with_have(self):
- have = {"parameters": {"router_id": "192.0.2.10"}}
- cmds = build_commands({}, have, "deleted")
- self.assertEqual(cmds, [("delete", ["protocols", "ospfv3"])])
- def test_deleted_without_have(self):
- cmds = build_commands({}, {}, "deleted")
- self.assertEqual(cmds, [])
+class TestKebabFields(unittest.TestCase):
+ def test_converts_multiword_keys(self):
+ result = _kebab_fields({"router_id": "1.1.1.1"})
+ self.assertEqual(result, {"router-id": "1.1.1.1"})
- def test_merged_parameters(self):
- config = {"parameters": {"router_id": "192.0.2.10"}}
- cmds = build_commands(config, {}, "merged")
- self.assertIn(
- ("set", ["protocols", "ospfv3", "parameters", "router-id", "192.0.2.10"]),
- cmds,
+
+class TestAreaEntry(unittest.TestCase):
+ def test_export_import_list_to_device(self):
+ result = _area_entry_to_device({"export_list": "el1", "import_list": "il1"})
+ self.assertEqual(result, {"export-list": "el1", "import-list": "il1"})
+
+ def test_range_to_device(self):
+ result = _area_entry_to_device(
+ {"range": [{"address": "2001:db1::/32", "not_advertise": True}]},
)
+ self.assertEqual(result, {"range": {"2001:db1::/32": {"not-advertise": {}}}})
- def test_merged_redistribute(self):
- config = {"redistribute": [{"route_type": "bgp"}]}
- cmds = build_commands(config, {}, "merged")
- self.assertIn(
- ("set", ["protocols", "ospfv3", "redistribute", "bgp"]),
- cmds,
+ def test_from_device(self):
+ entry = _area_entry_from_device(
+ {"export-list": "el1", "range": {"2001:db1::/32": {"advertise": {}}}},
)
+ self.assertEqual(entry["export_list"], "el1")
+ self.assertEqual(entry["range"][0]["address"], "2001:db1::/32")
+ self.assertTrue(entry["range"][0]["advertise"])
+
+ def test_empty(self):
+ self.assertEqual(_area_entry_to_device({}), {})
+ self.assertEqual(_area_entry_from_device({}), {})
- def test_merged_idempotent(self):
- config = {"parameters": {"router_id": "192.0.2.10"}}
- have = {"parameters": {"router_id": "192.0.2.10"}}
- cmds = build_commands(config, have, "merged")
- self.assertEqual(cmds, [])
- def test_merged_area_range(self):
+class TestDeviceToArgspecFixture(VyOSModuleTestCase):
+ def test_areas_parsed(self):
+ have = self.gather()
+ area_ids = {a["area_id"] for a in have["areas"]}
+ self.assertEqual(area_ids, {"2", "3"})
+
+ def test_area_export_import_list_parsed(self):
+ have = self.gather()
+ area2 = next(a for a in have["areas"] if a["area_id"] == "2")
+ self.assertEqual(area2["export_list"], "export1")
+ self.assertEqual(area2["import_list"], "import1")
+
+ def test_range_advertise_not_advertise_parsed(self):
+ have = self.gather()
+ area2 = next(a for a in have["areas"] if a["area_id"] == "2")
+ r1 = next(r for r in area2["range"] if r["address"] == "2001:db10::/32")
+ r2 = next(r for r in area2["range"] if r["address"] == "2001:db20::/32")
+ self.assertNotIn("advertise", r1)
+ self.assertNotIn("not_advertise", r1)
+ self.assertTrue(r2["not_advertise"])
+
+ area3 = next(a for a in have["areas"] if a["area_id"] == "3")
+ self.assertTrue(area3["range"][0]["advertise"])
+
+ def test_parameters_parsed(self):
+ have = self.gather()
+ self.assertEqual(have["parameters"]["router_id"], "192.0.2.10")
+
+ def test_redistribute_parsed(self):
+ have = self.gather()
+ route_types = {r["route_type"] for r in have["redistribute"]}
+ self.assertEqual(route_types, {"bgp", "static"})
+ bgp = next(r for r in have["redistribute"] if r["route_type"] == "bgp")
+ self.assertEqual(bgp["route_map"], "redist-map")
+
+ def test_empty(self):
+ self.assertEqual(_device_to_argspec({}), {})
+ self.assertEqual(_device_to_argspec(None), {})
+
+
+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_changing_existing_range_flag_via_replaced(self):
+ """Primary confirmed bug from the original hand-rolled
+ implementation: range advertise/not_advertise were only ever
+ set on initial range creation (an "if addr not in have_ranges"
+ guard) -- changing an existing range's flag generated no
+ command under either state. dict_op's purge (via "replaced")
+ correctly handles this generically."""
+ raw_have = {"area": {"2": {"range": {"2001:db20::/32": {"not-advertise": {}}}}}}
config = {
- "areas": [{"area_id": "2", "range": [{"address": "2001:db10::/32"}]}],
+ "areas": [
+ {
+ "area_id": "2",
+ "range": [
+ {"address": "2001:db20::/32", "advertise": True},
+ ],
+ },
+ ],
}
- cmds = build_commands(config, {}, "merged")
+ cmds = build_commands(config, raw_have, "replaced")
self.assertIn(
- ("set", ["protocols", "ospfv3", "area", "2", "range", "2001:db10::/32"]),
+ ("delete", _BASE + ["area", "2", "range", "2001:db20::/32", "not-advertise"]),
cmds,
)
-
- def test_replaced_idempotent(self):
- config = {"parameters": {"router_id": "192.0.2.10"}}
- have = {"parameters": {"router_id": "192.0.2.10"}}
- cmds = build_commands(config, have, "replaced")
- self.assertEqual(cmds, [])
-
- def test_replaced_rebuilds_on_change(self):
- config = {"parameters": {"router_id": "192.0.2.11"}}
- have = {"parameters": {"router_id": "192.0.2.10"}}
- cmds = build_commands(config, have, "replaced")
- self.assertEqual(cmds[0], ("delete", ["protocols", "ospfv3"]))
self.assertIn(
- ("set", ["protocols", "ospfv3", "parameters", "router-id", "192.0.2.11"]),
+ ("set", _BASE + ["area", "2", "range", "2001:db20::/32", "advertise"]),
cmds,
)
- def test_merged_area_export_list(self):
- config = {"areas": [{"area_id": "2", "export_list": "export1"}]}
+ def test_merged_does_not_clear_conflicting_flag(self):
+ """merged never runs a purge pass, by design (established,
+ consistent semantic throughout this collection) -- confirms
+ it correctly does NOT clear not-advertise even when advertise
+ is set instead, unlike replaced above."""
+ raw_have = {"area": {"2": {"range": {"2001:db20::/32": {"not-advertise": {}}}}}}
+ config = {
+ "areas": [
+ {
+ "area_id": "2",
+ "range": [
+ {"address": "2001:db20::/32", "advertise": True},
+ ],
+ },
+ ],
+ }
+ cmds = build_commands(config, raw_have, "merged")
+ self.assertFalse(any(c[0] == "delete" and "not-advertise" in str(c) for c in cmds))
+
+ def test_replaced_is_whole_resource_not_scoped(self):
+ """Confirmed matching this module's own documented contract
+ ("replaced replaces the module-managed OSPFv3 configuration") and
+ vyos_ospfv2's own established precedent -- unlike
+ vyos_static_routes/vyos_route_maps, which scope replaced per
+ named item."""
+ raw_have = {"parameters": {"router-id": "1.1.1.1"}, "redistribute": {"bgp": {}}}
+ config = {"parameters": {"router_id": "1.1.1.1"}}
+ cmds = build_commands(config, raw_have, "replaced")
+ self.assertIn(("delete", _BASE + ["redistribute"]), cmds)
+
+ def test_deleted_with_have(self):
+ cmds = build_commands({}, {"parameters": {"router-id": "1.1.1.1"}}, "deleted")
+ self.assertEqual(cmds, [("delete", _BASE)])
+
+ def test_deleted_no_have_is_noop(self):
+ self.assertEqual(build_commands({}, {}, "deleted"), [])
+
+ def test_merged_new_area(self):
+ config = {"areas": [{"area_id": "5", "export_list": "el5"}]}
cmds = build_commands(config, {}, "merged")
- self.assertIn(
- ("set", ["protocols", "ospfv3", "area", "2", "export-list", "export1"]),
- cmds,
- )
+ self.assertIn(("set", _BASE + ["area", "5", "export-list", "el5"]), cmds)
if __name__ == "__main__":