diff --git a/plugins/modules/vyos_vlan.py b/plugins/modules/vyos_vlan.py index f0b68bc9..9d23cc7c 100644 --- a/plugins/modules/vyos_vlan.py +++ b/plugins/modules/vyos_vlan.py @@ -1,395 +1,391 @@ #!/usr/bin/python # -*- coding: utf-8 -*- # Copyright: (c) 2017, Ansible by Red Hat, inc # 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 __metaclass__ = type DOCUMENTATION = """ module: vyos_vlan author: Trishna Guha (@trishnaguha) short_description: Manage VLANs on VyOS network devices description: - This module provides declarative management of VLANs on VyOS network devices. version_added: 1.0.0 notes: - Tested against VyOS 1.3.8, 1.4.2, the upcoming 1.5, and the rolling release of spring 2025. - This module works with connection C(ansible.netcommon.network_cli). See L(the VyOS OS Platform Options,../network/user_guide/platform_vyos.html). options: name: description: - Name of the VLAN. type: str address: description: - Configure Virtual interface address. type: str vlan_id: description: - ID of the VLAN. Range 0-4094. type: int interfaces: description: - List of interfaces that should be associated to the VLAN. type: list elements: str associated_interfaces: description: - This is a intent option and checks the operational state of the for given vlan C(name) for associated interfaces. If the value in the C(associated_interfaces) does not match with the operational state of vlan on device it will result in failure. type: list elements: str delay: description: - Delay the play should wait to check for declarative intent params values. default: 10 type: int aggregate: description: List of VLANs definitions. type: list elements: dict suboptions: name: description: - Name of the VLAN. type: str address: description: - Configure Virtual interface address. type: str vlan_id: description: - ID of the VLAN. Range 0-4094. type: int required: true interfaces: description: - List of interfaces that should be associated to the VLAN. type: list elements: str required: true associated_interfaces: description: - This is a intent option and checks the operational state of the for given vlan C(name) for associated interfaces. If the value in the C(associated_interfaces) does not match with the operational state of vlan on device it will result in failure. type: list elements: str delay: description: - Delay the play should wait to check for declarative intent params values. type: int state: description: - State of the VLAN configuration. type: str choices: - present - absent purge: description: - Purge VLANs not defined in the I(aggregate) parameter. default: false type: bool state: description: - State of the VLAN configuration. default: present type: str choices: - present - absent extends_documentation_fragment: - vyos.vyos.vyos """ EXAMPLES = """ - name: Create vlan vyos.vyos.vyos_vlan: vlan_id: 100 name: vlan-100 interfaces: eth1 state: present - name: Add interfaces to VLAN vyos.vyos.vyos_vlan: vlan_id: 100 interfaces: - eth1 - eth2 - name: Configure virtual interface address vyos.vyos.vyos_vlan: vlan_id: 100 interfaces: eth1 address: 172.26.100.37/24 - name: vlan interface config + intent vyos.vyos.vyos_vlan: vlan_id: 100 interfaces: eth0 associated_interfaces: - eth0 - name: vlan intent check vyos.vyos.vyos_vlan: vlan_id: 100 associated_interfaces: - eth3 - eth4 - name: Delete vlan vyos.vyos.vyos_vlan: vlan_id: 100 interfaces: eth1 state: absent """ RETURN = """ commands: description: The list of configuration mode commands to send to the device returned: always type: list sample: - set interfaces ethernet eth1 vif 100 description VLAN 100 - set interfaces ethernet eth1 vif 100 address 172.26.100.37/24 - delete interfaces ethernet eth1 vif 100 """ import re import time from copy import deepcopy from ansible.module_utils._text import to_text from ansible.module_utils.basic import AnsibleModule from ansible.module_utils.common.validation import check_required_one_of from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.utils import ( remove_default_spec, ) from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.vyos import ( load_config, run_commands, ) def search_obj_in_list(vlan_id, lst): obj = list() for o in lst: if o["vlan_id"] == vlan_id: obj.append(o) return obj def map_obj_to_commands(updates, module): commands = list() want, have = updates purge = module.params["purge"] for w in want: vlan_id = w["vlan_id"] name = w["name"] address = w["address"] state = w["state"] obj_in_have = search_obj_in_list(vlan_id, have) if state == "absent": if obj_in_have: for obj in obj_in_have: for i in obj["interfaces"]: commands.append("delete interfaces ethernet {0} vif {1}".format(i, vlan_id)) elif state == "present": if not obj_in_have: if w["interfaces"] and w["vlan_id"]: for i in w["interfaces"]: cmd = "set interfaces ethernet {0} vif {1}".format(i, vlan_id) if w["name"]: commands.append(cmd + " description {0}".format(name)) elif w["address"]: commands.append(cmd + " address {0}".format(address)) else: commands.append(cmd) if purge: for h in have: obj_in_want = search_obj_in_list(h["vlan_id"], want) if not obj_in_want: for i in h["interfaces"]: commands.append( "delete interfaces ethernet {0} vif {1}".format(i, h["vlan_id"]), ) return commands def map_params_to_obj(module): obj = [] aggregate = module.params.get("aggregate") if aggregate: for item in aggregate: for key in item: if item.get(key) is None: item[key] = module.params[key] d = item.copy() if not d["vlan_id"]: module.fail_json(msg="vlan_id is required") d["vlan_id"] = str(d["vlan_id"]) try: check_required_one_of(module.required_one_of, item) except TypeError as exc: module.fail_json(to_text(exc)) obj.append(d) else: obj.append( { "vlan_id": str(module.params["vlan_id"]), "name": module.params["name"], "address": module.params["address"], "state": module.params["state"], "interfaces": module.params["interfaces"], "associated_interfaces": module.params["associated_interfaces"], }, ) return obj def map_config_to_obj(module): objs = [] output = run_commands(module, "show interfaces") lines = output[0].strip().splitlines()[3:] for line in lines: splitted_line = re.split(r"\s{2,}", line.strip()) obj = {} eth = splitted_line[0].strip("'") - if eth.startswith("eth"): + if eth.startswith("eth") and "." in eth: obj["interfaces"] = [] - if "." in eth: - interface = eth.split(".")[0] - obj["interfaces"].append(interface) - obj["vlan_id"] = eth.split(".")[-1] - else: - obj["interfaces"].append(eth) - obj["vlan_id"] = None + interface = eth.split(".")[0] + obj["interfaces"].append(interface) + obj["vlan_id"] = eth.split(".")[-1] if splitted_line[1].strip("'") != "-": obj["address"] = splitted_line[1].strip("'") if len(splitted_line) > 3: obj["name"] = splitted_line[3].strip("'") obj["state"] = "present" objs.append(obj) return objs def check_declarative_intent_params(want, module, result): have = None obj_interface = list() is_delay = False for w in want: if w.get("associated_interfaces") is None: continue if result["changed"] and not is_delay: time.sleep(module.params["delay"]) is_delay = True if have is None: have = map_config_to_obj(module) obj_in_have = search_obj_in_list(w["vlan_id"], have) if obj_in_have: for obj in obj_in_have: obj_interface.extend(obj["interfaces"]) for w in want: if w.get("associated_interfaces") is None: continue for i in w["associated_interfaces"]: if (set(obj_interface) - set(w["associated_interfaces"])) != set([]): module.fail_json( msg="Interface {0} not configured on vlan {1}".format(i, w["vlan_id"]), ) def main(): """main entry point for module execution""" element_spec = dict( vlan_id=dict(type="int"), name=dict(), address=dict(), interfaces=dict(type="list", elements="str"), associated_interfaces=dict(type="list", elements="str"), delay=dict(default=10, type="int"), state=dict(default="present", choices=["present", "absent"]), ) aggregate_spec = deepcopy(element_spec) aggregate_spec["vlan_id"].update(required=True) aggregate_spec["interfaces"].update(required=True) # remove default in aggregate spec, to handle common arguments remove_default_spec(aggregate_spec) argument_spec = dict( aggregate=dict(type="list", elements="dict", options=aggregate_spec), purge=dict(default=False, type="bool"), ) argument_spec.update(element_spec) required_one_of = [ ["vlan_id", "aggregate"], ["aggregate", "interfaces", "associated_interfaces"], ] mutually_exclusive = [["vlan_id", "aggregate"]] module = AnsibleModule( argument_spec=argument_spec, supports_check_mode=True, required_one_of=required_one_of, mutually_exclusive=mutually_exclusive, ) warnings = list() result = {"changed": False} if warnings: result["warnings"] = warnings want = map_params_to_obj(module) have = map_config_to_obj(module) commands = map_obj_to_commands((want, have), module) result["commands"] = commands if commands: commit = not module.check_mode load_config(module, commands, commit=commit) result["changed"] = True check_declarative_intent_params(want, module, result) module.exit_json(**result) if __name__ == "__main__": main() diff --git a/tests/unit/modules/network/vyos/test_vyos_vlan.py b/tests/unit/modules/network/vyos/test_vyos_vlan.py index f84a6390..2b663489 100644 --- a/tests/unit/modules/network/vyos/test_vyos_vlan.py +++ b/tests/unit/modules/network/vyos/test_vyos_vlan.py @@ -1,185 +1,184 @@ +# (c) 2024 VyOS Networks +# # This file is part of Ansible # # Ansible is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Ansible is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Ansible. If not, see . +# Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type from unittest.mock import patch from ansible_collections.vyos.vyos.plugins.modules import vyos_vlan from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule # show interfaces output with eth0.100 (vlan 100, description vlan-100) # and eth1.200 (vlan 200, ip 192.0.2.1/24, description vlan-200). # The parser skips the first 3 lines (Codes line, header, dashes). SHOW_INTERFACES_OUTPUT = """\ Codes: S - State, L - Link, u - Up, D - Down, A - Admin Down Interface IP Address S/L Description --------- ---------- --- ----------- eth0 10.0.2.15/24 u/u eth0.100 - u/u vlan-100 eth1 - u/u eth1.200 192.0.2.1/24 u/u vlan-200 eth2 - u/u lo 127.0.0.1/8 u/u ::1/128 """ # Empty show interfaces — no VLANs configured. SHOW_INTERFACES_EMPTY = """\ Codes: S - State, L - Link, u - Up, D - Down, A - Admin Down Interface IP Address S/L Description --------- ---------- --- ----------- eth0 10.0.2.15/24 u/u eth1 - u/u eth2 - u/u lo 127.0.0.1/8 u/u """ class TestVyosVlanModule(TestVyosModule): module = vyos_vlan def setUp(self): super(TestVyosVlanModule, self).setUp() self.mock_load_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_vlan.load_config", ) self.load_config = self.mock_load_config.start() self.mock_run_commands = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_vlan.run_commands", ) self.run_commands = self.mock_run_commands.start() def tearDown(self): super(TestVyosVlanModule, self).tearDown() self.mock_load_config.stop() self.mock_run_commands.stop() def load_fixtures(self, commands=None, filename=None): self.load_config.return_value = dict(diff=None, session="session") - # Default: return the output with existing VLANs. - self.run_commands.return_value = [SHOW_INTERFACES_OUTPUT] + if filename == "empty": + self.run_commands.return_value = [SHOW_INTERFACES_EMPTY] + else: + self.run_commands.return_value = [SHOW_INTERFACES_OUTPUT] def test_vyos_vlan_present(self): """Create a new VLAN with a description on eth2 (not in have).""" set_module_args( dict( vlan_id=300, name="vlan-300", interfaces=["eth2"], state="present", ) ) commands = ["set interfaces ethernet eth2 vif 300 description vlan-300"] self.execute_module(changed=True, commands=commands) def test_vyos_vlan_present_no_change(self): """Existing VLAN 100 on eth0 — no commands should be generated.""" set_module_args( dict( vlan_id=100, name="vlan-100", interfaces=["eth0"], state="present", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_vlan_absent(self): """Delete an existing VLAN (200 on eth1).""" set_module_args( dict( vlan_id=200, interfaces=["eth1"], state="absent", ) ) commands = ["delete interfaces ethernet eth1 vif 200"] self.execute_module(changed=True, commands=commands) def test_vyos_vlan_absent_no_change(self): """Delete a VLAN that does not exist — no commands.""" set_module_args( dict( vlan_id=999, interfaces=["eth0"], state="absent", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_vlan_aggregate(self): """Create two new VLANs via aggregate; neither is in have.""" set_module_args( dict( aggregate=[ dict(vlan_id=301, interfaces=["eth2"], name="vlan-301"), dict(vlan_id=302, interfaces=["eth2"], name="vlan-302"), ], ) ) commands = [ "set interfaces ethernet eth2 vif 301 description vlan-301", "set interfaces ethernet eth2 vif 302 description vlan-302", ] self.execute_module(changed=True, commands=commands) def test_vyos_vlan_purge(self): """Purge VLANs not in want. Want only VLAN 100; VLAN 200 should be removed. - The parser stores all eth* lines including non-VLAN interfaces (vlan_id=None). - Purge deletes every have entry whose vlan_id is not in want, which includes - eth0/None, eth1/None, eth2/None, and eth1.200. + The fixed parser only maps ethX.Y sub-interfaces, so bare ethX interfaces + (vlan_id=None) are not in have. Only real VLANs (eth1.200) are purged. """ set_module_args( dict( vlan_id=100, interfaces=["eth0"], state="present", purge=True, ) ) # VLAN 100 already exists — no create command. - # All have entries with vlan_id not matching '100' are purged. + # Only eth1.200 (vlan_id=200) is purged; bare interfaces are excluded. commands = [ - "delete interfaces ethernet eth0 vif None", - "delete interfaces ethernet eth1 vif None", "delete interfaces ethernet eth1 vif 200", - "delete interfaces ethernet eth2 vif None", ] self.execute_module(changed=True, commands=commands) def test_vyos_vlan_with_address(self): """Create a VLAN with an IP address and no description.""" - # Switch to empty fixture so the new VLAN is not already present. - self.run_commands.return_value = [SHOW_INTERFACES_EMPTY] set_module_args( dict( vlan_id=400, address="10.10.40.1/24", interfaces=["eth1"], state="present", ) ) commands = ["set interfaces ethernet eth1 vif 400 address 10.10.40.1/24"] - self.execute_module(changed=True, commands=commands) + self.execute_module(changed=True, commands=commands, filename="empty")