62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
|
from typing import List, Tuple
|
||
|
|
|
||
|
|
from ansible_collections.finallycoffee.proxmox.plugins.module_utils.common import _proxmox_request, ProxmoxAuthInfo
|
||
|
|
|
||
|
|
|
||
|
|
PM_ACCEPTED_IFACE_TYPES = [
|
||
|
|
'bridge',
|
||
|
|
'bond',
|
||
|
|
'eth',
|
||
|
|
'alias',
|
||
|
|
'vlan',
|
||
|
|
'OVSBridge',
|
||
|
|
'OVSBond',
|
||
|
|
'OVSPort',
|
||
|
|
'OVSIntPort',
|
||
|
|
'unknown',
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def get_interfaces(auth_info: ProxmoxAuthInfo, node: str) -> [dict]:
|
||
|
|
realm_answer = _proxmox_request('get', f"/nodes/{node}/network", auth_info).json()
|
||
|
|
return realm_answer['data']
|
||
|
|
|
||
|
|
|
||
|
|
def set_interface_config(auth_info: ProxmoxAuthInfo, node: str, name: str, iface_type: str, config: dict[str, str], dry_run: bool, state: str = 'present') -> Tuple[dict, dict]:
|
||
|
|
existing_iface_data = None
|
||
|
|
iface_data = None
|
||
|
|
existing_iface = _proxmox_request('get', f"/nodes/{node}/network/{name}", auth_info)
|
||
|
|
if existing_iface.ok:
|
||
|
|
existing_iface_data = existing_iface.json()['data']
|
||
|
|
if state == 'absent':
|
||
|
|
iface = _delete_iface(auth_info, node, name, dry_run)
|
||
|
|
return existing_iface_data, iface_data
|
||
|
|
else:
|
||
|
|
iface = _upsert_iface(auth_info, node, name,
|
||
|
|
iface_type, config, (existing_iface_data is not None), dry_run)
|
||
|
|
return existing_iface_data, iface
|
||
|
|
|
||
|
|
|
||
|
|
def _upsert_iface(auth_info: ProxmoxAuthInfo, node: str, name: str, iface_type: str, config: dict[str, str], update: bool = False, dry_run: bool = False) -> dict[str, str]:
|
||
|
|
extra_config = {'node': node, 'iface': name, 'type': iface_type}
|
||
|
|
full_iface_spec = config | extra_config
|
||
|
|
if not dry_run:
|
||
|
|
res = _proxmox_request('put' if update else 'post', f"/nodes/{node}/network", auth_info, full_iface_spec)
|
||
|
|
if res.ok:
|
||
|
|
return res.json()['data'] | extra_config
|
||
|
|
else:
|
||
|
|
res.raise_for_status()
|
||
|
|
else:
|
||
|
|
return full_iface_spec
|
||
|
|
|
||
|
|
|
||
|
|
def _delete_iface(auth_info: ProxmoxAuthInfo, node: str, name: str, dry_run: bool = False) -> dict[str, str]:
|
||
|
|
if not dry_run:
|
||
|
|
res = _proxmox_request('delete', f"/nodes/{node}/network/{name}", auth_info)
|
||
|
|
if res.ok:
|
||
|
|
return {}
|
||
|
|
else:
|
||
|
|
res.raise_for_status()
|
||
|
|
else:
|
||
|
|
return {}
|