lib.switch

  1"""Managed switches of a lab: grading helpers around ``Grade.test_switch()``.
  2
  3A network declared with ``'mode': 'managed'`` in ``_network_specs`` is a ``vde_switch`` with VLANs
  4and a management console.  ``get_switch_ports(grade, 'lan')`` reads its port and VLAN tables and
  5gives the port of each machine; ``get_switch_vlans(grade, 'lan')`` gives the members of each
  6VLAN.  ``parse_switch_ports`` / ``parse_switch_vlans`` are the pure parsers of the
  7``port/allprint`` and ``vlan/allprint`` outputs (fixtures in ``tests/mock_data/switch/``:
  8VDE switch 2.3.3 behind the Kathara network plugin, which names each port's endpoint
  9``kathara <machine>:eth<N>``).
 10"""
 11import re
 12from typing import Any, Dict, List
 13
 14SWITCH_PORTS_CMD = "port/allprint"
 15SWITCH_VLANS_CMD = "vlan/allprint"
 16
 17_PORT_RE = re.compile(r"^Port (\d+) untagged_vlan=(\d+) (IN)?ACTIVE")
 18_ENDPOINT_RE = re.compile(r"^\s*-- endpoint ID \d+ module [^:]*: (.*?)(?: user=\S+ pid=\d+.*)?$")
 19_VLAN_RE = re.compile(r"^VLAN (\d+)")
 20_VLAN_PORT_RE = re.compile(r"^\s*-- Port (\d+) tagged=(\d)")
 21_MACHINE_ENDPOINT_RE = re.compile(r"^kathara (\S+):(eth\d+)$")
 22
 23
 24def parse_switch_vlans(vlans_output: str) -> Dict[int, Dict[str, List[int]]]:
 25    """Parse a ``vlan/allprint`` output: ``{vlan: {'untagged': [port, ...], 'tagged': [port, ...]}}``.
 26
 27    VLAN 0 is the default VLAN of the switch (ports without configuration)."""
 28    vlans: Dict[int, Dict[str, List[int]]] = {}
 29    current = None
 30    for line in (vlans_output or '').splitlines():
 31        m = _VLAN_RE.match(line)
 32        if m:
 33            current = vlans.setdefault(int(m.group(1)), {'untagged': [], 'tagged': []})
 34            continue
 35        m = _VLAN_PORT_RE.match(line)
 36        if m and current is not None:
 37            current['tagged' if m.group(2) == '1' else 'untagged'].append(int(m.group(1)))
 38    return vlans
 39
 40
 41def parse_switch_ports(ports_output: str, vlans_output: str = '') -> Dict[int, Dict[str, Any]]:
 42    """Parse a ``port/allprint`` output (and the ``vlan/allprint`` one for the tagged VLANs).
 43
 44    Returns ``{port number: {...}}`` with ``vlan`` (VLAN of the untagged frames, 0 = default),
 45    ``tagged_vlans`` (sorted), ``active`` (something is plugged), ``endpoints`` (descriptions of
 46    what is plugged) and, when it is the interface of a machine, ``machine`` and ``interface``
 47    (``'pc1'``, ``'eth0'``; ``None`` otherwise).
 48    """
 49    ports: Dict[int, Dict[str, Any]] = {}
 50    port = None
 51    for line in (ports_output or '').splitlines():
 52        m = _PORT_RE.match(line)
 53        if m:
 54            port = {'vlan': int(m.group(2)), 'tagged_vlans': [], 'active': not m.group(3),
 55                    'endpoints': [], 'machine': None, 'interface': None}
 56            ports[int(m.group(1))] = port
 57            continue
 58        m = _ENDPOINT_RE.match(line)
 59        if m and port is not None:
 60            description = m.group(1).strip()
 61            port['endpoints'].append(description)
 62            machine = _MACHINE_ENDPOINT_RE.match(description)
 63            if machine:
 64                port['machine'], port['interface'] = machine.group(1), machine.group(2)
 65    for vlan, members in parse_switch_vlans(vlans_output).items():
 66        for number in members['tagged']:
 67            if number in ports:
 68                ports[number]['tagged_vlans'].append(vlan)
 69    for port in ports.values():
 70        port['tagged_vlans'].sort()
 71    return ports
 72
 73
 74def ports_by_machine(ports: Dict[int, Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
 75    """``{machine: {'port', 'interface', 'vlan', 'tagged_vlans'}}`` from :func:`parse_switch_ports`."""
 76    return {port['machine']: {'port': number, 'interface': port['interface'], 'vlan': port['vlan'],
 77                              'tagged_vlans': list(port['tagged_vlans'])}
 78            for number, port in sorted(ports.items()) if port['machine']}
 79
 80
 81def get_switch_ports(grade, network: str, step: int = 1, allow_error: bool = False) -> Dict[str, Dict[str, Any]]:
 82    """Ports of the managed switch *network*, by machine name.
 83
 84    ``{'pc1': {'port': 2, 'interface': 'eth0', 'vlan': 10, 'tagged_vlans': []}, ...}``: ``vlan``
 85    is the VLAN of the untagged frames (0 = default VLAN of the switch), ``tagged_vlans`` the
 86    VLANs of a trunk port.  ``{}`` until the tests have run or when the console failed.
 87    """
 88    ports_out, ports_code = grade.test_switch(network, SWITCH_PORTS_CMD, step=step, allow_error=allow_error)
 89    vlans_out, vlans_code = grade.test_switch(network, SWITCH_VLANS_CMD, step=step, allow_error=allow_error)
 90    if ports_code != 0:
 91        return {}
 92    return ports_by_machine(parse_switch_ports(ports_out, vlans_out if vlans_code == 0 else ''))
 93
 94
 95def get_switch_vlans(grade, network: str, step: int = 1, allow_error: bool = False) -> Dict[int, Dict[str, List[int]]]:
 96    """VLANs of the managed switch *network*: ``{vlan: {'untagged': [ports], 'tagged': [ports]}}``
 97    (port numbers, see :func:`get_switch_ports` for the machines).  ``{}`` until the tests have
 98    run or when the console failed."""
 99    out, code = grade.test_switch(network, SWITCH_VLANS_CMD, step=step, allow_error=allow_error)
100    return parse_switch_vlans(out) if code == 0 else {}