diff --git a/meta/runtime.yml b/meta/runtime.yml index 45247116..86089e56 100644 --- a/meta/runtime.yml +++ b/meta/runtime.yml @@ -26,6 +26,8 @@ action_groups: - proxmox_nic - proxmox_node - proxmox_node_info + - proxmox_node_network + - proxmox_node_network_info - proxmox_pool - proxmox_pool_member - proxmox_snap diff --git a/plugins/modules/proxmox_node_network.py b/plugins/modules/proxmox_node_network.py new file mode 100644 index 00000000..fcdebef8 --- /dev/null +++ b/plugins/modules/proxmox_node_network.py @@ -0,0 +1,1769 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +# +# Copyright (c) 2025, aleskxyz +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +DOCUMENTATION = r""" +module: proxmox_node_network +version_added: "1.4.0" +short_description: Manage network interfaces on Proxmox nodes +requirements: + - ipaddress +description: + - This module allows you to manage network interfaces on Proxmox nodes. + - Supports various interface types including bridge, bond, eth, vlan, and OVS interfaces. + - Network configuration changes are staged and must be explicitly applied using C(state=apply) to take effect. + - Changes made with C(state=present) or C(state=absent) are only staged and will not affect the running network configuration until applied. +author: "aleskxyz (@aleskxyz)" +notes: + - Network configuration changes are staged and must be explicitly applied using C(state=apply) to take effect. + - Changes made with C(state=present) or C(state=absent) are only staged and will not affect the running network configuration until applied. + - Interface type cannot be changed after creation. Delete and recreate the interface to change its type. + - "Ethernet interfaces (C(eth) type) represent physical hardware and cannot be created via API. Only existing physical interfaces can be configured." + - "Bond and OVS bond interfaces must follow naming format C(bondX) where X is a number between 0 and 9999 (e.g., C(bond0), C(bond1), C(bond9999))." + - "VLAN interfaces support two naming formats: C(vlanXY) (e.g., C(vlan100)) and C(iface_name.vlan_id) (e.g., C(eth0.100))." + - For C(vlanXY) format, C(vlan_raw_device) parameter is required. + - For C(iface_name.vlan_id) format, C(vlan_raw_device) should not be specified. + - "Parameter deletion is supported for specific parameters. Set string parameters to empty string C('') or integer parameters to C(-1) to delete them." + - "Deletable parameters: C(cidr), C(gateway), C(cidr6), C(gateway6), C(comments), C(mtu), + C(bridge_ports), C(bridge_vids), C(ovs_ports), C(ovs_options), C(ovs_tag)." + - "When C(state=apply) or C(state=revert), only the C(node) parameter is accepted. All other parameters are not allowed." +seealso: + - module: community.proxmox.proxmox_node_network_info + description: Retrieve information about network interfaces on Proxmox nodes. + - name: Proxmox Network Configuration + description: Proxmox VE network configuration documentation. + link: https://pve.proxmox.com/wiki/Network_Configuration +attributes: + check_mode: + support: full + details: Check mode is fully supported for all states. + diff_mode: + support: full + details: Check mode is fully supported for all states. +options: + node: + description: + - The Proxmox node to manage network interfaces on. + type: str + required: true + state: + description: + - The desired state of the network interface. + - C(present) and C(absent) stage changes but do not apply them to the running configuration. + - C(apply) applies all staged network configuration changes to the running configuration. + - C(revert) discards all staged network configuration changes. + type: str + choices: + - present + - absent + - apply + - revert + default: present + iface: + description: + - Network interface name. + - Required when C(state=present) or C(state=absent). + - For C(vlan) interface type, should be in format C(vlanXY) (e.g., C(vlan100)) or C(iface_name.vlan_id) (e.g., C(eth0.100)). + type: str + required: false + iface_type: + description: + - Type of network interface. + - Required when C(state=present). + - Cannot be changed after interface creation. + type: str + choices: + - bridge + - bond + - eth + - vlan + - OVSBridge + - OVSBond + - OVSIntPort + required: false + cidr: + description: + - IPv4 host address with prefix length (e.g., '192.168.1.10/24'). + - Supported for C(eth), C(bridge), C(bond), C(vlan), C(OVSBridge), and C(OVSIntPort) interface types. + - Can be deleted by setting to empty string C(''). + type: str + required: false + gateway: + description: + - Default IPv4 gateway address. + - Supported for C(eth), C(bridge), C(bond), C(vlan), C(OVSBridge), and C(OVSIntPort) interface types. + - Can be deleted by setting to empty string C(''). + type: str + required: false + cidr6: + description: + - IPv6 host address with prefix length (e.g., '2001:db8::10/64'). + - Supported for C(eth), C(bridge), C(bond), C(vlan), C(OVSBridge), and C(OVSIntPort) interface types. + - Can be deleted by setting to empty string C(''). + type: str + required: false + gateway6: + description: + - Default IPv6 gateway address. + - Supported for C(eth), C(bridge), C(bond), C(vlan), C(OVSBridge), and C(OVSIntPort) interface types. + - Can be deleted by setting to empty string C(''). + type: str + required: false + autostart: + description: + - Automatically start interface on boot. + - Supported for C(eth), C(bridge), C(bond), C(vlan), and C(OVSBridge) interface types. + type: bool + required: false + comments: + description: + - Comments for the interface configuration. + - Supported for C(eth), C(bridge), C(bond), C(vlan), C(OVSBridge), C(OVSBond), and C(OVSIntPort) interface types. + - Can be deleted by setting to empty string C(''). + type: str + required: false + mtu: + description: + - Maximum Transmission Unit (1280 - 65520). + - Supported for C(eth), C(bridge), C(bond), C(vlan), C(OVSBridge), C(OVSBond), and C(OVSIntPort) interface types. + - Can be deleted by setting to C(-1). + type: int + required: false + # Bridge specific options + bridge_ports: + description: + - Specify the interfaces you want to add to your bridge. + - Supported for C(bridge) interface type only. + - Can be deleted by setting to empty string C(''). + type: str + required: false + bridge_vids: + description: + - Specify the allowed VLANs (e.g., '2 4 100-200'). + - Only used if C(bridge_vlan_aware) is enabled. + - Supported for C(bridge) interface type only. + - Can be deleted by setting to empty string C(''). + type: str + required: false + bridge_vlan_aware: + description: + - Enable bridge VLAN support. + - Supported for C(bridge) interface type only. + type: bool + required: false + # Bond specific options + bond_primary: + description: + - Primary interface for active-backup bond. + - Required for C(active-backup) bonding mode. + - Supported for C(bond) interface type only. + - Primary interface should be specified in C(slaves) parameter as well. + type: str + required: false + bond_mode: + description: + - Bonding mode for C(bond) or C(OVSBond) interface types. + - "Valid values for C(bond) interface type: C(balance-rr), C(active-backup), C(balance-xor), C(broadcast), C(802.3ad), C(balance-tlb), C(balance-alb)" + - "Valid values for C(OVSBond) interface type: C(active-backup), C(balance-slb), C(lacp-balance-slb), C(lacp-balance-tcp)" + type: str + choices: + - balance-rr + - active-backup + - balance-xor + - broadcast + - 802.3ad + - balance-tlb + - balance-alb + - balance-slb + - lacp-balance-slb + - lacp-balance-tcp + required: false + bond_xmit_hash_policy: + description: + - Transmit hash policy for bond type. + - Required for C(balance-xor) and C(802.3ad) bonding modes. + type: str + choices: + - layer2 + - layer2+3 + - layer3+4 + required: false + slaves: + description: + - Interfaces used by the bonding device (space separated). + - Required for C(bond) interface type. + type: str + required: false + # VLAN specific options + vlan_raw_device: + description: + - Raw device for VLAN interface. + - Required if iface is in format 'vlanXY'. + - Supported for C(vlan) interface type only. + type: str + required: false + # OVS specific options + ovs_ports: + description: + - Interfaces to add to OVS bridge. + - Supported for C(OVSBridge) interface type only. + - Can be deleted by setting to empty string C(''). + type: str + required: false + ovs_options: + description: + - OVS interface options. + - Supported for C(OVSBridge), C(OVSBond), and C(OVSIntPort) interface types. + - Can be deleted by setting to empty string C(''). + type: str + required: false + ovs_bonds: + description: + - Interfaces used by OVS bonding device. (space separated) + - Required for C(OVSBond) interface type. + type: str + required: false + ovs_bridge: + description: + - OVS bridge name. + - Required for C(OVSBond) and C(OVSIntPort) interface types. + type: str + required: false + ovs_tag: + description: + - VLAN tag (1 - 4094). + - Supported for C(OVSBond) and C(OVSIntPort) interface types. + - Can be deleted by setting to C(-1). + type: int + required: false +extends_documentation_fragment: + - community.proxmox.proxmox.actiongroup_proxmox + - community.proxmox.proxmox.documentation + - community.proxmox.attributes +""" + +EXAMPLES = r""" +# Configure a network interface +- name: Configure network interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: eth0 + iface_type: eth + cidr: 192.168.1.10/24 + gateway: 192.168.1.1 + cidr6: 2001:db8::10/64 + gateway6: 2001:db8::1 + autostart: true + mtu: 1500 + comments: "Management network" + +# Create a simple bridge interface +- name: Create bridge interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: vmbr0 + iface_type: bridge + cidr: 192.168.1.10/24 + gateway: 192.168.1.1 + autostart: true + +# Create a bond interface +- name: Create bond interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: bond0 + iface_type: bond + bond_mode: active-backup + bond_primary: eth0 + slaves: eth0 eth1 + cidr: 192.168.1.0/24 + gateway: 192.168.1.1 + +# Create a VLAN interface +- name: Create VLAN interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: eth0.100 + iface_type: vlan + cidr: 192.168.100.10/24 + +# Create a VLAN interface with vlanXY format +- name: Create VLAN interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: vlan100 + iface_type: vlan + vlan_raw_device: eth0 + cidr: 192.168.100.0/24 + +# Create a complex bridge with VLAN awareness +- name: Create VLAN-aware bridge + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: vmbr1 + iface_type: bridge + bridge_ports: eth1 eth2 + bridge_vlan_aware: true + bridge_vids: "2 4 100-200" + cidr: 192.168.2.0/24 + gateway: 192.168.2.1 + mtu: 9000 + comments: "VLAN-aware bridge for trunking" + +# Create an OVS bridge +- name: Create OVS bridge + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: ovsbr0 + iface_type: OVSBridge + ovs_ports: eth3 eth4 + ovs_options: "updelay=5000" + cidr: 192.168.3.10/24 + gateway: 192.168.3.1 + +# Create an OVS bond +- name: Create OVS bond + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: bond1 # Must follow bondX format where X is 0-9999 + iface_type: OVSBond + bond_mode: active-backup + ovs_bonds: eth5 eth6 + ovs_bridge: ovsbr0 + ovs_tag: 10 + ovs_options: "updelay=5000" + +# Create an OVS internal port +- name: Create OVS internal port + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: ovsint0 + iface_type: OVSIntPort + ovs_bridge: ovsbr0 + ovs_tag: 20 + ovs_options: "tag=20" + cidr: 192.168.20.10/24 + gateway: 192.168.20.1 + +# Create interface with IPv6 +- name: Create dual-stack interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: vmbr2 + iface_type: bridge + bridge_ports: eth7 + cidr: 192.168.4.0/24 + gateway: 192.168.4.1 + cidr6: 2001:db8::/64 + gateway6: 2001:db8::1 + autostart: true + +# Remove an interface +- name: Remove interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: absent + iface: vmbr0 + +# Apply network configuration +- name: Apply network + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: apply + +# Complete workflow example +- name: Create interface and apply changes + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: vmbr1 + iface_type: bridge + cidr: 192.168.2.10/24 + gateway: 192.168.2.1 + +- name: Apply staged network changes + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: apply + +# Revert network configuration changes +- name: Revert network changes + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: revert + +# Using API token authentication +- name: Create interface with API token + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_token_id: my-token + api_token_secret: my-token-secret + node: pve01 + state: present + iface: vmbr3 + iface_type: bridge + cidr: 192.168.5.10/24 + gateway: 192.168.5.1 + +# Delete specific parameters from an interface +- name: Remove IP configuration from bridge + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: vmbr0 + iface_type: bridge + cidr: "" # Delete IPv4 CIDR + gateway: "" # Delete IPv4 gateway + cidr6: "" # Delete IPv6 CIDR + gateway6: "" # Delete IPv6 gateway + comments: "" # Delete comments + mtu: -1 # Delete MTU (use -1 for integer parameters) + +# Remove bridge ports and VLAN configuration +- name: Remove bridge ports and VLAN settings + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: vmbr1 + iface_type: bridge + bridge_ports: "" # Remove all bridge ports + bridge_vlan_aware: true + bridge_vids: "" # Remove VLAN IDs (requires bridge_vlan_aware: true) + +# Remove OVS-specific parameters +- name: Remove OVS options and ports + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: ovsbr0 + iface_type: OVSBridge + ovs_ports: "" # Remove OVS ports + ovs_options: "" # Remove OVS options + ovs_tag: -1 # Remove VLAN tag (use -1 for integer parameters) + +# Configure existing physical Ethernet interface +- name: Configure physical Ethernet interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: eth0 + iface_type: eth + cidr: 192.168.1.10/24 + gateway: 192.168.1.1 + mtu: 9000 + comments: "Management interface" + # Note: eth interfaces represent physical hardware and cannot be created via API + # Only existing physical interfaces can be configured + +# Create bond interface with proper naming format +- name: Create bond interface + community.proxmox.proxmox_node_network: + api_host: proxmox.example.com:8006 + api_user: root@pam + api_password: secret + node: pve01 + state: present + iface: bond0 # Must follow bondX format where X is 0-9999 + iface_type: bond + bond_mode: active-backup + bond_primary: eth0 + slaves: eth0 eth1 + cidr: 192.168.10.0/24 + gateway: 192.168.10.1 + + +""" + +RETURN = r""" +changed: + description: Whether the module made changes. + returned: always + type: bool + sample: true +msg: + description: A message describing what happened. + returned: always + type: str + sample: "Interface vmbr0 created successfully" +interface: + description: The interface configuration that was applied. + returned: when state is present + type: dict + sample: + iface: vmbr0 + iface_type: bridge + cidr: 192.168.1.0/24 + gateway: 192.168.1.1 + autostart: true + mtu: 1500 + comments: "Management network" +diff: + description: Differences between configurations in YAML format with before/after states. + returned: when changes were made (create, update, delete) + type: list +""" + + +import re +from ipaddress import ip_address, ip_interface +import traceback + +from ansible.module_utils.basic import AnsibleModule +from ansible.module_utils.common.text.converters import to_native +from ansible.module_utils.common.yaml import yaml_dump +from ansible_collections.community.proxmox.plugins.module_utils.proxmox import ( + ProxmoxAnsible, + ansible_to_proxmox_bool, + proxmox_to_ansible_bool, + proxmox_auth_argument_spec, +) + + +# Single source of truth for all parameter definitions +PARAMETER_DEFINITIONS = { + "node": { + "type": "str", + "required": True, + }, + "state": { + "type": "str", + "choices": ["present", "absent", "apply", "revert"], + "default": "present", + }, + "iface": { + "type": "str", + }, + "iface_type": { + "type": "str", + "choices": [ + "bridge", + "bond", + "eth", + "vlan", + "OVSBridge", + "OVSBond", + "OVSIntPort", + ], + "api_name": "type", + }, + "cidr": { + "type": "str", + "deletable": True, + "iface_types": ["eth", "bridge", "bond", "vlan", "OVSBridge", "OVSIntPort"], + }, + "gateway": { + "type": "str", + "deletable": True, + "iface_types": ["eth", "bridge", "bond", "vlan", "OVSBridge", "OVSIntPort"], + }, + "cidr6": { + "type": "str", + "deletable": True, + "iface_types": ["eth", "bridge", "bond", "vlan", "OVSBridge", "OVSIntPort"], + }, + "gateway6": { + "type": "str", + "deletable": True, + "iface_types": ["eth", "bridge", "bond", "vlan", "OVSBridge", "OVSIntPort"], + }, + "autostart": { + "type": "bool", + "iface_types": ["eth", "bridge", "bond", "vlan", "OVSBridge"], + }, + "comments": { + "type": "str", + "deletable": True, + "iface_types": [ + "eth", + "bridge", + "bond", + "vlan", + "OVSBridge", + "OVSBond", + "OVSIntPort", + ], + }, + "mtu": { + "type": "int", + "deletable": True, + "iface_types": [ + "eth", + "bridge", + "bond", + "vlan", + "OVSBridge", + "OVSBond", + "OVSIntPort", + ], + }, + "bridge_ports": { + "type": "str", + "deletable": True, + "default_response": "", + "iface_types": ["bridge"], + }, + "bridge_vids": { + "type": "str", + "deletable": True, + "default_response": "2-4094", + "iface_types": ["bridge"], + }, + "bridge_vlan_aware": { + "type": "bool", + "iface_types": ["bridge"], + }, + "bond_primary": { + "type": "str", + "api_name": "bond-primary", + "iface_types": ["bond"], + }, + "bond_mode": { + "type": "str", + "choices": [ + "balance-rr", + "active-backup", + "balance-xor", + "broadcast", + "802.3ad", + "balance-tlb", + "balance-alb", + "balance-slb", + "lacp-balance-slb", + "lacp-balance-tcp", + ], + "iface_types": ["bond", "OVSBond"], + }, + "bond_xmit_hash_policy": { + "type": "str", + "choices": ["layer2", "layer2+3", "layer3+4"], + "iface_types": ["bond"], + }, + "slaves": { + "type": "str", + "iface_types": ["bond"], + }, + "vlan_raw_device": { + "type": "str", + "api_name": "vlan-raw-device", + "iface_types": ["vlan"], + }, + "ovs_ports": { + "type": "str", + "deletable": True, + "default_response": "", + "iface_types": ["OVSBridge"], + }, + "ovs_options": { + "type": "str", + "deletable": True, + "default_response": "", + "iface_types": ["OVSBridge", "OVSBond", "OVSIntPort"], + }, + "ovs_bonds": { + "type": "str", + "iface_types": ["OVSBond"], + }, + "ovs_bridge": { + "type": "str", + "iface_types": ["OVSBond", "OVSIntPort"], + }, + "ovs_tag": { + "type": "int", + "deletable": True, + "iface_types": ["OVSBond", "OVSIntPort"], + }, + # Read-only parameters returned by API + "active": { + "type": "bool", + "read_only": True, + }, + "vlan_id": { + "type": "int", + "read_only": True, + "api_name": "vlan-id", + }, + "exists": { + "type": "bool", + "read_only": True, + }, + "priority": { + "type": "int", + "read_only": True, + }, +} + + +def _is_valid_cidr(cidr): + """Validate IPv4 host address with prefix using ipaddress.""" + if not cidr: + return False + try: + iface = ip_interface(cidr) + return iface.version == 4 + except Exception: + return False + + +def _is_valid_cidr6(cidr): + """Validate IPv6 host address with prefix using ipaddress.""" + if not cidr: + return False + try: + iface = ip_interface(cidr) + return iface.version == 6 + except Exception: + return False + + +def _is_valid_ipv4(addr): + """Validate IPv4 address using ipaddress.""" + if not addr: + return False + try: + return ip_address(addr).version == 4 + except Exception: + return False + + +def _is_valid_ipv6(addr): + """Validate IPv6 address using ipaddress.""" + if not addr: + return False + try: + return ip_address(addr).version == 6 + except Exception: + return False + + +def get_network_args(): + """Get network-specific arguments for AnsibleModule.""" + args = {} + for param_name, param_def in PARAMETER_DEFINITIONS.items(): + if param_def.get("read_only"): + continue + + arg_def = {"type": param_def["type"]} + + if "choices" in param_def: + arg_def["choices"] = param_def["choices"] + if "default" in param_def: + arg_def["default"] = param_def["default"] + if "required" in param_def: + arg_def["required"] = param_def["required"] + + args[param_name] = arg_def + + return args + + +class ProxmoxNetworkManager(ProxmoxAnsible): + """Manages Proxmox network interfaces.""" + + def __init__(self, module): + super(ProxmoxNetworkManager, self).__init__(module) + self.params = module.params + self.node = self.params["node"] + + def get_params_for_interface_type(self, iface_type): + """Get parameters applicable to a specific interface type.""" + params = [] + for param_name, param_def in PARAMETER_DEFINITIONS.items(): + if param_def.get("read_only"): + continue + + if "iface_types" in param_def and iface_type in param_def["iface_types"]: + params.append(param_name) + + return params + + def get_all_valid_params(self): + """Get all valid parameter names.""" + return list(PARAMETER_DEFINITIONS.keys()) + + def get_core_params(self): + """Get core parameters (not read-only and not interface-specific).""" + return { + param_name + for param_name, param_def in PARAMETER_DEFINITIONS.items() + if not param_def.get("read_only") and "iface_types" not in param_def + } + + def normalize_comment(self, comment): + """Normalize comment string.""" + if comment is None: + return None + return str(comment).strip() + + def get_api_name(self, param_name): + """Get API parameter name for a given Ansible parameter name.""" + if param_name in PARAMETER_DEFINITIONS: + return PARAMETER_DEFINITIONS[param_name].get("api_name", param_name) + return param_name + + def get_ansible_name(self, api_param_name): + """Get Ansible parameter name for a given API parameter name.""" + for param_name, param_def in PARAMETER_DEFINITIONS.items(): + if param_def.get("api_name") == api_param_name: + return param_name + return api_param_name + + def is_delete_intention(self, value, param_name): + """Check if a value represents delete intention for a parameter.""" + if param_name not in PARAMETER_DEFINITIONS: + return False + + param_def = PARAMETER_DEFINITIONS[param_name] + if not param_def.get("deletable"): + return False + + param_type = param_def["type"] + + if param_type == "str": + return value == "" + elif param_type == "int": + return value == -1 + else: + return False + + def get_effective_value(self, value, param_name): + """Get the effective value for validation (treat delete intentions as null).""" + if param_name not in PARAMETER_DEFINITIONS: + return value + + param_def = PARAMETER_DEFINITIONS[param_name] + if not param_def.get("deletable"): + return value + + param_type = param_def["type"] + + if param_type == "str" and value == "": + return None + elif param_type == "int" and value == -1: + return None + else: + return value + + def is_effectively_deleted(self, current_value, param_name): + """Check if a parameter is effectively deleted based on current value.""" + if param_name not in PARAMETER_DEFINITIONS: + return False + + param_def = PARAMETER_DEFINITIONS[param_name] + if not param_def.get("deletable"): + return False + + default_response = param_def.get("default_response") + + if default_response is None: + return current_value is None + else: + # If current_value is None (not present), it's effectively deleted + # because the API will return the default_response when we try to delete it + if current_value is None: + return True + return current_value == default_response + + def validate_common_params(self): + """Validate common parameters.""" + errors = [] + + mtu_value = self.get_effective_value(self.params.get("mtu"), "mtu") + if mtu_value is not None: + try: + mtu = int(mtu_value) + if not (1280 <= mtu <= 65520): + errors.append("mtu must be between 1280 and 65520") + except (ValueError, TypeError): + errors.append("mtu must be an integer") + + cidr_value = self.get_effective_value(self.params.get("cidr"), "cidr") + if cidr_value: + if not _is_valid_cidr(cidr_value): + errors.append("Invalid IPv4 cidr format") + + cidr6_value = self.get_effective_value(self.params.get("cidr6"), "cidr6") + if cidr6_value: + if not _is_valid_cidr6(cidr6_value): + errors.append("Invalid IPv6 cidr format") + + # Gateway requires corresponding CIDR to be defined + gateway_value = self.get_effective_value(self.params.get("gateway"), "gateway") + if gateway_value and not cidr_value: + errors.append("gateway cannot be set when cidr is not defined") + elif gateway_value: + if not _is_valid_ipv4(gateway_value): + errors.append("gateway must be a valid IPv4 address") + + gateway6_value = self.get_effective_value( + self.params.get("gateway6"), "gateway6" + ) + if gateway6_value and not cidr6_value: + errors.append("gateway6 cannot be set when cidr6 is not defined") + elif gateway6_value: + if not _is_valid_ipv6(gateway6_value): + errors.append("gateway6 must be a valid IPv6 address") + + return errors + + def validate_eth_params(self): + """Validate eth interface parameters.""" + errors = [] + return errors + + def validate_bridge_params(self): + """Validate bridge-specific parameters.""" + errors = [] + + # bridge_vids requires bridge_vlan_aware to be enabled + bridge_vids_value = self.get_effective_value( + self.params.get("bridge_vids"), "bridge_vids" + ) + if bridge_vids_value is not None: + if not self.params.get("bridge_vlan_aware"): + errors.append( + "bridge_vids should not be defined if bridge_vlan_aware is not set or false" + ) + + return errors + + def validate_bond_params(self): + """Validate bond-specific parameters.""" + errors = [] + + if not self.params.get("bond_mode"): + errors.append("bond_mode is required for bond type") + else: + bond_mode = self.params.get("bond_mode") + valid_modes = [ + "balance-rr", + "active-backup", + "balance-xor", + "broadcast", + "802.3ad", + "balance-tlb", + "balance-alb", + ] + if bond_mode not in valid_modes: + errors.append( + f"Invalid bond_mode for bond type. Must be one of: {', '.join(valid_modes)}" + ) + + if not self.params.get("slaves"): + errors.append("slaves is required for bond type") + + # Mode-specific validation + bond_mode = self.params.get("bond_mode") + if bond_mode == "active-backup": + if not self.params.get("bond_primary"): + errors.append("bond_primary is required for active-backup mode") + else: + bond_primary = self.params.get("bond_primary") + slaves = self.params.get("slaves", "").split() + if bond_primary not in slaves: + errors.append( + "bond_primary must be included in slaves for active-backup mode" + ) + elif bond_mode in ["balance-xor", "802.3ad"]: + if not self.params.get("bond_xmit_hash_policy"): + errors.append( + "bond_xmit_hash_policy is required for balance-xor and 802.3ad modes" + ) + + # Validate parameter combinations based on bond mode + if self.params.get("bond_primary") is not None and bond_mode != "active-backup": + errors.append( + "bond_primary should not be defined if bond_mode is not active-backup" + ) + + if self.params.get("bond_xmit_hash_policy") is not None and bond_mode not in [ + "balance-xor", + "802.3ad", + ]: + errors.append( + "bond_xmit_hash_policy should not be defined if bond_mode is not balance-xor or 802.3ad" + ) + + return errors + + def validate_vlan_params(self): + """Validate VLAN-specific parameters.""" + errors = [] + + iface = self.params.get("iface", "") + + # Validate vlan_raw_device based on interface naming format + if iface.startswith("vlan"): + if not self.params.get("vlan_raw_device"): + errors.append( + f"vlan_raw_device is required for VLAN interface '{iface}' in vlanXY format" + ) + else: + if self.params.get("vlan_raw_device") is not None: + errors.append( + f"vlan_raw_device should not be specified for VLAN interface '{iface}' in iface_name.vlan_id format" + ) + + return errors + + def validate_ovs_bridge_params(self): + """Validate OVS bridge parameters.""" + errors = [] + return errors + + def validate_ovs_bond_params(self): + """Validate OVS bond parameters.""" + errors = [] + + if not self.params.get("bond_mode"): + errors.append("bond_mode is required for OVSBond type") + else: + bond_mode = self.params.get("bond_mode") + valid_modes = [ + "active-backup", + "balance-slb", + "lacp-balance-slb", + "lacp-balance-tcp", + ] + if bond_mode not in valid_modes: + errors.append( + f"Invalid bond_mode for OVSBond. Must be one of: {', '.join(valid_modes)}" + ) + + if not self.params.get("ovs_bonds"): + errors.append("ovs_bonds is required for OVSBond type") + + if not self.params.get("ovs_bridge"): + errors.append("ovs_bridge is required for OVSBond type") + + # Validate VLAN tag range (1-4094) + ovs_tag_value = self.get_effective_value(self.params.get("ovs_tag"), "ovs_tag") + if ovs_tag_value is not None: + try: + ovs_tag = int(ovs_tag_value) + if not (1 <= ovs_tag <= 4094): + errors.append("ovs_tag must be between 1 and 4094") + except (ValueError, TypeError): + errors.append("ovs_tag must be an integer") + + return errors + + def validate_ovs_int_port_params(self): + """Validate OVS internal port parameters.""" + errors = [] + + if not self.params.get("ovs_bridge"): + errors.append("ovs_bridge is required for OVSIntPort type") + + # Validate VLAN tag range (1-4094) + ovs_tag_value = self.get_effective_value(self.params.get("ovs_tag"), "ovs_tag") + if ovs_tag_value is not None: + try: + ovs_tag = int(ovs_tag_value) + if not (1 <= ovs_tag <= 4094): + errors.append("ovs_tag must be between 1 and 4094") + except (ValueError, TypeError): + errors.append("ovs_tag must be an integer") + + return errors + + def validate_params(self): + """Validate all parameters based on interface type.""" + errors = [] + + # Basic state requirements + state = self.params.get("state", "present") + if state == "present": + if not self.params.get("iface"): + errors.append("iface is required when state is present") + if not self.params.get("iface_type"): + errors.append("iface_type is required when state is present") + elif state == "absent": + if not self.params.get("iface"): + errors.append("iface is required when state is absent") + elif state in ["apply", "revert"]: + # For apply/revert states, only node parameter is allowed + allowed_params = ["node", "state"] + auth_params = list(proxmox_auth_argument_spec().keys()) + all_allowed_params = allowed_params + auth_params + + # Check for any parameters other than node, state, and auth params + for param_name, value in self.params.items(): + if value is not None and param_name not in all_allowed_params: + errors.append( + f"Parameter '{param_name}' is not allowed when state is '{state}'. Only 'node' parameter is accepted." + ) + + # Return early for apply/revert states - no further validation needed + return errors + + # Interface name format validation + errors.extend(self.validate_interface_name()) + + # Parameter combination validation + errors.extend(self.validate_parameter_combinations()) + + if errors: + return errors + + # Parameter value validation + errors.extend(self.validate_common_params()) + + # Type-specific validation + iface_type = self.params.get("iface_type") + if iface_type: + if iface_type == "eth": + errors.extend(self.validate_eth_params()) + elif iface_type == "bridge": + errors.extend(self.validate_bridge_params()) + elif iface_type == "bond": + errors.extend(self.validate_bond_params()) + elif iface_type == "vlan": + errors.extend(self.validate_vlan_params()) + elif iface_type == "OVSBridge": + errors.extend(self.validate_ovs_bridge_params()) + elif iface_type == "OVSBond": + errors.extend(self.validate_ovs_bond_params()) + elif iface_type == "OVSIntPort": + errors.extend(self.validate_ovs_int_port_params()) + + return errors + + def validate_interface_name(self): + """Validate interface name format based on interface type.""" + errors = [] + iface = self.params.get("iface") + iface_type = self.params.get("iface_type") + + if not iface or not iface_type: + return errors + + # Bond interfaces must follow bondX format (X = 0-9999) + if iface_type in ["bond", "OVSBond"]: + if not re.match(r"^bond\d{1,5}$", iface): + errors.append( + f"Interface name '{iface}' for type '{iface_type}' must follow format 'bondX' where X is a number between 0 and 9999" + ) + else: + bond_number = int(iface[4:]) + if bond_number > 9999: + errors.append( + f"bond interface number must be between 0 and 9999, got {bond_number}" + ) + + # VLAN interfaces support two formats: eth0.100 and vlan100 + elif iface_type == "vlan": + if not ( + re.match(r"^(?!vlan\.)[a-zA-Z0-9_-]+\.\d+$", iface) + or re.match(r"^vlan\d+$", iface) + ): + errors.append( + f"VLAN interface name '{iface}' must follow format 'vlanXY' (e.g., vlan100) or 'iface_name.vlan_id' (e.g., eth0.100)" + ) + else: + try: + if iface.startswith("vlan"): + vlan_id = int(iface[4:]) + else: + vlan_id = int(iface.split(".")[-1]) + if not (1 <= vlan_id <= 4094): + errors.append( + f"vlan_id must be between 1 and 4094, got {vlan_id}" + ) + except (ValueError, TypeError): + errors.append("vlan_id must be a valid integer") + + return errors + + def validate_parameter_combinations(self): + """Validate parameter combinations based on interface type.""" + errors = [] + iface_type = self.params.get("iface_type") + + if not iface_type: + return errors + + # Get parameters that are set and validate against allowed parameters + set_params = [key for key, value in self.params.items() if value is not None] + allowed_params = self.get_params_for_interface_type(iface_type) + core_params = self.get_core_params() + auth_params = list(proxmox_auth_argument_spec().keys()) + all_allowed_params = allowed_params + list(core_params) + auth_params + + invalid_params = [ + param for param in set_params if param not in all_allowed_params + ] + + if invalid_params: + errors.append( + f"Parameters {', '.join(invalid_params)} are not valid for interface type '{iface_type}'" + ) + + return errors + + def _is_eth_interface_deleted(self, interface): + """Check if an eth interface is considered deleted (inactive).""" + has_priority = "priority" in interface + autostart = interface.get("autostart", False) + active = interface.get("active", False) + + # Interface is deleted if it has no priority, autostart is false, and active is false + return not has_priority and not autostart and not active + + def get_network_changes(self): + """Check if there are pending network configuration changes. + + Workaround for proxmoxer module which doesn't provide access to the 'changes' field. + """ + try: + resp = self.proxmox_api._store["session"].request( + "GET", + f"{self.proxmox_api._store['base_url']}/nodes/{self.node}/network", + ) + resp.raise_for_status() + changes = resp.json().get("changes", None) + return changes + except Exception as e: + self.module.fail_json( + msg=f"Failed to check network changes for node {self.node}: {e}" + ) + + def get_all_interfaces(self): + """Get all network interfaces.""" + try: + interfaces = self.proxmox_api.nodes(self.node).network.get() + # Convert API parameter names to Ansible format and normalize values + converted_interfaces = [] + for interface in interfaces: + converted_interface = {} + for key, value in interface.items(): + converted_key = self.get_ansible_name(key) + if converted_key == "comments": + value = self.normalize_comment(value) + elif isinstance(value, int) and value in [0, 1]: + if ( + converted_key in PARAMETER_DEFINITIONS + and PARAMETER_DEFINITIONS[converted_key]["type"] == "bool" + ): + value = proxmox_to_ansible_bool(value) + converted_interface[converted_key] = value + converted_interfaces.append(converted_interface) + return converted_interfaces + except Exception as e: + self.module.fail_json( + msg="Failed to get network interfaces: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def get_interface_config(self): + """Get current interface configuration.""" + try: + interfaces = self.get_all_interfaces() + iface = self.params.get("iface") + + # Find the specific interface + for interface in interfaces: + if interface.get("iface") == iface: + return interface + + return None # Interface not found + except Exception as e: + self.module.fail_json( + msg="Failed to get interface configuration: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def create_interface(self): + """Create network interface.""" + try: + data = self._build_interface_data() + # Remove 'delete' property for creation (only valid for updates) + if "delete" in data: + del data["delete"] + self.proxmox_api.nodes(self.node).network.post(**data) + return True + except Exception as e: + self.module.fail_json( + msg="Failed to create interface: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def update_interface(self): + """Update network interface.""" + try: + data = self._build_interface_data() + iface = self.params.get("iface") + self.proxmox_api.nodes(self.node).network(iface).put(**data) + return True + except Exception as e: + self.module.fail_json( + msg="Failed to update interface: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def delete_interface(self): + """Delete network interface.""" + try: + iface = self.params.get("iface") + self.proxmox_api.nodes(self.node).network(iface).delete() + return True + except Exception as e: + self.module.fail_json( + msg="Failed to delete interface: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def apply_network(self): + """Apply network configuration.""" + try: + self.proxmox_api.nodes(self.node).network.put() + return True + except Exception as e: + self.module.fail_json( + msg="Failed to apply network: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def revert_network(self): + """Revert network configuration.""" + try: + self.proxmox_api.nodes(self.node).network.delete() + return True + except Exception as e: + self.module.fail_json( + msg="Failed to revert network: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def execute(self): + """Execute the network management operation.""" + state = self.params.get("state", "present") + + # Validate node exists before any API calls + try: + node_info = self.get_node(self.node) + if not node_info: + self.module.fail_json( + msg=f"Node '{self.node}' not found in the Proxmox cluster" + ) + except Exception as e: + self.module.fail_json(msg=f"Failed to validate node '{self.node}': {e}") + + try: + if state == "present": + return self._handle_present_state() + elif state == "absent": + return self._handle_absent_state() + elif state == "apply": + return self._handle_apply_state() + elif state == "revert": + return self._handle_revert_state() + except Exception as e: + self.module.fail_json( + msg="Failed to manage network interface: %s" % to_native(e), + exception=traceback.format_exc(), + ) + + def _handle_present_state(self): + """Handle present state (create or update).""" + iface = self.params.get("iface") + current_config = self.get_interface_config() + + if current_config: + # Check if interface type is being changed (not allowed) + current_type = current_config.get("iface_type") + desired_type = self.params.get("iface_type") + if current_type and desired_type and current_type != desired_type: + self.module.fail_json( + msg=f"Cannot change interface type from '{current_type}' to '{desired_type}'. " + f"Interface type cannot be modified after creation. " + f"Delete the interface and recreate it with the desired type." + ) + + if self._has_differences(current_config): + if not self.module.check_mode: + self.update_interface() + updated_config = self.get_interface_config() + msg = f"Interface {iface} updated successfully" + after_config = updated_config + before_config = current_config + else: + after_config = self._build_interface_config() + before_config = self._filter_config_to_user_params(current_config) + msg = f"Interface {iface} would be updated" + + return { + "changed": True, + "msg": msg, + "interface": after_config, + "diff": self._format_diff(before_config, after_config, iface), + } + else: + return { + "changed": False, + "msg": f"Interface {iface} already exists with correct configuration", + "interface": current_config, + } + else: + # Prevent creation of eth interfaces (physical hardware) + iface_type = self.params.get("iface_type") + if iface_type == "eth": + self.module.fail_json( + msg=f"Cannot create interface '{iface}' of type 'eth'. " + f"Ethernet interfaces represent physical hardware and cannot be created via API. " + f"Only existing physical interfaces can be configured." + ) + + if not self.module.check_mode: + self.create_interface() + created_config = self.get_interface_config() + msg = f"Interface {iface} created successfully" + else: + created_config = self._build_interface_config() + msg = f"Interface {iface} would be created" + + return { + "changed": True, + "msg": msg, + "interface": created_config, + "diff": self._format_diff(None, created_config, iface), + } + + def _handle_absent_state(self): + """Handle absent state (delete).""" + iface = self.params.get("iface") + iface_type = self.params.get("iface_type") + current_config = self.get_interface_config() + + # Special handling for eth interfaces (check if actually deleted/inactive) + if iface_type == "eth" and current_config: + if self._is_eth_interface_deleted(current_config): + return {"changed": False, "msg": f"Interface {iface} does not exist"} + + if current_config: + if not self.module.check_mode: + self.delete_interface() + msg = f"Interface {iface} deleted successfully" + else: + msg = f"Interface {iface} would be deleted" + return { + "changed": True, + "msg": msg, + "diff": self._format_diff(current_config, None, iface), + } + else: + return {"changed": False, "msg": f"Interface {iface} does not exist"} + + def _handle_apply_state(self): + """Handle apply state.""" + if self.module.check_mode: + return { + "changed": True, + "msg": "Staged network configuration changes may be applied", + } + + changes = self.get_network_changes() + + if changes: + self.apply_network() + return { + "changed": True, + "msg": "Staged network configuration changes applied successfully", + "diff": {"prepared": changes}, + } + else: + return { + "changed": False, + "msg": "No staged network configuration changes to apply", + } + + def _handle_revert_state(self): + """Handle revert state.""" + if self.module.check_mode: + return { + "changed": True, + "msg": "Staged network configuration changes may be reverted", + } + + changes = self.get_network_changes() + + if changes: + self.revert_network() + return { + "changed": True, + "msg": "Staged network configuration changes reverted successfully", + } + else: + return { + "changed": False, + "msg": "No staged network configuration changes to revert", + } + + def _build_interface_data(self): + """Build interface data for API calls.""" + data = {} + delete_list = [] + + if self.params.get("iface"): + data["iface"] = self.params.get("iface") + + if self.params.get("iface_type"): + data["type"] = self.params.get("iface_type") + + # Process all parameters for the interface type + for param_name in self.get_params_for_interface_type( + self.params.get("iface_type") + ): + if self.params.get(param_name) is not None: + value = self.params.get(param_name) + + # Handle delete intentions (empty string for str, -1 for int) + if self.is_delete_intention(value, param_name): + api_param_name = self.get_api_name(param_name) + delete_list.append(api_param_name) + else: + # Handle boolean values: False = delete, True = set to 1 + if isinstance(value, bool): + if value is False: + api_param_name = self.get_api_name(param_name) + delete_list.append(api_param_name) + continue + else: + value = ansible_to_proxmox_bool(value) + + api_param_name = self.get_api_name(param_name) + data[api_param_name] = value + + if delete_list: + data["delete"] = delete_list + + return data + + def _has_differences(self, current_config): + """Check if there are differences between current and desired configuration.""" + core_params = self.get_core_params() + network_params = get_network_args() + for param_name in self.params: + if ( + param_name in network_params + and param_name not in core_params + and self.params.get(param_name) is not None + ): + current_value = current_config.get(param_name) + desired_value = self.params.get(param_name) + + # Check if parameter should be deleted + if self.is_delete_intention(desired_value, param_name): + if not self.is_effectively_deleted(current_value, param_name): + return True + continue + + # Handle boolean comparison with Proxmox format conversion + if isinstance(desired_value, bool): + if current_value is None: + # Assume False if not present in config + if desired_value is not False: + return True + continue + if current_value in [0, 1]: + current_value = proxmox_to_ansible_bool(current_value) + desired_value_proxmox = ansible_to_proxmox_bool(desired_value) + if current_value != desired_value_proxmox: + return True + elif param_name == "comments": + # Normalize comments for comparison + current_value_normalized = self.normalize_comment(current_value) + if current_value_normalized != desired_value: + return True + else: + if str(current_value) != str(desired_value): + return True + + return False + + def _format_diff(self, before_config, after_config, iface): + """Format configuration differences in YAML format for Ansible diff.""" + # Determine operation type for diff formatting + is_create = before_config is None and after_config is not None + is_delete = before_config is not None and after_config is None + + if is_create: + diff_entry = { + "after": yaml_dump(after_config, default_flow_style=False, indent=2), + "after_header": iface, + "before": None, + } + elif is_delete: + diff_entry = { + "before": yaml_dump(before_config, default_flow_style=False, indent=2), + "before_header": iface, + "after": None, + } + else: + diff_entry = { + "before": yaml_dump(before_config, default_flow_style=False, indent=2), + "after": yaml_dump(after_config, default_flow_style=False, indent=2), + "before_header": iface, + "after_header": iface, + } + + return [diff_entry] + + def _filter_config_to_user_params(self, config): + """Filter configuration to only include parameters that the user has defined.""" + if config is None: + return None + + filtered_config = {} + + # Always include interface identification + if "iface" in config: + filtered_config["iface"] = config["iface"] + if "iface_type" in config: + filtered_config["iface_type"] = config["iface_type"] + + # Include only parameters that the user has set + for param_name in self.get_all_valid_params(): + if self.params.get(param_name) is not None: + if param_name in config: + filtered_config[param_name] = config[param_name] + + return filtered_config + + def _build_interface_config(self): + """Build interface configuration dictionary.""" + config = { + "iface": self.params.get("iface"), + "iface_type": self.params.get("iface_type"), + } + + # Add all parameters that are set, with proper value conversion + for param_name in self.get_all_valid_params(): + if self.params.get(param_name) is not None: + value = self.params.get(param_name) + + # Skip delete intentions in config output + if self.is_delete_intention(value, param_name): + continue + + # Convert boolean values to Proxmox format for consistency + if isinstance(value, bool): + value = ansible_to_proxmox_bool(value) + + # Normalize comments for return data + if param_name == "comments" and isinstance(value, str): + value = self.normalize_comment(value) + + config[param_name] = value + + return config + + +def main(): + """Main function.""" + module_args = proxmox_auth_argument_spec() + network_args = get_network_args() + module_args.update(network_args) + + module = AnsibleModule( + argument_spec=module_args, + required_if=[ + ("state", "present", ["iface", "iface_type"]), + ], + required_one_of=[("api_password", "api_token_id")], + required_together=[("api_token_id", "api_token_secret")], + supports_check_mode=True, + ) + + # Create network manager instance + network_manager = ProxmoxNetworkManager(module) + + # Validate parameters + validation_errors = network_manager.validate_params() + if validation_errors: + module.fail_json( + msg="Parameter validation failed: " + "; ".join(validation_errors) + ) + + # Execute the operation + result = network_manager.execute() + + module.exit_json(**result) + + +if __name__ == "__main__": + main() diff --git a/plugins/modules/proxmox_node_network_info.py b/plugins/modules/proxmox_node_network_info.py new file mode 100644 index 00000000..09a65d98 --- /dev/null +++ b/plugins/modules/proxmox_node_network_info.py @@ -0,0 +1,316 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +# +# Copyright (c) 2025, aleskxyz +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +DOCUMENTATION = r""" +module: proxmox_node_network_info +short_description: Retrieve information about Proxmox VE node network interfaces +version_added: "1.4.0" +description: + - Retrieve information about network interfaces on Proxmox VE nodes. + - This module does not make any changes to the system. +author: "aleskxyz (@aleskxyz)" +options: + node: + description: + - The Proxmox node to retrieve network interface information from. + type: str + required: true + iface: + description: + - Name of a specific network interface to retrieve information for. + - If not specified, information for all interfaces will be returned. + type: str + required: false + iface_type: + description: + - Filter results by interface type. + type: str + choices: ['bridge', 'bond', 'eth', 'vlan', 'OVSBridge', 'OVSBond', 'OVSPort', 'OVSIntPort'] + required: false + check_changes: + description: + - Whether to check for pending network configuration changes. + - When enabled, the module will return only information about any staged changes that are waiting to be applied. + - When disabled (default), the module will return network interface information. + - This parameter cannot be used together with C(iface) or C(iface_type) parameters, as checking for pending changes is a node-level operation. + type: bool + default: false + required: false +seealso: + - module: community.proxmox.proxmox_node_network + description: Manage network interfaces on Proxmox nodes. + - name: Proxmox Network Configuration + description: Proxmox VE network configuration documentation. + link: "https://pve.proxmox.com/wiki/Network_Configuration" + - name: Proxmox API Documentation + description: Proxmox VE API documentation. + link: "https://pve.proxmox.com/pve-docs/api-viewer/index.html#/nodes/{node}/network" +attributes: + check_mode: + support: full + details: Check mode is fully supported. + diff_mode: + support: none + details: This is an info module and does not make changes. +extends_documentation_fragment: + - community.proxmox.proxmox.actiongroup_proxmox + - community.proxmox.proxmox.documentation + - community.proxmox.attributes + - community.proxmox.attributes.info_module +""" + +EXAMPLES = r""" +- name: Get all network interfaces on a node + community.proxmox.proxmox_node_network_info: + api_host: proxmox.example.com + api_user: root@pam + api_password: secret + node: pve01 + +- name: Get information about a specific network interface + community.proxmox.proxmox_node_network_info: + api_host: proxmox.example.com + api_user: root@pam + api_password: secret + node: pve01 + iface: vmbr0 + +- name: Get all bridge interfaces on a node + community.proxmox.proxmox_node_network_info: + api_host: proxmox.example.com + api_user: root@pam + api_password: secret + node: pve01 + iface_type: bridge + +- name: Check only for pending changes + community.proxmox.proxmox_node_network_info: + api_host: proxmox.example.com + api_user: root@pam + api_password: secret + node: pve01 + check_changes: true +""" + +RETURN = r""" +proxmox_node_networks: + description: List of network interfaces on the node + returned: success, when check_changes is false or not specified + type: list + elements: dict + sample: [ + { + "iface": "vmbr0", + "type": "bridge", + "active": true, + "autostart": true, + "bridge_ports": "eth0", + "cidr": "192.168.1.1/24", + "address": "192.168.1.1", + "netmask": "255.255.255.0", + "gateway": "192.168.1.254", + "mtu": 1500, + "method": "static", + "families": ["inet"], + "exists": true + }, + { + "iface": "bond0", + "type": "bond", + "active": true, + "autostart": true, + "slaves": "eth1 eth2", + "bond_mode": "active-backup", + "bond-primary": "eth1", + "mtu": 1500, + "method": "manual", + "families": ["inet"], + "exists": true + } + ] +pending_changes: + description: Pending network configuration changes + returned: success, when check_changes is true + type: str + sample: | + --- /etc/network/interfaces + +++ /etc/network/interfaces + @@ -10,6 +10,12 @@ + # The primary network interface + auto eth0 + iface eth0 inet dhcp + + + +# Bridge interface + +auto vmbr1 + +iface vmbr1 inet static + + address 192.168.2.1/24 + + bridge_ports eth1 +has_pending_changes: + description: Whether there are any pending network configuration changes + returned: success, when check_changes is true + type: bool + sample: true +""" + +import traceback + +from ansible.module_utils.basic import AnsibleModule +from ansible.module_utils.common.text.converters import to_native +from ansible_collections.community.proxmox.plugins.module_utils.proxmox import ( + proxmox_auth_argument_spec, + ProxmoxAnsible, + proxmox_to_ansible_bool, +) + + +def main(): + module_args = proxmox_auth_argument_spec() + module_args.update( + node=dict(type="str", required=True), + iface=dict(type="str", required=False), + iface_type=dict( + type="str", + required=False, + choices=[ + "bridge", + "bond", + "eth", + "vlan", + "OVSBridge", + "OVSBond", + "OVSPort", + "OVSIntPort", + ], + ), + check_changes=dict(type="bool", default=False, required=False), + ) + + module = AnsibleModule( + argument_spec=module_args, + supports_check_mode=True, + ) + + proxmox = ProxmoxNodeNetworkInfoAnsible(module) + result = proxmox.run() + module.exit_json(**result) + + +class ProxmoxNodeNetworkInfoAnsible(ProxmoxAnsible): + def check_network_changes(self, node): + """Check if there are pending network configuration changes. + + This is a workaround for the proxmoxer module which doesn't provide + access to the 'changes' field in the API response. We use a direct + HTTP request to access this field for idempotency checks. + """ + try: + resp = self.proxmox_api._store["session"].request( + "GET", f"{self.proxmox_api._store['base_url']}/nodes/{node}/network" + ) + resp.raise_for_status() + changes = resp.json().get("changes", None) + return changes + except Exception as e: + self.module.fail_json( + msg=f"Failed to check network changes for node {node}: {to_native(e)}", + exception=traceback.format_exc(), + ) + + def convert_boolean_values(self, network_data): + """Convert boolean values from Proxmox format (0/1) to Ansible format (True/False).""" + # List of known boolean fields in network interface data + boolean_fields = ["active", "autostart", "bridge_vlan_aware", "exists"] + + converted_data = {} + for key, value in network_data.items(): + if key in boolean_fields and isinstance(value, int) and value in [0, 1]: + converted_data[key] = proxmox_to_ansible_bool(value) + else: + converted_data[key] = value + + return converted_data + + def run(self): + node = self.module.params["node"] + iface = self.module.params.get("iface") + iface_type = self.module.params.get("iface_type") + check_changes = self.module.params.get("check_changes") + result = dict(changed=False) + + # Validate parameter combinations + if check_changes and (iface or iface_type): + self.module.fail_json( + msg="check_changes cannot be used with iface or iface_type parameters. " + "Checking for pending changes is a node-level operation that applies to all network interfaces." + ) + + # Validate node exists + try: + node_info = self.get_node(node) + if not node_info: + self.module.fail_json( + msg=f"Node '{node}' not found in the Proxmox cluster" + ) + except Exception as e: + self.module.fail_json( + msg=f"Failed to validate node '{node}': {to_native(e)}", + exception=traceback.format_exc(), + ) + + try: + # Check for pending changes if requested + if check_changes: + pending_changes = self.check_network_changes(node) + result["pending_changes"] = pending_changes + result["has_pending_changes"] = ( + pending_changes is not None and len(pending_changes.strip()) > 0 + ) + return result + + # Get all interfaces or filter by type using API parameter + if iface_type: + networks = self.proxmox_api.nodes(node).network.get(type=iface_type) + else: + networks = self.proxmox_api.nodes(node).network.get() + + # Convert boolean values for each network interface + converted_networks = [] + for network in networks: + converted_network = self.convert_boolean_values(network) + converted_networks.append(converted_network) + + if iface: + # Search for interface by name + converted_networks = [ + network + for network in converted_networks + if network["iface"] == iface + ] + + result["proxmox_node_networks"] = converted_networks + + except Exception as e: + if iface: + self.module.fail_json( + msg=f"Failed to retrieve information for interface '{iface}' on node '{node}': {to_native(e)}", + exception=traceback.format_exc(), + ) + else: + self.module.fail_json( + msg=f"Failed to retrieve network interfaces from node '{node}': {to_native(e)}", + exception=traceback.format_exc(), + ) + + return result + + +if __name__ == "__main__": + main() diff --git a/tests/unit/plugins/modules/test_proxmox_node_network.py b/tests/unit/plugins/modules/test_proxmox_node_network.py new file mode 100644 index 00000000..669fefa5 --- /dev/null +++ b/tests/unit/plugins/modules/test_proxmox_node_network.py @@ -0,0 +1,1970 @@ +# -*- coding: utf-8 -*- +# +# Copyright (c) 2025, aleskxyz +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from unittest.mock import patch +import pytest + +proxmoxer = pytest.importorskip("proxmoxer") + +from ansible_collections.community.proxmox.plugins.modules import proxmox_node_network +from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import ( + AnsibleExitJson, + AnsibleFailJson, + ModuleTestCase, + set_module_args, +) +import ansible_collections.community.proxmox.plugins.module_utils.proxmox as proxmox_utils + +# Mock API response for existing network interfaces +EXISTING_NETWORK_OUTPUT = [ + { + "iface": "eth0", + "type": "eth", + "active": 1, + "autostart": 1, + "mtu": 1500, + "method": "manual", + "families": ["inet"], + }, + { + "iface": "vmbr0", + "type": "bridge", + "active": 1, + "autostart": 1, + "mtu": 1500, + "method": "static", + "families": ["inet"], + "bridge_ports": "eth0", + "address": "192.168.1.1", + "netmask": "255.255.255.0", + }, +] + + +class TestProxmoxNodeNetwork(ModuleTestCase): + def setUp(self): + super(TestProxmoxNodeNetwork, self).setUp() + proxmox_utils.HAS_PROXMOXER = True + self.module = proxmox_node_network + + self.connect_mock = patch( + "ansible_collections.community.proxmox.plugins.module_utils.proxmox.ProxmoxAnsible._connect", + ).start() + + mock_nodes = self.connect_mock.return_value.nodes + mock_node_obj = mock_nodes.return_value + mock_nodes.side_effect = lambda node=None: mock_node_obj + + mock_network_obj = mock_node_obj.network.return_value + mock_node_obj.network = mock_network_obj + mock_node_obj.network.return_value = mock_network_obj + mock_network_obj.get.return_value = EXISTING_NETWORK_OUTPUT + + mock_nodes.get.return_value = [{"node": "pve"}] + + mock_network_obj.get.side_effect = lambda type=None: [ + interface + for interface in EXISTING_NETWORK_OUTPUT + if type is None or interface["type"] == type + ] + + def tearDown(self): + self.connect_mock.stop() + super(TestProxmoxNodeNetwork, self).tearDown() + + def test_invalid_node(self): + """Test invalid node.""" + with patch.object( + self.module.ProxmoxNetworkManager, "get_node", return_value=None + ): + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "nonexistent", + "iface": "eth0", + "iface_type": "eth", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "Node 'nonexistent' not found" in result["msg"] + + def test_invalid_state(self): + """Test invalid state.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "state": "invalid_state", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "value of state must be one of: present, absent, apply, revert" + in result["msg"] + ) + + def test_create_interface_all_types_minimum_params(self): + """Test creating interface for all if_types with minimum params.""" + interface_types = [ + "bridge", + "bond", + "vlan", + "OVSBridge", + "OVSBond", + "OVSIntPort", + ] + + for iface_type in interface_types: + # Use proper interface names based on type + if iface_type == "bond": + iface_name = "bond0" + elif iface_type == "OVSBond": + iface_name = "bond1" + elif iface_type == "OVSIntPort": + iface_name = "ovsint0" + elif iface_type == "OVSBridge": + iface_name = "ovsbr0" + elif iface_type == "vlan": + iface_name = "vlan100" + else: + iface_name = f"test_{iface_type}" + + created_config = {"iface": iface_name, "iface_type": iface_type} + + # Add required parameters based on interface type + module_args = { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": iface_name, + "iface_type": iface_type, + } + + if iface_type == "bond": + module_args.update( + { + "bond_mode": "balance-rr", + "slaves": "eth0 eth1", + } + ) + elif iface_type == "vlan": + module_args.update( + { + "vlan_raw_device": "eth0", + } + ) + elif iface_type == "OVSBond": + module_args.update( + { + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + } + ) + elif iface_type == "OVSIntPort": + module_args.update( + { + "ovs_bridge": "ovsbr0", + } + ) + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, created_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, + "create_interface", + return_value=True, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args(module_args): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == iface_name + assert "created" in result["msg"].lower() + + def test_create_interface_wrong_if_type(self): + """Test creating interface with wrong if_type.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "test0", + "iface_type": "invalid_type", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "value of iface_type must be one of: bridge, bond, eth, vlan, OVSBridge, OVSBond, OVSIntPort, got: invalid_type" + in result["msg"] + ) + + def test_create_eth_interface_non_existing(self): + """Test creating eth interface with non existing interface name.""" + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=None, # Interface doesn't exist + ): + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "nonexistent_eth", + "iface_type": "eth", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "Cannot create interface 'nonexistent_eth' of type 'eth'" + in result["msg"] + ) + + def test_create_eth_interface_existing(self): + """Test creating eth interface with existing interface name.""" + existing_config = { + "iface": "eth0", + "type": "eth", + "active": 1, + "autostart": 1, + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=existing_config, + ): + with patch.object( + self.module.ProxmoxNetworkManager, "update_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "eth0", + "iface_type": "eth", + "cidr": "192.168.1.0/24", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "eth0" + assert "updated" in result["msg"].lower() + + def test_delete_eth_interface(self): + """Test deleting a eth interface.""" + existing_config = { + "iface": "eth0", + "type": "eth", + "active": 1, + "autostart": 1, + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=existing_config, + ): + with patch.object( + self.module.ProxmoxNetworkManager, "delete_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "eth0", + "state": "absent", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "deleted" in result["msg"].lower() + + def test_create_bridge_all_params_and_delete_one_by_one(self): + """Test creating bridge interface with all possible params and deleting them one by one.""" + # First create with all parameters + all_params_config = { + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + "cidr": "192.168.1.0/24", + "gateway": "192.168.1.1", + "cidr6": "2001:db8::/64", + "gateway6": "2001:db8::1", + "comments": "Test bridge", + "mtu": 9000, + "bridge_vids": "100 200", + "bridge_vlan_aware": True, + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, all_params_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + "cidr": "192.168.1.0/24", + "gateway": "192.168.1.1", + "cidr6": "2001:db8::/64", + "gateway6": "2001:db8::1", + "comments": "Test bridge", + "mtu": 9000, + "bridge_vids": "100 200", + "bridge_vlan_aware": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vmbr0" + + # Now test deleting parameters one by one + params_to_delete = [ + ("cidr", ""), + ("gateway", ""), + ("cidr6", ""), + ("gateway6", ""), + ("comments", ""), + ("mtu", -1), + ("bridge_ports", ""), + ("bridge_vids", ""), + ] + + for param_name, delete_value in params_to_delete: + # Create config without the parameter being deleted + updated_config = all_params_config.copy() + if param_name in ["cidr", "gateway", "cidr6", "gateway6", "comments"]: + updated_config.pop(param_name, None) + elif param_name == "mtu": + updated_config.pop(param_name, None) + elif param_name == "bridge_ports": + updated_config[param_name] = "" + elif param_name == "bridge_vids": + updated_config[param_name] = "2-4094" # Default when deleted + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[all_params_config, updated_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, + "update_interface", + return_value=True, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + module_args = { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + "bridge_vlan_aware": True, + } + module_args[param_name] = delete_value + + with set_module_args(module_args): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vmbr0" + assert "updated" in result["msg"].lower() + + def test_create_bridge_all_params_and_update_one_by_one(self): + """Test creating bridge interface with all possible params and updating them one by one.""" + # First create with all parameters + all_params_config = { + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + "cidr": "192.168.1.0/24", + "gateway": "192.168.1.1", + "cidr6": "2001:db8::/64", + "gateway6": "2001:db8::1", + "comments": "Test bridge", + "mtu": 9000, + "bridge_vids": "100 200", + "bridge_vlan_aware": True, + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, all_params_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + "cidr": "192.168.1.0/24", + "gateway": "192.168.1.1", + "cidr6": "2001:db8::/64", + "gateway6": "2001:db8::1", + "comments": "Test bridge", + "mtu": 9000, + "bridge_vids": "100 200", + "bridge_vlan_aware": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vmbr0" + + # Now test updating parameters one by one + params_to_update = [ + ("cidr", "192.168.2.0/24"), + ("gateway", "192.168.2.1"), + ("cidr6", "2001:db9::/64"), + ("gateway6", "2001:db9::1"), + ("comments", "Updated bridge"), + ("mtu", 1500), + ("bridge_ports", "eth2 eth3"), + ("bridge_vids", "300 400"), + ] + + for param_name, new_value in params_to_update: + # Create config with the updated parameter + updated_config = all_params_config.copy() + updated_config[param_name] = new_value + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[all_params_config, updated_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, + "update_interface", + return_value=True, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + module_args = { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + "bridge_vlan_aware": True, + } + + # For gateway updates, we need to include cidr + if param_name == "gateway": + module_args["cidr"] = "192.168.1.0/24" + elif param_name == "gateway6": + module_args["cidr6"] = "2001:db8::/64" + + module_args[param_name] = new_value + + with set_module_args(module_args): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vmbr0" + assert "updated" in result["msg"].lower() + + def test_gateway_without_cidr_should_fail(self): + """gateway requires cidr to be set.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + # Intentionally omit cidr + "gateway": "192.168.1.1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "gateway cannot be set when cidr is not defined" in result["msg"] + + def test_gateway6_without_cidr6_should_fail(self): + """gateway6 requires cidr6 to be set.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + # Intentionally omit cidr6 + "gateway6": "2001:db8::1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "gateway6 cannot be set when cidr6 is not defined" in result["msg"] + + def test_invalid_gateway_format_should_fail(self): + """gateway must be a valid IPv4 address when provided.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "cidr": "192.168.1.10/24", + "gateway": "999.999.999.999", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "gateway must be a valid IPv4 address" in result["msg"] + + # Also fail if IPv6 is passed to IPv4 gateway + with pytest.raises(AnsibleFailJson) as exc_info2: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "cidr": "192.168.1.10/24", + "gateway": "2001:db8::1", + } + ): + self.module.main() + + result2 = exc_info2.value.args[0] + assert "gateway must be a valid IPv4 address" in result2["msg"] + + def test_invalid_gateway6_format_should_fail(self): + """gateway6 must be a valid IPv6 address when provided.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "cidr6": "2001:db8::10/64", + "gateway6": "not_an_ip", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "gateway6 must be a valid IPv6 address" in result["msg"] + + # Also fail if IPv4 is passed to IPv6 gateway6 + with pytest.raises(AnsibleFailJson) as exc_info2: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "cidr6": "2001:db8::10/64", + "gateway6": "192.168.1.1", + } + ): + self.module.main() + + result2 = exc_info2.value.args[0] + assert "gateway6 must be a valid IPv6 address" in result2["msg"] + + def test_create_bond_invalid_name(self): + """Test creating bond with invalid name.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "invalid_bond", + "iface_type": "bond", + "bond_mode": "balance-rr", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "Interface name 'invalid_bond' for type 'bond' must follow format 'bondX'" + in result["msg"] + ) + + def test_create_bond_lacp_balance_slb_mode(self): + """Test creating bond with 'lacp-balance-slb' mode which is only valid for ovsbond.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "lacp-balance-slb", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "Invalid bond_mode for bond type. Must be one of: balance-rr, active-backup, balance-xor, broadcast, 802.3ad, balance-tlb, balance-alb" + in result["msg"] + ) + + def test_create_bond_active_backup_without_primary(self): + """Test creating bond with 'active-backup' mode and don't mention bond_primary.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "active-backup", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "bond_primary is required for active-backup mode" in result["msg"] + + def test_create_bond_active_backup_primary_not_in_slaves(self): + """Test creating bond with 'active-backup' mode and mention bond_primary but don't include it in slaves.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "active-backup", + "bond_primary": "eth0", + "slaves": "eth1 eth2", # eth0 not in slaves + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "bond_primary must be included in slaves for active-backup mode" + in result["msg"] + ) + + def test_create_bond_active_backup_primary_in_slaves(self): + """Test creating bond with 'active-backup' mode and mention bond_primary and include it in slaves.""" + created_config = { + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "active-backup", + "bond_primary": "eth0", + "slaves": "eth0 eth1", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, created_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "active-backup", + "bond_primary": "eth0", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "bond0" + assert "created" in result["msg"].lower() + + def test_create_bond_balance_xor_without_hash_policy(self): + """Test creating bond with 'balance-xor' mode and don't mention bond_xmit_hash_policy.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-xor", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "bond_xmit_hash_policy is required for balance-xor and 802.3ad modes" + in result["msg"] + ) + + def test_create_bond_balance_xor_invalid_hash_policy(self): + """Test creating bond with 'balance-xor' mode and mention invalid bond_xmit_hash_policy.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-xor", + "bond_xmit_hash_policy": "invalid_policy", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "value of bond_xmit_hash_policy must be one of: layer2, layer2+3, layer3+4" + in result["msg"] + ) + + def test_create_bond_balance_xor_valid_hash_policy(self): + """Test creating bond with 'balance-xor' mode and mention valid bond_xmit_hash_policy.""" + created_config = { + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-xor", + "bond_xmit_hash_policy": "layer2", + "slaves": "eth0 eth1", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, created_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-xor", + "bond_xmit_hash_policy": "layer2", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "bond0" + assert "created" in result["msg"].lower() + + def test_create_bond_balance_rr_with_primary(self): + """Test creating bond with 'balance-rr' mode and mention bond_primary.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-rr", + "bond_primary": "eth0", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "bond_primary should not be defined if bond_mode is not active-backup" + in result["msg"] + ) + + def test_create_bond_balance_rr_mode(self): + """Test creating bond with 'balance-rr' mode.""" + created_config = { + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-rr", + "slaves": "eth0 eth1", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, created_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-rr", + "slaves": "eth0 eth1", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "bond0" + assert "created" in result["msg"].lower() + + def test_create_bond_balance_rr_without_slaves(self): + """Test creating bond with 'balance-rr' mode without slaves.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "balance-rr", + # Missing slaves + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "slaves is required for bond type" in result["msg"] + + def test_create_vlan_interface_vlan10_name(self): + """Test creating vlan interface with vlan.10 name.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vlan.10", + "iface_type": "vlan", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "VLAN interface name 'vlan.10' must follow format 'vlanXY'" in result["msg"] + ) + + def test_create_vlan_interface_eth10_name(self): + """Test creating vlan interface with eth10 name.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "eth10", + "iface_type": "vlan", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "VLAN interface name 'eth10' must follow format 'vlanXY'" in result["msg"] + ) + + def test_create_vlan_interface_vlan10_name_no_raw_device(self): + """Test creating vlan interface with vlan10 name and don't mention vlan_raw_device.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vlan10", + "iface_type": "vlan", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "vlan_raw_device is required for VLAN interface 'vlan10' in vlanXY format" + in result["msg"] + ) + + def test_create_vlan_interface_vlan10_name_with_raw_device(self): + """Test creating vlan interface with vlan10 name and mention vlan_raw_device.""" + created_config = { + "iface": "vlan10", + "iface_type": "vlan", + "vlan_raw_device": "eth0", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, created_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vlan10", + "iface_type": "vlan", + "vlan_raw_device": "eth0", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vlan10" + assert "created" in result["msg"].lower() + + def test_create_vlan_interface_eth0_10_name_no_raw_device(self): + """Test creating vlan interface with eth0.10 name and don't mention vlan_raw_device.""" + created_config = { + "iface": "eth0.10", + "iface_type": "vlan", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, created_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "eth0.10", + "iface_type": "vlan", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "eth0.10" + assert "created" in result["msg"].lower() + + def test_create_vlan_interface_eth0_10_name_with_raw_device(self): + """Test creating vlan interface with eth0.10 name and mention vlan_raw_device.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "eth0.10", + "iface_type": "vlan", + "vlan_raw_device": "eth0", # Should not be specified for dot format + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "vlan_raw_device should not be specified for VLAN interface 'eth0.10' in iface_name.vlan_id format" + in result["msg"] + ) + + def test_create_ovsbond_without_bridge(self): + """Test creating ovsbond without bridge.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + # Missing ovs_bridge + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "ovs_bridge is required for OVSBond type" in result["msg"] + + def test_create_ovsbond_without_slave(self): + """Test creating ovsbond without slave.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bridge": "ovsbr0", + # Missing ovs_bonds + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "ovs_bonds is required for OVSBond type" in result["msg"] + + def test_create_ovsbond_with_balance_rr_type(self): + """Test creating ovsbond with balance-rr type.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "balance-rr", # Invalid for OVSBond + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "Invalid bond_mode for OVSBond. Must be one of: active-backup, balance-slb, lacp-balance-slb, lacp-balance-tcp" + in result["msg"] + ) + + def test_create_ovsbond_with_invalid_name(self): + """Test creating ovsbond with invalid name.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "invalid_ovsbond", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "Interface name 'invalid_ovsbond' for type 'OVSBond' must follow format 'bondX'" + in result["msg"] + ) + + def test_create_ovsbond_with_autostart(self): + """Test creating ovsbond with AutoStart.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + "autostart": True, # This should fail - autostart not supported for OVSBond + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "Parameters autostart are not valid for interface type 'OVSBond'" + in result["msg"] + ) + + def test_create_valid_ovsbond(self): + """Test creating a valid ovsbond.""" + created_config = { + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + "ovs_tag": 100, + "ovs_options": "updelay=5000", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[None, created_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "create_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + "ovs_tag": 100, + "ovs_options": "updelay=5000", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "bond0" + assert "created" in result["msg"].lower() + + def test_update_valid_ovsbond_modes_and_slaves(self): + """Test updating a valid ovsbond modes and slaves.""" + existing_config = { + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + } + + updated_config = { + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "balance-slb", + "ovs_bonds": "eth2 eth3", + "ovs_bridge": "ovsbr0", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + side_effect=[existing_config, updated_config], + ): + with patch.object( + self.module.ProxmoxNetworkManager, "update_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "balance-slb", + "ovs_bonds": "eth2 eth3", + "ovs_bridge": "ovsbr0", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "bond0" + assert "updated" in result["msg"].lower() + + def test_delete_valid_ovsbond_and_options_and_comments(self): + """Test deleting a valid ovsbond and options and comments.""" + existing_config = { + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + "ovs_tag": 100, + "ovs_options": "updelay=5000", + "comments": "Test OVS bond", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=existing_config, + ): + with patch.object( + self.module.ProxmoxNetworkManager, "delete_interface", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "state": "absent", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "deleted" in result["msg"].lower() + + def test_apply_network_changes(self): + """Test applying staged network changes.""" + mock_pending_changes = """--- /etc/network/interfaces ++++ /etc/network/interfaces +@@ -10,6 +10,12 @@ + # The primary network interface + auto eth0 + iface eth0 inet dhcp ++ ++# Bridge interface ++auto vmbr1 ++iface vmbr1 inet static ++ address 192.168.2.1/24 ++ bridge_ports eth1""" + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_network_changes", + return_value=mock_pending_changes, + ): + with patch.object( + self.module.ProxmoxNetworkManager, "apply_network", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "state": "apply", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "applied" in result["msg"].lower() + + def test_apply_network_no_pending_changes(self): + """Test applying when there are no pending changes.""" + with patch.object( + self.module.ProxmoxNetworkManager, + "get_network_changes", + return_value=None, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "state": "apply", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is False + assert "No staged network configuration changes to apply" in result["msg"] + + def test_revert_network_changes(self): + """Test reverting staged network changes.""" + with patch.object( + self.module.ProxmoxNetworkManager, "revert_network", return_value=True + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "state": "revert", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "reverted" in result["msg"].lower() + + def test_revert_network_no_pending_changes(self): + """Test reverting when there are no pending changes.""" + with patch.object( + self.module.ProxmoxNetworkManager, "revert_network", return_value=False + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "state": "revert", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "reverted" in result["msg"].lower() + + def test_apply_network_check_mode(self): + """Test applying network changes in check mode.""" + mock_pending_changes = """--- /etc/network/interfaces ++++ /etc/network/interfaces +@@ -10,6 +10,12 @@ + # The primary network interface + auto eth0 + iface eth0 inet dhcp ++ ++# Bridge interface ++auto vmbr1 ++iface vmbr1 inet static ++ address 192.168.2.1/24 ++ bridge_ports eth1""" + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_network_changes", + return_value=mock_pending_changes, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "state": "apply", + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "may be applied" in result["msg"].lower() + + def test_revert_network_check_mode(self): + """Test reverting network changes in check mode.""" + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "state": "revert", + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "may be reverted" in result["msg"].lower() + + def test_create_bridge_check_mode(self): + """Test creating a bridge interface in check mode.""" + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=None, # Interface doesn't exist + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr1", + "iface_type": "bridge", + "bridge_ports": "eth1", + "cidr": "192.168.2.1/24", + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vmbr1" + assert "would be created" in result["msg"].lower() + + def test_update_bridge_check_mode(self): + """Test updating a bridge interface in check mode.""" + existing_config = { + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0", + "cidr": "192.168.1.1/24", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=existing_config, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + "mtu": 9000, + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vmbr0" + assert "would be updated" in result["msg"].lower() + + def test_delete_interface_check_mode(self): + """Test deleting an interface in check mode.""" + existing_config = { + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0", + } + + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=existing_config, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "state": "absent", + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert "would be deleted" in result["msg"].lower() + + def test_create_bond_check_mode(self): + """Test creating a bond interface in check mode.""" + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=None, # Interface doesn't exist + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "active-backup", + "bond_primary": "eth0", + "slaves": "eth0 eth1", + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "bond0" + assert "would be created" in result["msg"].lower() + + def test_create_vlan_check_mode(self): + """Test creating a VLAN interface in check mode.""" + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=None, # Interface doesn't exist + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vlan100", + "iface_type": "vlan", + "vlan_raw_device": "eth0", + "cidr": "192.168.100.0/24", + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "vlan100" + assert "would be created" in result["msg"].lower() + + def test_create_ovsbond_check_mode(self): + """Test creating an OVS bond interface in check mode.""" + with patch.object( + self.module.ProxmoxNetworkManager, + "get_interface_config", + return_value=None, # Interface doesn't exist + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + "_ansible_check_mode": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert result["changed"] is True + assert result["interface"]["iface"] == "bond0" + assert "would be created" in result["msg"].lower() + + def test_get_all_interfaces_direct_call(self): + """Test the get_all_interfaces method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + + with patch.object( + manager.proxmox_api.nodes, + "return_value.network.return_value.get", + return_value=EXISTING_NETWORK_OUTPUT, + ): + result = manager.get_all_interfaces() + assert result is not None + assert len(result) > 0 + # Verify conversion from API format to Ansible format + for interface in result: + assert "iface" in interface + assert ( + "iface_type" in interface + ) # API 'type' becomes 'iface_type' in Ansible format + + def test_validate_params_direct_call(self): + """Test the validate_params method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr1", + "iface_type": "bridge", + "bridge_ports": "eth1", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager.validate_params() + assert isinstance(result, list) + + def test_validate_interface_name_direct_call(self): + """Test the validate_interface_name method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr1", + "iface_type": "bridge", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager.validate_interface_name() + assert isinstance(result, list) + + def test_validate_parameter_combinations_direct_call(self): + """Test the validate_parameter_combinations method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr1", + "iface_type": "bridge", + "bridge_ports": "eth1", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager.validate_parameter_combinations() + assert isinstance(result, list) + + def test_build_interface_data_direct_call(self): + """Test the _build_interface_data method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr1", + "iface_type": "bridge", + "bridge_ports": "eth1", + "cidr": "192.168.1.0/24", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager._build_interface_data() + assert isinstance(result, dict) + + def test_has_differences_direct_call(self): + """Test the _has_differences method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0 eth1", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + current_config = { + "iface": "vmbr0", + "iface_type": "bridge", + "bridge_ports": "eth0", + } + result = manager._has_differences(current_config) + assert isinstance(result, bool) + + def test_get_network_changes_direct_call(self): + """Test the get_network_changes method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + + with patch.object( + manager.proxmox_api.nodes, + "return_value.network.return_value.get", + return_value="mock diff output", + ): + result = manager.get_network_changes() + assert result is not None + + def test_validate_bridge_params_direct_call(self): + """Test the validate_bridge_params method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr1", + "iface_type": "bridge", + "bridge_ports": "eth1", + "bridge_vlan_aware": True, + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager.validate_bridge_params() + assert isinstance(result, list) + + def test_validate_bond_params_direct_call(self): + """Test the validate_bond_params method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "bond", + "bond_mode": "active-backup", + "bond_primary": "eth0", + "slaves": "eth0 eth1", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager.validate_bond_params() + assert isinstance(result, list) + + def test_validate_vlan_params_direct_call(self): + """Test the validate_vlan_params method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vlan100", + "iface_type": "vlan", + "vlan_raw_device": "eth0", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager.validate_vlan_params() + assert isinstance(result, list) + + def test_validate_ovs_bond_params_direct_call(self): + """Test the validate_ovs_bond_params method directly.""" + # Create a mock module for the manager + mock_module = type( + "MockModule", + (), + { + "fail_json": lambda self, **kwargs: None, + "params": { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + "iface_type": "OVSBond", + "bond_mode": "active-backup", + "ovs_bonds": "eth0 eth1", + "ovs_bridge": "ovsbr0", + }, + }, + )() + + manager = self.module.ProxmoxNetworkManager(mock_module) + result = manager.validate_ovs_bond_params() + assert isinstance(result, list) diff --git a/tests/unit/plugins/modules/test_proxmox_node_network_info.py b/tests/unit/plugins/modules/test_proxmox_node_network_info.py new file mode 100644 index 00000000..909740bd --- /dev/null +++ b/tests/unit/plugins/modules/test_proxmox_node_network_info.py @@ -0,0 +1,425 @@ +# -*- coding: utf-8 -*- +# +# Copyright (c) 2025, aleskxyz +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +from unittest.mock import patch +import pytest + +proxmoxer = pytest.importorskip("proxmoxer") + +from ansible_collections.community.proxmox.plugins.modules import ( + proxmox_node_network_info, +) +from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import ( + AnsibleExitJson, + AnsibleFailJson, + ModuleTestCase, + set_module_args, +) +import ansible_collections.community.proxmox.plugins.module_utils.proxmox as proxmox_utils + +# Mock API response from Proxmox node network endpoint +RAW_NETWORK_OUTPUT = [ + { + "iface": "eth0", + "type": "eth", + "active": 1, + "autostart": 1, + "mtu": 1500, + "method": "manual", + "families": ["inet"], + }, + { + "iface": "vmbr0", + "type": "bridge", + "active": 1, + "autostart": 1, + "mtu": 1500, + "method": "static", + "families": ["inet"], + "bridge_ports": "eth0", + "address": "192.168.1.1", + "netmask": "255.255.255.0", + }, + { + "iface": "bond0", + "type": "bond", + "active": 0, + "autostart": 1, + "mtu": 1500, + "method": "manual", + "families": ["inet"], + "slaves": "eth1 eth2", + "bond_mode": "active-backup", + }, +] + +# Expected output after boolean conversion (0/1 -> True/False) +EXPECTED_NETWORK_OUTPUT = [ + { + "iface": "eth0", + "type": "eth", + "active": True, + "autostart": True, + "mtu": 1500, + "method": "manual", + "families": ["inet"], + }, + { + "iface": "vmbr0", + "type": "bridge", + "active": True, + "autostart": True, + "mtu": 1500, + "method": "static", + "families": ["inet"], + "bridge_ports": "eth0", + "address": "192.168.1.1", + "netmask": "255.255.255.0", + }, + { + "iface": "bond0", + "type": "bond", + "active": False, + "autostart": True, + "mtu": 1500, + "method": "manual", + "families": ["inet"], + "slaves": "eth1 eth2", + "bond_mode": "active-backup", + }, +] + +# Mock pending changes response +MOCK_PENDING_CHANGES = """--- /etc/network/interfaces ++++ /etc/network/interfaces +@@ -10,6 +10,12 @@ + # The primary network interface + auto eth0 + iface eth0 inet dhcp ++ ++# Bridge interface ++auto vmbr1 ++iface vmbr1 inet static ++ address 192.168.2.1/24 ++ bridge_ports eth1""" + + +class TestProxmoxNodeNetworkInfo(ModuleTestCase): + def setUp(self): + super(TestProxmoxNodeNetworkInfo, self).setUp() + proxmox_utils.HAS_PROXMOXER = True + self.module = proxmox_node_network_info + + self.connect_mock = patch( + "ansible_collections.community.proxmox.plugins.module_utils.proxmox.ProxmoxAnsible._connect", + ).start() + + mock_nodes = self.connect_mock.return_value.nodes + mock_node_obj = mock_nodes.return_value + mock_nodes.side_effect = lambda node=None: mock_node_obj + + mock_network_obj = mock_node_obj.network.return_value + mock_node_obj.network = mock_network_obj + mock_node_obj.network.return_value = mock_network_obj + mock_network_obj.get.return_value = RAW_NETWORK_OUTPUT + + mock_nodes.get.return_value = [{"node": "pve"}] + + mock_network_obj.get.side_effect = lambda type=None: [ + interface + for interface in RAW_NETWORK_OUTPUT + if type is None or interface["type"] == type + ] + + def tearDown(self): + self.connect_mock.stop() + super(TestProxmoxNodeNetworkInfo, self).tearDown() + + def test_basic_network_info(self): + """Test basic network interface retrieval.""" + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + assert result["proxmox_node_networks"] == EXPECTED_NETWORK_OUTPUT + + def test_filter_by_iface(self): + """Test filtering by specific interface name.""" + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + assert len(result["proxmox_node_networks"]) == 1 + assert result["proxmox_node_networks"][0]["iface"] == "vmbr0" + + def test_filter_by_iface_type(self): + """Test filtering by interface type.""" + mock_network_obj = ( + self.connect_mock.return_value.nodes.return_value.network.return_value + ) + mock_network_obj.get.side_effect = lambda type=None: [ + interface + for interface in RAW_NETWORK_OUTPUT + if type is None or interface["type"] == type + ] + + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface_type": "bridge", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + for network in result["proxmox_node_networks"]: + assert network["type"] == "bridge" + + def test_check_changes_with_pending_changes(self): + """Test check_changes functionality with pending changes.""" + + class MockResponse: + def json(self): + return { + "data": RAW_NETWORK_OUTPUT, + "changes": MOCK_PENDING_CHANGES, + "success": 1, + } + + def raise_for_status(self, *args, **kwargs): + pass + + mock_response = MockResponse() + + with patch.object( + self.connect_mock.return_value._store["session"], + "request", + return_value=mock_response, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "check_changes": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + assert result["pending_changes"] == MOCK_PENDING_CHANGES + assert result["has_pending_changes"] is True + + def test_check_changes_without_pending_changes(self): + """Test check_changes functionality without pending changes.""" + + class MockResponse: + def json(self): + return {"data": RAW_NETWORK_OUTPUT, "changes": None, "success": 1} + + def raise_for_status(self, *args, **kwargs): + pass + + mock_response = MockResponse() + + with patch.object( + self.connect_mock.return_value._store["session"], + "request", + return_value=mock_response, + ): + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "check_changes": True, + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + assert result["pending_changes"] is None + assert result["has_pending_changes"] is False + + def test_invalid_parameter_combination(self): + """Test that check_changes cannot be used with iface or iface_type.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "check_changes": True, + "iface": "eth0", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "check_changes cannot be used with iface or iface_type parameters" + in result["msg"] + ) + + def test_node_not_found(self): + """Test error handling when node doesn't exist.""" + with patch.object( + self.module.ProxmoxNodeNetworkInfoAnsible, "get_node", return_value=None + ): + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "nonexistent", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert ( + "Node 'nonexistent' not found in the Proxmox cluster" in result["msg"] + ) + + def test_boolean_conversion(self): + """Test that boolean values are properly converted.""" + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "bond0", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + bond0_interface = result["proxmox_node_networks"][0] + assert bond0_interface["iface"] == "bond0" + assert bond0_interface["active"] is False + assert bond0_interface["autostart"] is True + + def test_node_not_specified(self): + """Test error handling when node parameter is not specified.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + {"api_host": "host", "api_user": "user", "api_password": "password"} + ): + self.module.main() + + result = exc_info.value.args[0] + assert "missing required arguments: node" in result["msg"] + + def test_iface_specified_but_not_found(self): + """Test when iface is specified but the interface doesn't exist.""" + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "nonexistent_interface", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + assert result["proxmox_node_networks"] == [] + + def test_iface_type_specified_but_not_found(self): + """Test when iface_type is specified but no interfaces of that type exist.""" + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface_type": "OVSPort", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + assert result["proxmox_node_networks"] == [] + + def test_iface_type_invalid(self): + """Test when iface_type is specified with an invalid value.""" + with pytest.raises(AnsibleFailJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface_type": "invalid_type", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert "value of iface_type must be one of:" in result["msg"] + + def test_combination_iface_and_iface_type_not_found(self): + """Test when both iface and iface_type are specified but combination doesn't exist.""" + with pytest.raises(AnsibleExitJson) as exc_info: + with set_module_args( + { + "api_host": "host", + "api_user": "user", + "api_password": "password", + "node": "pve", + "iface": "vmbr0", + "iface_type": "eth", + } + ): + self.module.main() + + result = exc_info.value.args[0] + assert not result["changed"] + assert result["proxmox_node_networks"] == []