tests/integration/fixtures/network.py

   1 import pytest
   2 import pyroute2
   3 import struct
   4 import socket
   5 import fcntl
   6 import time
   7 from .namespaces import Namespace
   8
   9
  10 def int_to_mac(c):
  11     """Turn an int into a MAC address."""
  12     return ":".join(('{:02x}',)*6).format(
  13         *struct.unpack('BBBBBB', c.to_bytes(6, byteorder='big')))
  14
  15
  16 class LinksFactory(object):
  17     """A factory for veth pair of interfaces and other L2 stuff.
  18
  19     Each veth interfaces will get named ethX with X strictly
  20     increasing at each call.
  21
  22     """
  23
  24     def __init__(self):
  25         # We create all those links in a dedicated namespace to avoid
  26         # conflict with other namespaces.
  27         self.ns = Namespace('net')
  28         self.count = 0
  29
  30     def __call__(self, *args, **kwargs):
  31         return self.veth(*args, **kwargs)
  32
  33     def veth(self, ns1, ns2, sleep=0, mtu=1500):
  34         """Create a veth pair between two namespaces."""
  35         with self.ns:
  36             # First, create a link
  37             first = 'eth{}'.format(self.count)
  38             second = 'eth{}'.format(self.count + 1)
  39             ipr = pyroute2.IPRoute()
  40             ipr.link('add',
  41                      ifname=first,
  42                      peer=second,
  43                      kind='veth')
  44             idx = [ipr.link_lookup(ifname=x)[0]
  45                    for x in (first, second)]
  46
  47             # Set an easy to remember MAC address
  48             ipr.link('set', index=idx[0],
  49                      address=int_to_mac(self.count + 1))
  50             ipr.link('set', index=idx[1],
  51                      address=int_to_mac(self.count + 2))
  52
  53             # Set MTU
  54             ipr.link('set', index=idx[0], mtu=mtu)
  55             ipr.link('set', index=idx[1], mtu=mtu)
  56
  57             # Then, move each to the target namespace
  58             ipr.link('set', index=idx[0], net_ns_fd=ns1.fd('net'))
  59             ipr.link('set', index=idx[1], net_ns_fd=ns2.fd('net'))
  60
  61             # And put them up
  62             with ns1:
  63                 ipr = pyroute2.IPRoute()
  64                 ipr.link('set', index=idx[0], state='up')
  65             time.sleep(sleep)
  66             with ns2:
  67                 ipr = pyroute2.IPRoute()
  68                 ipr.link('set', index=idx[1], state='up')
  69
  70             self.count += 2
  71
  72     def bridge(self, name, *ifaces):
  73         """Create a bridge."""
  74         ipr = pyroute2.IPRoute()
  75         # Create the bridge
  76         ipr.link('add',
  77                  ifname=name,
  78                  kind='bridge')
  79         idx = ipr.link_lookup(ifname=name)[0]
  80         # Attach interfaces
  81         for iface in ifaces:
  82             port = ipr.link_lookup(ifname=iface)[0]
  83             ipr.link('set', index=port, master=idx)
  84         # Put the bridge up
  85         ipr.link('set', index=idx, state='up')
  86         return idx
  87
  88     def _bond_or_team(self, kind, name, *ifaces):
  89         """Create a bond or a team."""
  90         ipr = pyroute2.RawIPRoute()
  91         # Create the bond
  92         ipr.link('add',
  93                  ifname=name,
  94                  kind=kind)
  95         idx = ipr.link_lookup(ifname=name)[0]
  96         # Attach interfaces
  97         for iface in ifaces:
  98             slave = ipr.link_lookup(ifname=iface)[0]
  99             ipr.link('set', index=slave, state='down')
 100             ipr.link('set', index=slave, master=idx)
 101         # Put the bond up
 102         ipr.link('set', index=idx, state='up')
 103         return idx
 104
 105     def team(self, name, *ifaces):
 106         """Create a team."""
 107         # Unfortunately, pyroute2 will try to run teamd too. This
 108         # doesn't work.
 109         return self._bond_or_team("team", name, *ifaces)
 110
 111     def bond(self, name, *ifaces):
 112         """Create a bond."""
 113         return self._bond_or_team("bond", name, *ifaces)
 114
 115     def dummy(self, name):
 116         """Create a dummy interface."""
 117         ipr = pyroute2.IPRoute()
 118         ipr.link('add', ifname=name, kind='dummy')
 119         idx = ipr.link_lookup(ifname=name)[0]
 120         ipr.link('set', index=idx, state='up')
 121         return idx
 122
 123     def vlan(self, name, id, iface):
 124         """Create a VLAN."""
 125         ipr = pyroute2.IPRoute()
 126         idx = ipr.link_lookup(ifname=iface)[0]
 127         ipr.link('add',
 128                  ifname=name,
 129                  kind='vlan',
 130                  vlan_id=id,
 131                  link=idx)
 132         idx = ipr.link_lookup(ifname=name)[0]
 133         ipr.link('set', index=idx, state='up')
 134         return idx
 135
 136     def bridge_vlan(self, iface, vid, tagged=True, pvid=False, remove=False):
 137         ipr = pyroute2.IPRoute()
 138         idx = ipr.link_lookup(ifname=iface)[0]
 139         flags = []
 140         if not tagged:
 141             flags.append('untagged')
 142         if pvid:
 143             flags.append('pvid')
 144         if not remove:
 145             ipr.vlan_filter('del', index=idx, vlan_info={"vid": 1})
 146         ipr.vlan_filter('add' if not remove else 'del', index=idx,
 147                         vlan_info={'vid': vid,
 148                                    'flags': flags})
 149
 150     def up(self, name):
 151         ipr = pyroute2.IPRoute()
 152         idx = ipr.link_lookup(ifname=name)[0]
 153         ipr.link('set', index=idx, state='up')
 154
 155     def down(self, name):
 156         ipr = pyroute2.IPRoute()
 157         idx = ipr.link_lookup(ifname=name)[0]
 158         ipr.link('set', index=idx, state='down')
 159
 160     def remove(self, name):
 161         ipr = pyroute2.IPRoute()
 162         idx = ipr.link_lookup(ifname=name)[0]
 163         ipr.link('del', index=idx)
 164
 165     def unbridge(self, bridgename, name):
 166         ipr = pyroute2.IPRoute()
 167         idx = ipr.link_lookup(ifname=name)[0]
 168         s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
 169         ifr = struct.pack("16si", b"br42", idx)
 170         fcntl.ioctl(s,
 171                     0x89a3,     # SIOCBRDELIF
 172                     ifr)
 173         s.close()
 174
 175
 176 @pytest.fixture
 177 def links():
 178     return LinksFactory()