ci: move fixtures to the source tree as a module

pyroute2/fixtures/iproute.py

@@ -0,0 +1,110 @@
+import errno
+from collections import namedtuple
+from collections.abc import AsyncGenerator, Generator
+from typing import Union
+
+import pytest
+import pytest_asyncio
+
+from pyroute2 import netns
+from pyroute2.common import uifname
+from pyroute2.iproute.linux import AsyncIPRoute, IPRoute
+from pyroute2.netlink.exceptions import NetlinkError
+
+TestInterface = namedtuple('TestInterface', ('index', 'ifname'))
+
+
+@pytest.fixture
+def nsname() -> Generator[str]:
+    '''Create a unique network namespace and yield its name.
+
+    Remove the netns on exit.
+    '''
+    nsname = uifname()
+    netns.create(nsname)
+    with IPRoute(netns=nsname) as ipr:
+        ipr.link('set', index=1, state='up')
+        ipr.poll(ipr.addr, 'dump', address='127.0.0.1', timeout=5)
+    yield nsname
+    try:
+        netns.remove(nsname)
+    except Exception:
+        pass
+
+
+@pytest.fixture
+def test_link(nsname: str) -> Generator[TestInterface]:
+    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'))
+        try:
+            ipr.link('del', index=link.get('index'))
+        except NetlinkError as e:
+            if e.code != errno.ENODEV:
+                raise
+
+
+@pytest.fixture
+def test_link_index(test_link: TestInterface) -> Generator[int]:
+    yield test_link.index
+
+
+@pytest.fixture
+def test_link_ifname(test_link: TestInterface) -> Generator[str]:
+    yield test_link.ifname
+
+
+@pytest.fixture
+def tmp_link_ifname(nsname: str) -> Generator[str]:
+    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
+
+
+class TestContext:
+    def __init__(
+        self, ipr: Union[IPRoute, AsyncIPRoute], test_link: TestInterface
+    ):
+        self.ipr = ipr
+        self.test_link = test_link
+        self.netns = self.ipr.status['netns']
+
+
+@pytest_asyncio.fixture
+async def async_ipr(
+    request, nsname: str, test_link: TestInterface
+) -> AsyncGenerator[AsyncIPRoute]:
+    kwarg = getattr(request, 'param', {})
+    async with AsyncIPRoute(netns=nsname, **kwarg) as ipr:
+        yield ipr
+
+
+@pytest.fixture
+def sync_ipr(
+    request, nsname: str, test_link: TestInterface
+) -> Generator[IPRoute]:
+    kwarg = getattr(request, 'param', {})
+    with IPRoute(netns=nsname, **kwarg) as ipr:
+        yield ipr
+
+
+@pytest_asyncio.fixture
+async def async_context(
+    async_ipr: AsyncIPRoute, test_link: TestInterface
+) -> AsyncGenerator[TestContext]:
+    yield TestContext(async_ipr, test_link)
+
+
+@pytest.fixture
+def sync_context(
+    sync_ipr: IPRoute, test_link: TestInterface
+) -> Generator[TestContext]:
+    yield TestContext(sync_ipr, test_link)

tests/test_core/pr2test/plan9.py → pyroute2/fixtures/plan9.py

@@ -1,5 +1,8 @@
+from collections.abc import AsyncGenerator
 from socket import socketpair
 
+import pytest_asyncio
+
 from pyroute2.plan9.client import Plan9ClientSocket
 from pyroute2.plan9.server import Plan9ServerSocket
 
@@ -24,3 +27,11 @@ async def ensure_session(self):
 
     async def close(self):
         self.task.cancel()
+
+
+@pytest_asyncio.fixture
+async def async_p9_context() -> AsyncGenerator[AsyncPlan9Context]:
+    ctx = AsyncPlan9Context()
+    await ctx.ensure_session()
+    yield ctx
+    await ctx.close()

tests/test_core/conftest.py

@@ -1,81 +1,25 @@
-import errno
-
-import pytest
-import pytest_asyncio
-from pr2test.plan9 import AsyncPlan9Context
-
-from pyroute2 import AsyncIPRoute, IPRoute, NetlinkError, netns
-from pyroute2.common import uifname
-
-
-@pytest_asyncio.fixture
-async def p9(request, tmpdir):
-    ctx = AsyncPlan9Context()
-    await ctx.ensure_session()
-    yield ctx
-    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(nsname, request):
-    kwarg = getattr(request, 'param', {})
-    async with AsyncIPRoute(netns=nsname, **kwarg) as ctx:
-        yield ctx
-
-
-@pytest.fixture
-def sync_ipr(nsname, request):
-    kwarg = getattr(request, 'param', {})
-    with IPRoute(netns=nsname, **kwarg) as ctx:
-        yield ctx
+from pyroute2.fixtures.iproute import (
+    async_context,
+    async_ipr,
+    nsname,
+    sync_context,
+    sync_ipr,
+    test_link,
+    test_link_ifname,
+    test_link_index,
+    tmp_link_ifname,
+)
+from pyroute2.fixtures.plan9 import async_p9_context
+
+fixtures = [
+    async_p9_context,
+    async_context,
+    async_ipr,
+    nsname,
+    sync_context,
+    sync_ipr,
+    test_link,
+    test_link_index,
+    test_link_ifname,
+    tmp_link_ifname,
+]

tests/test_core/test_ipr/test_addr_async.py

@@ -18,8 +18,8 @@ async def test_addr_dump(async_ipr):
 
 
 @pytest.mark.asyncio
-async def test_addr_add(async_ipr, ifname, index, nsname):
+async def test_addr_add(async_ipr, test_link_ifname, test_link_index, nsname):
     await async_ipr.addr(
-        'add', index=index, address='192.168.145.150', prefixlen=24
+        'add', index=test_link_index, address='192.168.145.150', prefixlen=24
     )
-    assert address_exists('192.168.145.150', ifname, netns=nsname)
+    assert address_exists('192.168.145.150', test_link_ifname, netns=nsname)

tests/test_core/test_ipr/test_addr_sync.py

@@ -15,6 +15,8 @@ def test_addr_dump(sync_ipr):
         assert prefixlen > 0
 
 
-def test_addr_add(sync_ipr, ifname, index, nsname):
-    sync_ipr.addr('add', index=index, address='192.168.145.150', prefixlen=24)
-    assert address_exists('192.168.145.150', ifname, netns=nsname)
+def test_addr_add(sync_ipr, test_link_ifname, test_link_index, nsname):
+    sync_ipr.addr(
+        'add', index=test_link_index, address='192.168.145.150', prefixlen=24
+    )
+    assert address_exists('192.168.145.150', test_link_ifname, netns=nsname)

tests/test_core/test_ipr/test_link_async.py

@@ -10,28 +10,32 @@ async def test_link_dump(async_ipr):
 
 
 @pytest.mark.asyncio
-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)
+async def test_link_add(async_ipr, tmp_link_ifname, nsname):
+    await async_ipr.link(
+        'add', ifname=tmp_link_ifname, kind='dummy', state='up'
+    )
+    assert interface_exists(tmp_link_ifname, netns=nsname)
 
 
 @pytest.mark.asyncio
-async def test_link_get(async_ipr, ifname):
-    (link,) = await async_ipr.link('get', ifname=ifname)
+async def test_link_get(async_ipr, test_link_ifname):
+    (link,) = await async_ipr.link('get', ifname=test_link_ifname)
     assert link.get('state') == 'up'
     assert link.get('index') > 1
-    assert link.get('ifname') == ifname
+    assert link.get('ifname') == test_link_ifname
     assert link.get(('linkinfo', 'kind')) == 'dummy'
 
 
 @pytest.mark.asyncio
-async def test_link_del_by_index(async_ipr, ifname, index, nsname):
-    (link,) = await async_ipr.link('get', ifname=ifname)
-    await async_ipr.link('del', index=index)
-    assert not interface_exists(ifname, netns=nsname)
+async def test_link_del_by_index(
+    async_ipr, test_link_ifname, test_link_index, nsname
+):
+    (link,) = await async_ipr.link('get', ifname=test_link_ifname)
+    await async_ipr.link('del', index=test_link_index)
+    assert not interface_exists(test_link_ifname, netns=nsname)
 
 
 @pytest.mark.asyncio
-async def test_link_del_by_name(async_ipr, ifname, nsname):
-    await async_ipr.link('del', ifname=ifname)
-    assert not interface_exists(ifname, netns=nsname)
+async def test_link_del_by_name(async_ipr, test_link_ifname, nsname):
+    await async_ipr.link('del', ifname=test_link_ifname)
+    assert not interface_exists(test_link_ifname, netns=nsname)

tests/test_core/test_ipr/test_link_sync.py

@@ -7,25 +7,27 @@ def test_link_dump(sync_ipr):
         assert 1 < len(link.get('ifname')) < 16
 
 
-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, tmp_link_ifname, nsname):
+    sync_ipr.link('add', ifname=tmp_link_ifname, kind='dummy', state='up')
+    assert interface_exists(tmp_link_ifname, netns=nsname)
 
 
-def test_link_get(sync_ipr, ifname):
-    (link,) = sync_ipr.link('get', ifname=ifname)
+def test_link_get(sync_ipr, test_link_ifname):
+    (link,) = sync_ipr.link('get', ifname=test_link_ifname)
     assert link.get('state') == 'up'
     assert link.get('index') > 1
-    assert link.get('ifname') == ifname
+    assert link.get('ifname') == test_link_ifname
     assert link.get(('linkinfo', 'kind')) == 'dummy'
 
 
-def test_link_del_by_index(sync_ipr, ifname, index, nsname):
-    (link,) = sync_ipr.link('get', ifname=ifname)
-    sync_ipr.link('del', index=index)
-    assert not interface_exists(ifname, netns=nsname)
+def test_link_del_by_index(
+    sync_ipr, test_link_ifname, test_link_index, nsname
+):
+    (link,) = sync_ipr.link('get', ifname=test_link_ifname)
+    sync_ipr.link('del', index=test_link_index)
+    assert not interface_exists(test_link_ifname, netns=nsname)
 
 
-def test_link_del_by_name(sync_ipr, ifname, nsname):
-    sync_ipr.link('del', ifname=ifname)
-    assert not interface_exists(ifname, netns=nsname)
+def test_link_del_by_name(sync_ipr, test_link_ifname, nsname):
+    sync_ipr.link('del', ifname=test_link_ifname)
+    assert not interface_exists(test_link_ifname, netns=nsname)