ci/integration: simplify fixtures

tests/test_core/conftest.py

@@ -1,5 +1,4 @@
 import errno
-import os
 
 import pytest
 import pytest_asyncio
@@ -9,64 +8,6 @@
 from pyroute2.common import uifname
 
 
-class AsyncIPRouteContext(AsyncIPRoute):
-    def __init__(self, *argv, **kwarg):
-        self.remove_netns_on_exit = False
-        self.registry_ifname = set()
-        if kwarg.get('netns') is True:
-            kwarg['netns'] = uifname()
-            kwarg['flags'] = os.O_CREAT
-            self.remove_netns_on_exit = True
-        super().__init__(*argv, **kwarg)
-
-    def register_temporary_ifname(self, ifname=None):
-        ifname = ifname if ifname is not None else uifname()
-        self.registry_ifname.add(ifname)
-        return ifname
-
-    async def close(self, *argv, **kwarg):
-        for ifname in self.registry_ifname:
-            try:
-                await self.link('del', ifname=ifname)
-            except NetlinkError as e:
-                if e.code != errno.ENODEV:
-                    raise
-        await super().close(*argv, **kwarg)
-        if self.remove_netns_on_exit:
-            netns.remove(self.status['netns'])
-
-
-class SyncIPRouteContext(IPRoute):
-    def __init__(self, *argv, **kwarg):
-        self.remove_netns_on_exit = False
-        self.registry_ifname = set()
-        if kwarg.get('netns') is True:
-            kwarg['netns'] = uifname()
-            kwarg['flags'] = os.O_CREAT
-            self.remove_netns_on_exit = True
-        super().__init__(*argv, **kwarg)
-
-    def register_temporary_ifname(self, ifname=None):
-        ifname = ifname if ifname is not None else uifname()
-        self.registry_ifname.add(ifname)
-        return ifname
-
-    def register_temporary_netns(self, netns=None):
-        netns = netns if netns is not None else uifname()
-        self.registry_netns.add(netns)
-
-    def close(self, *argv, **kwarg):
-        for ifname in self.registry_ifname:
-            try:
-                self.link('del', ifname=ifname)
-            except NetlinkError as e:
-                if e.code != errno.ENODEV:
-                    raise
-        super().close(*argv, **kwarg)
-        if self.remove_netns_on_exit:
-            netns.remove(self.status['netns'])
-
-
 @pytest_asyncio.fixture
 async def p9(request, tmpdir):
     ctx = AsyncPlan9Context()
@@ -75,15 +16,66 @@ async def p9(request, tmpdir):
     await ctx.close()
 
 
+@pytest.fixture
+def nsname():
+    ns = uifname()
+    netns.create(ns)
+    with IPRoute(netns=ns) as ipr:
+        ipr.link('set', index=1, state='up')
+        ipr.poll(ipr.addr, 'dump', address='127.0.0.1', timeout=5)
+    yield ns
+    try:
+        netns.delete(ns)
+    except Exception:
+        pass
+
+
+@pytest.fixture
+def link(nsname):
+    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 link
+        try:
+            ipr.link('del', index=link.get('index'))
+        except NetlinkError as e:
+            if e.code != errno.ENODEV:
+                raise
+
+
+@pytest.fixture
+def index(link):
+    yield link.get('index')
+
+
+@pytest.fixture
+def ifname(link):
+    yield link.get('ifname')
+
+
+@pytest.fixture
+def tmp_link(nsname):
+    ifname = uifname()
+    with IPRoute(netns=nsname) as ipr:
+        yield ifname
+        try:
+            (link,) = ipr.link('get', ifname=ifname)
+            ipr.link('del', index=link.get('index'))
+        except NetlinkError as e:
+            if e.code != errno.ENODEV:
+                raise
+
+
 @pytest_asyncio.fixture
-async def async_ipr(request, tmpdir):
+async def async_ipr(nsname, request):
     kwarg = getattr(request, 'param', {})
-    async with AsyncIPRouteContext(**kwarg) as ctx:
+    async with AsyncIPRoute(netns=nsname, **kwarg) as ctx:
         yield ctx
 
 
 @pytest.fixture
-def sync_ipr(request, tmpdir):
+def sync_ipr(nsname, request):
     kwarg = getattr(request, 'param', {})
-    with SyncIPRouteContext(**kwarg) as ctx:
+    with IPRoute(netns=nsname, **kwarg) as ctx:
         yield ctx

tests/test_core/test_ipr/test_addr_async.py

@@ -1,7 +1,7 @@
 import re
 
 import pytest
-from net_tools import address_exists, interface_exists
+from net_tools import address_exists
 
 ip4v6 = re.compile('^[.:0-9a-f]*$')
 
@@ -17,18 +17,9 @@ async def test_addr_dump(async_ipr):
         assert prefixlen > 0
 
 
-async def util_link_add(async_ipr):
-    ifname = async_ipr.register_temporary_ifname()
-    await async_ipr.link('add', ifname=ifname, kind='dummy', state='up')
-    assert interface_exists(ifname)
-    (link,) = await async_ipr.link('get', ifname=ifname)
-    return ifname, link.get('index')
-
-
 @pytest.mark.asyncio
-async def test_addr_add(async_ipr):
-    ifname, index = await util_link_add(async_ipr)
+async def test_addr_add(async_ipr, link, nsname):
     await async_ipr.addr(
-        'add', index=index, address='192.168.145.150', prefixlen=24
+        'add', index=link.get('index'), address='192.168.145.150', prefixlen=24
     )
-    assert address_exists('192.168.145.150', ifname)
+    assert address_exists('192.168.145.150', link.get('ifname'), netns=nsname)

tests/test_core/test_ipr/test_link_async.py

@@ -9,21 +9,15 @@ async def test_link_dump(async_ipr):
         assert 1 < len(link.get('ifname')) < 16
 
 
-async def util_link_add(async_ipr):
-    ifname = async_ipr.register_temporary_ifname()
-    await async_ipr.link('add', ifname=ifname, kind='dummy', state='up')
-    assert interface_exists(ifname)
-    return ifname
-
-
 @pytest.mark.asyncio
-async def test_link_add(async_ipr):
-    await util_link_add(async_ipr)
+async def test_link_add(async_ipr, tmp_link, nsname):
+    await async_ipr.link('add', ifname=tmp_link, kind='dummy', state='up')
+    assert interface_exists(tmp_link, netns=nsname)
 
 
 @pytest.mark.asyncio
-async def test_link_get(async_ipr):
-    ifname = await util_link_add(async_ipr)
+async def test_link_get(async_ipr, link):
+    ifname = link.get('ifname')
     (link,) = await async_ipr.link('get', ifname=ifname)
     assert link.get('state') == 'up'
     assert link.get('index') > 1
@@ -32,15 +26,15 @@ async def test_link_get(async_ipr):
 
 
 @pytest.mark.asyncio
-async def test_link_del_by_index(async_ipr):
-    ifname = await util_link_add(async_ipr)
+async def test_link_del_by_index(async_ipr, link, nsname):
+    ifname = link.get('ifname')
     (link,) = await async_ipr.link('get', ifname=ifname)
     await async_ipr.link('del', index=link['index'])
-    assert not interface_exists(ifname)
+    assert not interface_exists(ifname, netns=nsname)
 
 
 @pytest.mark.asyncio
-async def test_link_del_by_name(async_ipr):
-    ifname = await util_link_add(async_ipr)
+async def test_link_del_by_name(async_ipr, link, nsname):
+    ifname = link.get('ifname')
     await async_ipr.link('del', ifname=ifname)
-    assert not interface_exists(ifname)
+    assert not interface_exists(ifname, netns=nsname)

tests/test_core/test_ipr/test_link_sync.py

@@ -7,34 +7,28 @@ def test_link_dump(sync_ipr):
         assert 1 < len(link.get('ifname')) < 16
 
 
-def util_link_add(sync_ipr):
-    ifname = sync_ipr.register_temporary_ifname()
-    sync_ipr.link('add', ifname=ifname, kind='dummy', state='up')
-    assert interface_exists(ifname)
-    return ifname
+def test_link_add(sync_ipr, tmp_link, nsname):
+    sync_ipr.link('add', ifname=tmp_link, kind='dummy', state='up')
+    assert interface_exists(tmp_link, netns=nsname)
 
 
-def test_link_add(sync_ipr):
-    util_link_add(sync_ipr)
-
-
-def test_link_get(sync_ipr):
-    ifname = util_link_add(sync_ipr)
+def test_link_get(sync_ipr, link):
+    ifname = link.get('ifname')
     (link,) = sync_ipr.link('get', ifname=ifname)
     assert link.get('state') == 'up'
     assert link.get('index') > 1
     assert link.get('ifname') == ifname
     assert link.get(('linkinfo', 'kind')) == 'dummy'
 
 
-def test_link_del_by_index(sync_ipr):
-    ifname = util_link_add(sync_ipr)
+def test_link_del_by_index(sync_ipr, link, nsname):
+    ifname = link.get('ifname')
     (link,) = sync_ipr.link('get', ifname=ifname)
     sync_ipr.link('del', index=link['index'])
-    assert not interface_exists(ifname)
+    assert not interface_exists(ifname, netns=nsname)
 
 
-def test_link_del_by_name(sync_ipr):
-    ifname = util_link_add(sync_ipr)
+def test_link_del_by_name(sync_ipr, link, nsname):
+    ifname = link.get('ifname')
     sync_ipr.link('del', ifname=ifname)
-    assert not interface_exists(ifname)
+    assert not interface_exists(ifname, netns=nsname)

tests/test_core/test_ipr/test_route_dump_async.py

@@ -22,44 +22,38 @@
 '''
 
 
-@pytest.mark.parametrize('async_ipr', [{'netns': True}], indirect=True)
 @pytest.mark.asyncio
-async def test_load(async_ipr):
-    netns = async_ipr.status['netns']
-    await async_ipr.link('set', index=1, state='up')
-    assert address_exists('127.0.0.1', ifname='lo', netns=netns)
-    assert not route_exists(dst='10.1.2.0/24', table=100, netns=netns)
-    assert not route_exists(dst='10.1.3.0/24', table=100, netns=netns)
+async def test_load(async_ipr, nsname):
+    assert address_exists('127.0.0.1', ifname='lo', netns=nsname)
+    assert not route_exists(dst='10.1.2.0/24', table=100, netns=nsname)
+    assert not route_exists(dst='10.1.3.0/24', table=100, netns=nsname)
     fd = io.BytesIO()
     fd.write(load_dump(test_dump_data))
     fd.seek(0)
     await async_ipr.route_load(fd)
-    assert route_exists(dst='10.1.2.0/24', table=100, netns=netns)
-    assert route_exists(dst='10.1.3.0/24', table=100, netns=netns)
+    assert route_exists(dst='10.1.2.0/24', table=100, netns=nsname)
+    assert route_exists(dst='10.1.3.0/24', table=100, netns=nsname)
 
 
-@pytest.mark.parametrize('async_ipr', [{'netns': True}], indirect=True)
 @pytest.mark.asyncio
-async def test_loads(async_ipr):
-    netns = async_ipr.status['netns']
-    await async_ipr.link('set', index=1, state='up')
-    assert address_exists('127.0.0.1', ifname='lo', netns=netns)
-    assert not route_exists(dst='10.1.2.0/24', table=100, netns=netns)
-    assert not route_exists(dst='10.1.3.0/24', table=100, netns=netns)
+async def test_loads(async_ipr, nsname):
+    assert address_exists('127.0.0.1', ifname='lo', netns=nsname)
+    assert not route_exists(dst='10.1.2.0/24', table=100, netns=nsname)
+    assert not route_exists(dst='10.1.3.0/24', table=100, netns=nsname)
     await async_ipr.route_loads(load_dump(test_dump_data))
-    assert route_exists(dst='10.1.2.0/24', table=100, netns=netns)
-    assert route_exists(dst='10.1.3.0/24', table=100, netns=netns)
+    assert route_exists(dst='10.1.2.0/24', table=100, netns=nsname)
+    assert route_exists(dst='10.1.3.0/24', table=100, netns=nsname)
 
 
 @pytest.mark.parametrize(
     'family,target_tables,target_families,fmt,offset',
     [
-        (AF_UNSPEC, {254, 255}, {AF_INET, AF_INET6}, 'iproute2', 4),
-        (AF_INET, {254, 255}, {AF_INET}, 'iproute2', 4),
-        (AF_INET6, {254, 255}, {AF_INET6}, 'iproute2', 4),
-        (AF_UNSPEC, {254, 255}, {AF_INET, AF_INET6}, 'raw', 0),
-        (AF_INET, {254, 255}, {AF_INET}, 'raw', 0),
-        (AF_INET6, {254, 255}, {AF_INET6}, 'raw', 0),
+        (AF_UNSPEC, {255}, {AF_INET, AF_INET6}, 'iproute2', 4),
+        (AF_INET, {255}, {AF_INET}, 'iproute2', 4),
+        (AF_INET6, {255}, {AF_INET6}, 'iproute2', 4),
+        (AF_UNSPEC, {255}, {AF_INET, AF_INET6}, 'raw', 0),
+        (AF_INET, {255}, {AF_INET}, 'raw', 0),
+        (AF_INET6, {255}, {AF_INET6}, 'raw', 0),
     ],
     ids=(
         'iproute2/AF_UNSPEC',
@@ -81,19 +75,19 @@ async def test_dump(
     for route in async_ipr.marshal.parse(fd.getvalue()[offset:]):
         tables.add(route.get('table'))
         families.add(route.get('family'))
-    assert tables >= target_tables
+    assert tables == target_tables
     assert families == target_families
 
 
 @pytest.mark.parametrize(
     'family,target_tables,target_families,fmt,offset',
     [
-        (AF_UNSPEC, {254, 255}, {AF_INET, AF_INET6}, 'iproute2', 4),
-        (AF_INET, {254, 255}, {AF_INET}, 'iproute2', 4),
-        (AF_INET6, {254, 255}, {AF_INET6}, 'iproute2', 4),
-        (AF_UNSPEC, {254, 255}, {AF_INET, AF_INET6}, 'raw', 0),
-        (AF_INET, {254, 255}, {AF_INET}, 'raw', 0),
-        (AF_INET6, {254, 255}, {AF_INET6}, 'raw', 0),
+        (AF_UNSPEC, {255}, {AF_INET, AF_INET6}, 'iproute2', 4),
+        (AF_INET, {255}, {AF_INET}, 'iproute2', 4),
+        (AF_INET6, {255}, {AF_INET6}, 'iproute2', 4),
+        (AF_UNSPEC, {255}, {AF_INET, AF_INET6}, 'raw', 0),
+        (AF_INET, {255}, {AF_INET}, 'raw', 0),
+        (AF_INET6, {255}, {AF_INET6}, 'raw', 0),
     ],
     ids=(
         'iproute2/AF_UNSPEC',
@@ -114,5 +108,5 @@ async def test_dumps(
     for route in async_ipr.marshal.parse(data[offset:]):
         tables.add(route.get('table'))
         families.add(route.get('family'))
-    assert tables >= target_tables
+    assert tables == target_tables
     assert families == target_families