ci/integration: use unified fixtures

pyroute2/fixtures/iproute.py

@@ -6,12 +6,13 @@
 import pytest
 import pytest_asyncio
 
-from pyroute2 import netns
+from pyroute2 import NDB, netns
 from pyroute2.common import uifname
 from pyroute2.iproute.linux import AsyncIPRoute, IPRoute
 from pyroute2.netlink.exceptions import NetlinkError
+from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg
 
-TestInterface = namedtuple('TestInterface', ('index', 'ifname'))
+TestInterface = namedtuple('TestInterface', ('index', 'ifname', 'address'))
 
 
 @pytest.fixture(name='nsname')
@@ -32,24 +33,40 @@ def _nsname() -> Generator[str]:
         pass
 
 
-@pytest.fixture(name='test_link')
-def _test_link(nsname: str) -> Generator[TestInterface]:
-    '''Create a test interface in the test netns and yield TestInterface.
+@pytest.fixture(name='test_link_ifinfmsg')
+def _test_link_ifinfmsg(nsname: str) -> Generator[ifinfmsg]:
+    '''Create a test interface in the test netns and yield ifinfmsg.
 
     Remove the interface on exit.
     '''
     ifname = uifname()
     with IPRoute(netns=nsname) as ipr:
         ipr.link('add', ifname=ifname, kind='dummy', state='up')
         (link,) = ipr.poll(ipr.link, 'dump', ifname=ifname, timeout=5)
-        yield TestInterface(index=link.get('index'), ifname=link.get('ifname'))
+        yield link
         try:
             ipr.link('del', index=link.get('index'))
         except NetlinkError as e:
             if e.code != errno.ENODEV:
                 raise
 
 
+@pytest.fixture(name='test_link')
+def _test_link(test_link_ifinfmsg: ifinfmsg) -> Generator[TestInterface]:
+    '''Yield test interface spec as TestInterface.'''
+    yield TestInterface(
+        index=test_link_ifinfmsg.get('index'),
+        ifname=test_link_ifinfmsg.get('ifname'),
+        address=test_link_ifinfmsg.get('address'),
+    )
+
+
+@pytest.fixture(name='test_link_address')
+def _test_link_address(test_link: TestInterface) -> Generator[str]:
+    '''Yield test interface address.'''
+    yield test_link.address
+
+
 @pytest.fixture(name='test_link_index')
 def _test_link_index(test_link: TestInterface) -> Generator[int]:
     '''Yield test interface index.'''
@@ -133,3 +150,9 @@ def _sync_context(
     sync_ipr: IPRoute, test_link: TestInterface
 ) -> Generator[TestContext]:
     yield TestContext(sync_ipr, test_link)
+
+
+@pytest.fixture(name='ndb')
+def _ndb(nsname: str) -> Generator[NDB]:
+    with NDB(sources=[{'target': 'localhost', 'netns': nsname}]) as ndb:
+        yield ndb

tests/test_integration/conftest.py

@@ -1,46 +1 @@
-import pytest
-
-from pyroute2 import NDB, IPRoute, netns
-from pyroute2.common import uifname
-
-
-@pytest.fixture
-def nsname(request, tmpdir):
-    name = uifname()
-    netns.create(name)
-    yield name
-    try:
-        netns.remove(name)
-    except:
-        pass
-
-
-@pytest.fixture
-def link(request, tmpdir, nsname):
-    name = uifname()
-    with IPRoute(netns=nsname) as ipr:
-        ipr.link('add', ifname=name, kind='dummy')
-        (link,) = ipr.link('get', ifname=name)
-        link.netns = nsname
-        yield link
-        try:
-            ipr.link('del', index=link['index'])
-        except:
-            pass
-
-
-@pytest.fixture
-def ifname(link):
-    return link.get('ifname')
-
-
-@pytest.fixture
-def ipr(nsname):
-    with IPRoute(netns=nsname) as ipr:
-        yield ipr
-
-
-@pytest.fixture
-def ndb(nsname):
-    with NDB(sources=[{'target': 'localhost', 'netns': nsname}]) as ndb:
-        yield ndb
+pytest_plugins = ["pyroute2.fixtures.iproute", "pyroute2.fixtures.plan9"]

tests/test_integration/test_kuryr.py

@@ -13,8 +13,8 @@
 
 
 @pytest.fixture
-def ipdb(link):
-    with IPDB(sources=[{'target': 'localhost', 'netns': link.netns}]) as ip:
+def ipdb(nsname):
+    with IPDB(sources=[{'target': 'localhost', 'netns': nsname}]) as ip:
         yield ip
 
 
@@ -30,13 +30,13 @@ def test_exception_types(exc):
     assert issubclass(exc, Exception)
 
 
-def test_ipdb_create_exception(link, ipdb):
+def test_ipdb_create_exception(test_link_ifname, ipdb):
     with pytest.raises(pyroute2.CreateException):
-        ipdb.create(ifname=link.get('ifname'), kind='dummy').commit()
+        ipdb.create(ifname=test_link_ifname, kind='dummy').commit()
 
 
-def test_ipdb_create_reuse(link, ipdb):
-    ipdb.create(ifname=link.get('ifname'), kind='dummy', reuse=True).commit()
+def test_ipdb_create_reuse(test_link_ifname, ipdb):
+    ipdb.create(ifname=test_link_ifname, kind='dummy', reuse=True).commit()
 
 
 @pytest.mark.parametrize(
@@ -48,41 +48,43 @@ def test_ipdb_create_reuse(link, ipdb):
         ('up', [], lambda x: x['flags'] & 1),
     ),
 )
-def test_ipdb_iface_methods(link, ipdb, method, argv, check):
-    iface = ipdb.interfaces[link.get('ifname')]
+def test_ipdb_iface_methods(test_link_ifname, ipdb, method, argv, check):
+    iface = ipdb.interfaces[test_link_ifname]
     with iface:
         getattr(iface, method)(*argv)
     assert check(iface)
 
 
-def test_utils_remove(link, ifname, ipdb):
-    index = ipdb.interfaces.get(ifname, {}).get('index', None)
+def test_utils_remove(nsname, test_link_index, test_link_ifname, ipdb):
+    index = ipdb.interfaces.get(test_link_ifname, {}).get('index', None)
     assert isinstance(index, int)
-    with ipdb.interfaces[ifname] as iface:
+    assert index == test_link_index
+    with ipdb.interfaces[test_link_ifname] as iface:
         iface.remove()
-    assert ifname not in ipdb.interfaces
-    assert not interface_exists(ifname, link.netns, timeout=0.1)
+    assert test_link_ifname not in ipdb.interfaces
+    assert test_link_index not in ipdb.interfaces
+    assert not interface_exists(test_link_ifname, netns=nsname, timeout=0.1)
 
 
-def test_get_iface(link, ifname, ipdb):
-    with ipdb.interfaces[ifname] as link:
+def test_get_iface(test_link_ifname, ipdb):
+    with ipdb.interfaces[test_link_ifname] as link:
         link.set_address(TADDR)
     target = None
     for name, data in ipdb.interfaces.items():
         if data['address'] == TADDR:
             target = data['ifname']
-    assert target == ifname
+    assert target == test_link_ifname
 
 
-def test_create_ipvlan(link, ifname, ipdb):
+def test_create_ipvlan(test_link_index, test_link_ifname, ipdb):
     ipvlname = uifname()
     with ipdb.create(
         ifname=ipvlname,
         kind=KIND,
-        link=ipdb.interfaces[ifname],
+        link=ipdb.interfaces[test_link_ifname],
         ipvlan_mode=IPVLAN_MODE_L2,
     ) as iface:
         assert iface['mode'] == IPVLAN_MODE_L2
         assert iface['ifname'] == ipvlname
-        assert iface['link'] == link.get('index')
+        assert iface['link'] == test_link_index
         assert iface['kind'] == KIND

tests/test_integration/test_lnst.py

@@ -15,12 +15,9 @@
     ],
 )
 def test_interface_manager_dump_link(
-    link, msg_type, dump_method, field, event
+    nsname, msg_type, dump_method, field, event
 ):
-    with (
-        IPRSocket(netns=link.netns) as iprsock,
-        IPRoute(netns=link.netns) as ipr,
-    ):
+    with IPRSocket(netns=nsname) as iprsock, IPRoute(netns=nsname) as ipr:
 
         # bring up loopback
         ipr.link('set', index=1, state='up')

tests/test_integration/test_octavia.py

@@ -21,30 +21,32 @@ def test_amphora_info(nsname):
                     assert isinstance(item[1]['rx_bytes'], int)
 
 
-def test_api_server_plug(nsname, link):
+def test_api_server_plug(nsname, test_link_ifname, test_link_address):
     newname = uifname()
-    with pyroute2.IPRoute(netns=link.netns) as ipr:
-        idx = ipr.link_lookup(address=link.get('address'))[0]
+    with pyroute2.IPRoute(netns=nsname) as ipr:
+        idx = ipr.link_lookup(address=test_link_address)[0]
         ipr.link('set', index=idx, net_ns_fd=nsname, IFLA_IFNAME=newname)
     assert not interface_exists(
-        ifname=link.get('ifname'), netns=link.netns, timeout=0.1
+        ifname=test_link_ifname, netns=nsname, timeout=0.1
     )
     assert interface_exists(ifname=newname, netns=nsname)
 
 
-def test_api_server_mac(link):
-    with pyroute2.IPRoute(netns=link.netns) as ipr:
-        idx = ipr.link_lookup(address=link.get('address'))[0]
+def test_api_server_mac(nsname, test_link_address, test_link_ifname):
+    with pyroute2.IPRoute(netns=nsname) as ipr:
+        idx = ipr.link_lookup(address=test_link_address)[0]
         addr = ipr.get_links(idx)[0]
         for attr in addr['attrs']:
             if attr[0] == 'IFLA_IFNAME':
-                assert attr[1] == link.get('ifname')
+                assert attr[1] == test_link_ifname
 
 
-def test_api_server_attrs(link):
-    attr_dict = dict(link['attrs'])
-    assert attr_dict.get('IFLA_ADDRESS') == link.get('address')
-    assert attr_dict.get('IFLA_IFNAME') == link.get('ifname')
+def test_api_server_attrs(
+    test_link_ifinfmsg, test_link_address, test_link_ifname
+):
+    attr_dict = dict(test_link_ifinfmsg['attrs'])
+    assert attr_dict.get('IFLA_ADDRESS') == test_link_address
+    assert attr_dict.get('IFLA_IFNAME') == test_link_ifname
 
 
 def test_api_server_netns_flags(nsname):
@@ -60,33 +62,37 @@ def test_utils_exception_ref():
     assert e.code == errno.EINVAL
 
 
-def test_utils_link_attr(link, ipr):
-    idx = ipr.link_lookup(ifname=link.get('ifname'))[0]
-    ref = ipr.get_links(idx)[0]
-    assert ref.get('state') == 'down'
-    ipr.link('set', index=idx, state='up', mtu=1000)
-    ipr.poll(ipr.link, 'dump', index=idx, state='up', mtu=1000, timeout=5)
+def test_utils_link_attr(test_link_ifname, sync_ipr):
+    idx = sync_ipr.link_lookup(ifname=test_link_ifname)[0]
+    ref = sync_ipr.get_links(idx)[0]
+    assert ref.get('state') == 'up'
+    sync_ipr.link('set', index=idx, state='down', mtu=1000)
+    sync_ipr.poll(
+        sync_ipr.link, 'dump', index=idx, state='down', mtu=1000, timeout=5
+    )
     with pytest.raises(TimeoutError):
-        ipr.poll(ipr.link, 'dump', index=idx, state='down', timeout=0.1)
+        sync_ipr.poll(
+            sync_ipr.link, 'dump', index=idx, state='up', timeout=0.1
+        )
     with pytest.raises(TimeoutError):
-        ipr.poll(ipr.link, 'dump', index=idx, mtu=1500, timeout=0.1)
+        sync_ipr.poll(sync_ipr.link, 'dump', index=idx, mtu=1500, timeout=0.1)
 
 
-def test_utils_addr_flags(link, ipr, ndb):
-    with ndb.interfaces[link] as i:
+def test_utils_addr_flags(test_link_index, test_link_ifname, sync_ipr, ndb):
+    with ndb.interfaces[test_link_ifname] as i:
         i.set('state', 'up')
         i.add_ip('10.1.2.3/24')
 
-    for addr in ipr.get_addr(index=link.get('index')):
+    for addr in sync_ipr.get_addr(index=test_link_index):
         attrs = dict(addr['attrs'])
         if attrs['IFA_FLAGS'] & ifaddrmsg.IFA_F_PERMANENT:
             break
     else:
         raise NetlinkError(errno.ENOENT, 'no static addresses')
 
 
-def test_utils_route_proto(ndb, ipr):
+def test_utils_route_proto(ndb, sync_ipr):
     with ndb.interfaces['lo'] as i:
         i.set('state', 'up')
-    for route in ipr.get_routes(oif=1):
+    for route in sync_ipr.get_routes(oif=1):
         assert route['proto'] != rt_proto['static']

tests/test_integration/test_os_vif.py

@@ -1,11 +1,11 @@
 from pyroute2.common import uifname
 
 
-def test_impl_lookup(ipr, link):
-    assert len(ipr.link_lookup(ifname=link.get('ifname'))) == 1
+def test_impl_lookup(sync_ipr, test_link_ifname):
+    assert len(sync_ipr.link_lookup(ifname=test_link_ifname)) == 1
 
 
-def test_impl_add_bridge(ipr):
+def test_impl_add_bridge(sync_ipr):
     brname = uifname()
     args = {
         'ifname': brname,
@@ -15,22 +15,28 @@ def test_impl_add_bridge(ipr):
         'IFLA_BR_MCAST_SNOOPING': 0,
         'IFLA_BR_AGEING_TIME': 1500,
     }
-    ipr.link('add', **args)
-    link = [x for x in ipr.poll(ipr.link, 'dump', ifname=brname, timeout=5)][0]
+    sync_ipr.link('add', **args)
+    link = [
+        x
+        for x in sync_ipr.poll(sync_ipr.link, 'dump', ifname=brname, timeout=5)
+    ][0]
     assert link.get(('linkinfo', 'data', 'br_forward_delay')) == 0
     assert link.get(('linkinfo', 'data', 'br_stp_state')) == 0
     assert link.get(('linkinfo', 'data', 'br_mcast_snooping')) == 0
     assert link.get(('linkinfo', 'data', 'br_ageing_time')) == 1500
 
 
-def test_impl_add_vlan(ipr, link):
+def test_impl_add_vlan(sync_ipr, test_link_index):
     vname = uifname()
     args = {
         'ifname': vname,
         'kind': 'vlan',
         'vlan_id': 1001,
-        'link': link.get('index'),
+        'link': test_link_index,
     }
-    ipr.link('add', **args)
-    link = [x for x in ipr.poll(ipr.link, 'dump', ifname=vname, timeout=5)][0]
+    sync_ipr.link('add', **args)
+    link = [
+        x
+        for x in sync_ipr.poll(sync_ipr.link, 'dump', ifname=vname, timeout=5)
+    ][0]
     assert link.get(('linkinfo', 'data', 'vlan_id')) == 1001

tests/test_integration/test_ovn_bgp_agent.py

@@ -16,11 +16,15 @@ def test_iproute_call(method, argv):
 @pytest.mark.parametrize(
     'param,value', (('neigh_suppress', True), ('learning', False))
 )
-def test_brport_set(ipr, ndb, link, param, value):
+def test_brport_set(
+    sync_ipr, ndb, test_link_index, test_link_ifname, param, value
+):
     brname = uifname()
     ndb.interfaces.create(ifname=brname, kind='bridge').add_port(
-        link.get('ifname')
+        test_link_ifname
     ).commit()
     kwarg = {param: value}
-    ipr.brport('set', index=link.get('index'), **kwarg)
-    ipr.poll(ipr.brport, 'dump', index=link.get('index'), timeout=5, **kwarg)
+    sync_ipr.brport('set', index=test_link_index, **kwarg)
+    sync_ipr.poll(
+        sync_ipr.brport, 'dump', index=test_link_index, timeout=5, **kwarg
+    )