summaryrefslogtreecommitdiffstats
path: root/tests/topotests/munet/testing/fixtures.py
blob: 3c6d9460ffdf8ab4af8ffc67d9656aa64029a53b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
# SPDX-License-Identifier: GPL-2.0-or-later
#
# April 22 2022, Christian Hopps <chopps@gmail.com>
#
# Copyright (c) 2022, LabN Consulting, L.L.C
#
"""A module that implements pytest fixtures.

To use in your project, in your conftest.py add:

  from munet.testing.fixtures import *
"""
import contextlib
import logging
import os

from pathlib import Path
from typing import Union

import pytest
import pytest_asyncio

from ..base import BaseMunet
from ..base import Bridge
from ..base import get_event_loop
from ..cleanup import cleanup_current
from ..cleanup import cleanup_previous
from ..native import L3NodeMixin
from ..parser import async_build_topology
from ..parser import get_config
from .util import async_pause_test
from .util import pause_test


@contextlib.asynccontextmanager
async def achdir(ndir: Union[str, Path], desc=""):
    odir = os.getcwd()
    os.chdir(ndir)
    if desc:
        logging.debug("%s: chdir from %s to %s", desc, odir, ndir)
    try:
        yield
    finally:
        if desc:
            logging.debug("%s: chdir back from %s to %s", desc, ndir, odir)
        os.chdir(odir)


@contextlib.contextmanager
def chdir(ndir: Union[str, Path], desc=""):
    odir = os.getcwd()
    os.chdir(ndir)
    if desc:
        logging.debug("%s: chdir from %s to %s", desc, odir, ndir)
    try:
        yield
    finally:
        if desc:
            logging.debug("%s: chdir back from %s to %s", desc, ndir, odir)
        os.chdir(odir)


def get_test_logdir(nodeid=None, module=False):
    """Get log directory relative pathname."""
    xdist_worker = os.getenv("PYTEST_XDIST_WORKER", "")
    mode = os.getenv("PYTEST_XDIST_MODE", "no")

    # nodeid: all_protocol_startup/test_all_protocol_startup.py::test_router_running
    # may be missing "::testname" if module is True
    if not nodeid:
        nodeid = os.environ["PYTEST_CURRENT_TEST"].split(" ")[0]

    cur_test = nodeid.replace("[", "_").replace("]", "_")
    if module:
        idx = cur_test.rfind("::")
        path = cur_test if idx == -1 else cur_test[:idx]
        testname = ""
    else:
        path, testname = cur_test.split("::")
        testname = testname.replace("/", ".")
    path = path[:-3].replace("/", ".")

    # We use different logdir paths based on how xdist is running.
    if mode == "each":
        if module:
            return os.path.join(path, "worker-logs", xdist_worker)
        return os.path.join(path, testname, xdist_worker)
    assert mode in ("no", "load", "loadfile", "loadscope"), f"Unknown dist mode {mode}"
    return path if module else os.path.join(path, testname)


def _push_log_handler(desc, logpath):
    logpath = os.path.abspath(logpath)
    logging.debug("conftest: adding %s logging at %s", desc, logpath)
    root_logger = logging.getLogger()
    handler = logging.FileHandler(logpath, mode="w")
    fmt = logging.Formatter("%(asctime)s %(levelname)5s: %(name)s: %(message)s")
    handler.setFormatter(fmt)
    root_logger.addHandler(handler)
    return handler


def _pop_log_handler(handler):
    root_logger = logging.getLogger()
    logging.debug("conftest: removing logging handler %s", handler)
    root_logger.removeHandler(handler)


@contextlib.contextmanager
def log_handler(desc, logpath):
    handler = _push_log_handler(desc, logpath)
    try:
        yield
    finally:
        _pop_log_handler(handler)


# =================
# Sessions Fixtures
# =================


@pytest.fixture(autouse=True, scope="session")
def session_autouse():
    if "PYTEST_TOPOTEST_WORKER" not in os.environ:
        is_worker = False
    elif not os.environ["PYTEST_TOPOTEST_WORKER"]:
        is_worker = False
    else:
        is_worker = True

    if not is_worker:
        # This is unfriendly to multi-instance
        cleanup_previous()

    # We never pop as we want to keep logging
    _push_log_handler("session", "/tmp/unet-test/pytest-session.log")

    yield

    if not is_worker:
        cleanup_current()


# ===============
# Module Fixtures
# ===============


@pytest.fixture(autouse=True, scope="module")
def module_autouse(request):
    logpath = get_test_logdir(request.node.name, True)
    logpath = os.path.join("/tmp/unet-test", logpath, "pytest-exec.log")
    with log_handler("module", logpath):
        sdir = os.path.dirname(os.path.realpath(request.fspath))
        with chdir(sdir, "module autouse fixture"):
            yield

        if BaseMunet.g_unet:
            raise Exception("Base Munet was not cleaned up/deleted")


@pytest.fixture(scope="module")
def event_loop():
    """Create an instance of the default event loop for the session."""
    loop = get_event_loop()
    try:
        logging.info("event_loop_fixture: yielding with new event loop watcher")
        yield loop
    finally:
        loop.close()


@pytest.fixture(scope="module")
def rundir_module():
    d = os.path.join("/tmp/unet-test", get_test_logdir(module=True))
    logging.debug("conftest: test module rundir %s", d)
    return d


async def _unet_impl(
    _rundir, _pytestconfig, unshare=None, top_level_pidns=None, param=None
):
    try:
        # Default is not to unshare inline if not specified otherwise
        unshare_default = False
        pidns_default = True
        if isinstance(param, (tuple, list)):
            pidns_default = bool(param[2]) if len(param) > 2 else True
            unshare_default = bool(param[1]) if len(param) > 1 else False
            param = str(param[0])
        elif isinstance(param, bool):
            unshare_default = param
            param = None
        if unshare is None:
            unshare = unshare_default
        if top_level_pidns is None:
            top_level_pidns = pidns_default

        logging.info("unet fixture: basename=%s unshare_inline=%s", param, unshare)
        _unet = await async_build_topology(
            config=get_config(basename=param) if param else None,
            rundir=_rundir,
            unshare_inline=unshare,
            top_level_pidns=top_level_pidns,
            pytestconfig=_pytestconfig,
        )
    except Exception as error:
        logging.debug(
            "unet fixture: unet build failed: %s\nparam: %s",
            error,
            param,
            exc_info=True,
        )
        pytest.skip(
            f"unet fixture: unet build failed: {error}", allow_module_level=True
        )
        raise

    try:
        tasks = await _unet.run()
    except Exception as error:
        logging.debug("unet fixture: unet run failed: %s", error, exc_info=True)
        await _unet.async_delete()
        pytest.skip(f"unet fixture: unet run failed: {error}", allow_module_level=True)
        raise

    logging.debug("unet fixture: containers running")

    # Pytest is supposed to always return even if exceptions
    try:
        yield _unet
    except Exception as error:
        logging.error("unet fixture: yield unet unexpected exception: %s", error)

    logging.debug("unet fixture: module done, deleting unet")
    await _unet.async_delete()

    # No one ever awaits these so cancel them
    logging.debug("unet fixture: cleanup")
    for task in tasks:
        task.cancel()

    # Reset the class variables so auto number is predictable
    logging.debug("unet fixture: resetting ords to 1")
    L3NodeMixin.next_ord = 1
    Bridge.next_ord = 1


@pytest.fixture(scope="module")
async def unet(request, rundir_module, pytestconfig):  # pylint: disable=W0621
    """A unet creating fixutre.

    The request param is either the basename of the config file or a tuple of the form:
    (basename, unshare, top_level_pidns), with the second and third elements boolean and
    optional, defaulting to False, True.
    """
    param = request.param if hasattr(request, "param") else None
    sdir = os.path.dirname(os.path.realpath(request.fspath))
    async with achdir(sdir, "unet fixture"):
        async for x in _unet_impl(rundir_module, pytestconfig, param=param):
            yield x


@pytest.fixture(scope="module")
async def unet_share(request, rundir_module, pytestconfig):  # pylint: disable=W0621
    """A unet creating fixutre.

    This share variant keeps munet from unsharing the process to a new namespace so that
    root level commands and actions are execute on the host, normally they are executed
    in the munet namespace which allowing things like scapy inline in tests to work.

    The request param is either the basename of the config file or a tuple of the form:
    (basename, top_level_pidns), the second value is a boolean.
    """
    param = request.param if hasattr(request, "param") else None
    if isinstance(param, (tuple, list)):
        param = (param[0], False, param[1])
    sdir = os.path.dirname(os.path.realpath(request.fspath))
    async with achdir(sdir, "unet_share fixture"):
        async for x in _unet_impl(
            rundir_module, pytestconfig, unshare=False, param=param
        ):
            yield x


@pytest.fixture(scope="module")
async def unet_unshare(request, rundir_module, pytestconfig):  # pylint: disable=W0621
    """A unet creating fixutre.

    This unshare variant has the top level munet unshare the process inline so that
    root level commands and actions are execute in a new namespace. This allows things
    like scapy inline in tests to work.

    The request param is either the basename of the config file or a tuple of the form:
    (basename, top_level_pidns), the second value is a boolean.
    """
    param = request.param if hasattr(request, "param") else None
    if isinstance(param, (tuple, list)):
        param = (param[0], True, param[1])
    sdir = os.path.dirname(os.path.realpath(request.fspath))
    async with achdir(sdir, "unet_unshare fixture"):
        async for x in _unet_impl(
            rundir_module, pytestconfig, unshare=True, param=param
        ):
            yield x


# =================
# Function Fixtures
# =================


@pytest.fixture(autouse=True, scope="function")
async def function_autouse(request):
    async with achdir(
        os.path.dirname(os.path.realpath(request.fspath)), "func.fixture"
    ):
        yield


@pytest.fixture(autouse=True)
async def check_for_pause(request, pytestconfig):
    # When we unshare inline we can't pause in the pytest_runtest_makereport hook
    # so do it here.
    if BaseMunet.g_unet and BaseMunet.g_unet.unshare_inline:
        pause = bool(pytestconfig.getoption("--pause"))
        if pause:
            await async_pause_test(f"XXX before test '{request.node.name}'")
    yield


@pytest.fixture(scope="function")
def stepf(pytestconfig):
    class Stepnum:
        """Track the stepnum in closure."""

        num = 0

        def inc(self):
            self.num += 1

    pause = pytestconfig.getoption("pause")
    stepnum = Stepnum()

    def stepfunction(desc=""):
        desc = f": {desc}" if desc else ""
        if pause:
            pause_test(f"before step {stepnum.num}{desc}")
        logging.info("STEP %s%s", stepnum.num, desc)
        stepnum.inc()

    return stepfunction


@pytest_asyncio.fixture(scope="function")
async def astepf(pytestconfig):
    class Stepnum:
        """Track the stepnum in closure."""

        num = 0

        def inc(self):
            self.num += 1

    pause = pytestconfig.getoption("pause")
    stepnum = Stepnum()

    async def stepfunction(desc=""):
        desc = f": {desc}" if desc else ""
        if pause:
            await async_pause_test(f"before step {stepnum.num}{desc}")
        logging.info("STEP %s%s", stepnum.num, desc)
        stepnum.inc()

    return stepfunction


@pytest.fixture(scope="function")
def rundir():
    d = os.path.join("/tmp/unet-test", get_test_logdir(module=False))
    logging.debug("conftest: test function rundir %s", d)
    return d


# Configure logging
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_setup(item):
    d = os.path.join(
        "/tmp/unet-test", get_test_logdir(nodeid=item.nodeid, module=False)
    )
    config = item.config
    logging_plugin = config.pluginmanager.get_plugin("logging-plugin")
    filename = Path(d, "pytest-exec.log")
    logging_plugin.set_log_path(str(filename))
    logging.debug("conftest: test function setup: rundir %s", d)
    yield


@pytest.fixture
async def unet_perfunc(request, rundir, pytestconfig):  # pylint: disable=W0621
    param = request.param if hasattr(request, "param") else None
    async for x in _unet_impl(rundir, pytestconfig, param=param):
        yield x


@pytest.fixture
async def unet_perfunc_unshare(request, rundir, pytestconfig):  # pylint: disable=W0621
    """Build unet per test function with an optional topology basename parameter.

    The fixture can be parameterized to choose different config files.
    For example, use as follows to run the test with unet_perfunc configured
    first with a config file named `cfg1.yaml` then with config file `cfg2.yaml`
    (where the actual files could end with `json` or `toml` rather than `yaml`).

        @pytest.mark.parametrize(
            "unet_perfunc", ["cfg1", "cfg2]", indirect=["unet_perfunc"]
        )
        def test_example(unet_perfunc)
    """
    param = request.param if hasattr(request, "param") else None
    async for x in _unet_impl(rundir, pytestconfig, unshare=True, param=param):
        yield x


@pytest.fixture
async def unet_perfunc_share(request, rundir, pytestconfig):  # pylint: disable=W0621
    """Build unet per test function with an optional topology basename parameter.

    This share variant keeps munet from unsharing the process to a new namespace so that
    root level commands and actions are execute on the host, normally they are executed
    in the munet namespace which allowing things like scapy inline in tests to work.

    The fixture can be parameterized to choose different config files.  For example, use
    as follows to run the test with unet_perfunc configured first with a config file
    named `cfg1.yaml` then with config file `cfg2.yaml` (where the actual files could
    end with `json` or `toml` rather than `yaml`).

        @pytest.mark.parametrize(
            "unet_perfunc", ["cfg1", "cfg2]", indirect=["unet_perfunc"]
        )
        def test_example(unet_perfunc)
    """
    param = request.param if hasattr(request, "param") else None
    async for x in _unet_impl(rundir, pytestconfig, unshare=False, param=param):
        yield x