]>
git.proxmox.com Git - mirror_frr.git/blob - tests/topotests/conftest.py
2 Topotest conftest.py file.
4 # pylint: disable=consider-using-f-string
17 from lib
import topolog
18 from lib
.micronet
import Commander
, proc_error
19 from lib
.micronet_cli
import cli
20 from lib
.micronet_compat
import Mininet
, cleanup_current
, cleanup_previous
21 from lib
.topogen
import diagnose_env
, get_topogen
22 from lib
.topolog
import logger
23 from lib
.topotest
import g_extra_config
as topotest_extra_config
24 from lib
.topotest
import json_cmp_result
27 def pytest_addoption(parser
):
29 Add topology-only option to the topology tester. This option makes pytest
30 only run the setup_module() to setup the topology without running any tests.
35 help="Configure address sanitizer to abort process on error",
41 help="Mininet cli on test failure",
46 metavar
="SYMBOL[,SYMBOL...]",
47 help="Comma-separated list of functions to set gdb breakpoints on",
52 metavar
="DAEMON[,DAEMON...]",
53 help="Comma-separated list of daemons to spawn gdb on, or 'all'",
58 metavar
="ROUTER[,ROUTER...]",
59 help="Comma-separated list of routers to spawn gdb on, or 'all'",
65 help="Pause after each test",
71 help="Do not pause after (disables default when --shell or -vtysh given)",
75 "--no-pause-on-error",
76 dest
="pause_on_error",
78 help="Do not pause after (disables default when --shell or -vtysh given)",
81 rundir_help
= "directory for running in and log files"
82 parser
.addini("rundir", rundir_help
, default
="/tmp/topotests")
83 parser
.addoption("--rundir", metavar
="DIR", help=rundir_help
)
87 metavar
="ROUTER[,ROUTER...]",
88 help="Comma-separated list of routers to spawn shell on, or 'all'",
94 help="Spawn shell on all routers on test failure",
99 metavar
="DAEMON[,DAEMON...]",
100 help="Comma-separated list of daemons to strace, or 'all'",
107 help="Only set up this topology, don't run tests",
113 help="Generate suppression file, and enable more precise (slower) valgrind checks",
117 "--valgrind-memleaks",
119 help="Run all daemons under valgrind for memleak detection",
124 metavar
="ROUTER[,ROUTER...]",
125 help="Comma-separated list of routers to spawn vtysh on, or 'all'",
131 help="Spawn vtysh on all routers on test failure",
135 def check_for_memleaks():
136 assert topotest_extra_config
["valgrind_memleaks"]
139 tgen
= get_topogen() # pylint: disable=redefined-outer-name
144 if hasattr(tgen
, "valgrind_existing_files"):
145 existing
= tgen
.valgrind_existing_files
146 latest
= glob
.glob(os
.path
.join(logdir
, "*.valgrind.*"))
147 latest
= [x
for x
in latest
if "core" not in x
]
151 if vfile
in existing
:
153 # do not consider memleaks from parent fork (i.e., owned by root)
154 if os
.stat(vfile
).st_uid
== 0:
155 existing
.append(vfile
) # do not check again
156 logger
.debug("Skipping valgrind file %s owned by root", vfile
)
158 logger
.debug("Checking valgrind file %s not owned by root", vfile
)
159 with
open(vfile
, encoding
="ascii") as vf
:
160 vfcontent
= vf
.read()
161 match
= re
.search(r
"ERROR SUMMARY: (\d+) errors", vfcontent
)
163 existing
.append(vfile
) # have summary don't check again
164 if match
and match
.group(1) != "0":
165 emsg
= "{} in {}".format(match
.group(1), vfile
)
167 daemon
= re
.match(r
".*\.valgrind\.(.*)\.\d+", vfile
).group(1)
168 daemons
.add("{}({})".format(daemon
, match
.group(1)))
171 tgen
.valgrind_existing_files
= existing
174 logger
.error("valgrind memleaks found:\n\t%s", "\n\t".join(leaks
))
175 pytest
.fail("valgrind memleaks found for daemons: " + " ".join(daemons
))
178 @pytest.fixture(autouse
=True, scope
="module")
179 def module_check_memtest(request
):
180 del request
# disable unused warning
182 if topotest_extra_config
["valgrind_memleaks"]:
183 if get_topogen() is not None:
187 def pytest_runtest_logstart(nodeid
, location
):
188 # location is (filename, lineno, testname)
189 topolog
.logstart(nodeid
, location
, topotest_extra_config
["rundir"])
192 def pytest_runtest_logfinish(nodeid
, location
):
193 # location is (filename, lineno, testname)
194 topolog
.logfinish(nodeid
, location
)
197 @pytest.hookimpl(hookwrapper
=True)
198 def pytest_runtest_call(item
: pytest
.Item
) -> None:
199 "Hook the function that is called to execute the test."
200 del item
# disable unused warning
202 # For topology only run the CLI then exit
203 if topotest_extra_config
["topology_only"]:
205 pytest
.exit("exiting after --topology-only")
207 # Let the default pytest_runtest_call execute the test function
210 # Check for leaks if requested
211 if topotest_extra_config
["valgrind_memleaks"]:
215 def pytest_assertrepr_compare(op
, left
, right
):
217 Show proper assertion error message for json_cmp results.
222 if not isinstance(json_result
, json_cmp_result
):
224 if not isinstance(json_result
, json_cmp_result
):
227 return json_result
.gen_report()
230 def pytest_configure(config
):
232 Assert that the environment is correctly configured, and get extra config.
235 if config
.getoption("--collect-only"):
238 if "PYTEST_XDIST_WORKER" not in os
.environ
:
239 os
.environ
["PYTEST_XDIST_MODE"] = config
.getoption("dist", "no")
240 os
.environ
["PYTEST_TOPOTEST_WORKER"] = ""
241 is_xdist
= os
.environ
["PYTEST_XDIST_MODE"] != "no"
244 os
.environ
["PYTEST_TOPOTEST_WORKER"] = os
.environ
["PYTEST_XDIST_WORKER"]
249 resource
.RLIMIT_CORE
, (resource
.RLIM_INFINITY
, resource
.RLIM_INFINITY
)
251 # -----------------------------------------------------
252 # Set some defaults for the pytest.ini [pytest] section
253 # ---------------------------------------------------
255 rundir
= config
.getoption("--rundir")
257 rundir
= config
.getini("rundir")
259 rundir
= "/tmp/topotests"
260 if not config
.getoption("--junitxml"):
261 config
.option
.xmlpath
= os
.path
.join(rundir
, "topotests.xml")
262 xmlpath
= config
.option
.xmlpath
264 # Save an existing topotest.xml
265 if os
.path
.exists(xmlpath
):
266 fmtime
= time
.localtime(os
.path
.getmtime(xmlpath
))
267 suffix
= "-" + time
.strftime("%Y%m%d%H%M%S", fmtime
)
268 commander
= Commander("pytest")
269 mv_path
= commander
.get_exec_path("mv")
270 commander
.cmd_status([mv_path
, xmlpath
, xmlpath
+ suffix
])
272 topotest_extra_config
["rundir"] = rundir
274 # Set the log_file (exec) to inside the rundir if not specified
275 if not config
.getoption("--log-file") and not config
.getini("log_file"):
276 config
.option
.log_file
= os
.path
.join(rundir
, "exec.log")
278 # Turn on live logging if user specified verbose and the config has a CLI level set
279 if config
.getoption("--verbose") and not is_xdist
and not config
.getini("log_cli"):
280 if config
.getoption("--log-cli-level", None) is None:
281 # By setting the CLI option to the ini value it enables log_cli=1
282 cli_level
= config
.getini("log_cli_level")
283 if cli_level
is not None:
284 config
.option
.log_cli_level
= cli_level
286 have_tmux
= bool(os
.getenv("TMUX", ""))
287 have_screen
= not have_tmux
and bool(os
.getenv("STY", ""))
288 have_xterm
= not have_tmux
and not have_screen
and bool(os
.getenv("DISPLAY", ""))
289 have_windows
= have_tmux
or have_screen
or have_xterm
290 have_windows_pause
= have_tmux
or have_xterm
291 xdist_no_windows
= is_xdist
and not is_worker
and not have_windows_pause
293 def assert_feature_windows(b
, feature
):
294 if b
and xdist_no_windows
:
296 "{} use requires byobu/TMUX/XTerm under dist {}".format(
297 feature
, os
.environ
["PYTEST_XDIST_MODE"]
300 elif b
and not is_xdist
and not have_windows
:
301 pytest
.exit("{} use requires byobu/TMUX/SCREEN/XTerm".format(feature
))
303 # ---------------------------------------
304 # Record our options in global dictionary
305 # ---------------------------------------
307 topotest_extra_config
["rundir"] = rundir
309 asan_abort
= config
.getoption("--asan-abort")
310 topotest_extra_config
["asan_abort"] = asan_abort
312 gdb_routers
= config
.getoption("--gdb-routers")
313 gdb_routers
= gdb_routers
.split(",") if gdb_routers
else []
314 topotest_extra_config
["gdb_routers"] = gdb_routers
316 gdb_daemons
= config
.getoption("--gdb-daemons")
317 gdb_daemons
= gdb_daemons
.split(",") if gdb_daemons
else []
318 topotest_extra_config
["gdb_daemons"] = gdb_daemons
319 assert_feature_windows(gdb_routers
or gdb_daemons
, "GDB")
321 gdb_breakpoints
= config
.getoption("--gdb-breakpoints")
322 gdb_breakpoints
= gdb_breakpoints
.split(",") if gdb_breakpoints
else []
323 topotest_extra_config
["gdb_breakpoints"] = gdb_breakpoints
325 cli_on_error
= config
.getoption("--cli-on-error")
326 topotest_extra_config
["cli_on_error"] = cli_on_error
327 assert_feature_windows(cli_on_error
, "--cli-on-error")
329 shell
= config
.getoption("--shell")
330 topotest_extra_config
["shell"] = shell
.split(",") if shell
else []
331 assert_feature_windows(shell
, "--shell")
333 strace
= config
.getoption("--strace-daemons")
334 topotest_extra_config
["strace_daemons"] = strace
.split(",") if strace
else []
336 shell_on_error
= config
.getoption("--shell-on-error")
337 topotest_extra_config
["shell_on_error"] = shell_on_error
338 assert_feature_windows(shell_on_error
, "--shell-on-error")
340 topotest_extra_config
["valgrind_extra"] = config
.getoption("--valgrind-extra")
341 topotest_extra_config
["valgrind_memleaks"] = config
.getoption("--valgrind-memleaks")
343 vtysh
= config
.getoption("--vtysh")
344 topotest_extra_config
["vtysh"] = vtysh
.split(",") if vtysh
else []
345 assert_feature_windows(vtysh
, "--vtysh")
347 vtysh_on_error
= config
.getoption("--vtysh-on-error")
348 topotest_extra_config
["vtysh_on_error"] = vtysh_on_error
349 assert_feature_windows(vtysh_on_error
, "--vtysh-on-error")
351 pause_on_error
= vtysh
or shell
or config
.getoption("--pause-on-error")
352 if config
.getoption("--no-pause-on-error"):
353 pause_on_error
= False
355 topotest_extra_config
["pause_on_error"] = pause_on_error
356 assert_feature_windows(pause_on_error
, "--pause-on-error")
358 pause
= config
.getoption("--pause")
359 topotest_extra_config
["pause"] = pause
360 assert_feature_windows(pause
, "--pause")
362 topology_only
= config
.getoption("--topology-only")
363 if topology_only
and is_xdist
:
364 pytest
.exit("Cannot use --topology-only with distributed test mode")
365 topotest_extra_config
["topology_only"] = topology_only
367 # Check environment now that we have config
368 if not diagnose_env(rundir
):
369 pytest
.exit("environment has errors, please read the logs in %s" % rundir
)
372 @pytest.fixture(autouse
=True, scope
="session")
373 def setup_session_auto():
374 if "PYTEST_TOPOTEST_WORKER" not in os
.environ
:
376 elif not os
.environ
["PYTEST_TOPOTEST_WORKER"]:
381 logger
.debug("Before the run (is_worker: %s)", is_worker
)
387 logger
.debug("After the run (is_worker: %s)", is_worker
)
390 def pytest_runtest_setup(item
):
391 module
= item
.parent
.module
392 script_dir
= os
.path
.abspath(os
.path
.dirname(module
.__file
__))
393 os
.environ
["PYTEST_TOPOTEST_SCRIPTDIR"] = script_dir
396 def pytest_runtest_makereport(item
, call
):
397 "Log all assert messages to default logger with error level"
399 pause
= bool(item
.config
.getoption("--pause"))
402 if call
.excinfo
is None:
406 modname
= parent
.module
.__name
__
408 # Treat skips as non errors, don't pause after
409 if call
.excinfo
.typename
== "Skipped":
413 'test skipped at "{}/{}": {}'.format(
414 modname
, item
.name
, call
.excinfo
.value
419 # Handle assert failures
420 parent
._previousfailed
= item
# pylint: disable=W0212
422 'test failed at "{}/{}": {}'.format(
423 modname
, item
.name
, call
.excinfo
.value
426 title
= "{}/{}".format(modname
, item
.name
)
428 # We want to pause, if requested, on any error not just test cases
429 # (e.g., call.when == "setup")
432 topotest_extra_config
["pause_on_error"]
433 or topotest_extra_config
["pause"]
436 # (topogen) Set topology error to avoid advancing in the test.
437 tgen
= get_topogen() # pylint: disable=redefined-outer-name
439 # This will cause topogen to report error on `routers_have_failure`.
440 tgen
.set_error("{}/{}".format(modname
, item
.name
))
442 commander
= Commander("pytest")
443 isatty
= sys
.stdout
.isatty()
446 if error
and topotest_extra_config
["vtysh_on_error"]:
447 error_cmd
= commander
.get_exec_path(["vtysh"])
448 elif error
and topotest_extra_config
["shell_on_error"]:
449 error_cmd
= os
.getenv("SHELL", commander
.get_exec_path(["bash"]))
452 is_tmux
= bool(os
.getenv("TMUX", ""))
453 is_screen
= not is_tmux
and bool(os
.getenv("STY", ""))
454 is_xterm
= not is_tmux
and not is_screen
and bool(os
.getenv("DISPLAY", ""))
458 wait_for_channels
= []
460 # Really would like something better than using this global here.
461 # Not all tests use topogen though so get_topogen() won't work.
462 for node
in Mininet
.g_mnet_inst
.hosts
.values():
467 "{}-{}".format(os
.getpid(), Commander
.tmux_wait_gen
)
471 Commander
.tmux_wait_gen
+= 1
472 wait_for_channels
.append(channel
)
474 pane_info
= node
.run_in_window(
476 new_window
=win_info
is None,
478 title
="{} ({})".format(title
, node
.name
),
480 tmux_target
=win_info
,
487 assert isinstance(pane_info
, subprocess
.Popen
)
488 wait_for_procs
.append(pane_info
)
490 # Now wait on any channels
491 for channel
in wait_for_channels
:
492 logger
.debug("Waiting on TMUX channel %s", channel
)
493 commander
.cmd_raises([commander
.get_exec_path("tmux"), "wait", channel
])
494 for p
in wait_for_procs
:
495 logger
.debug("Waiting on TMUX xterm process %s", p
)
496 o
, e
= p
.communicate()
498 logger
.warning("xterm proc failed: %s:", proc_error(p
, o
, e
))
500 if error
and topotest_extra_config
["cli_on_error"]:
501 # Really would like something better than using this global here.
502 # Not all tests use topogen though so get_topogen() won't work.
503 if Mininet
.g_mnet_inst
:
504 cli(Mininet
.g_mnet_inst
, title
=title
, background
=False)
506 logger
.error("Could not launch CLI b/c no mininet exists yet")
508 while pause
and isatty
:
511 'PAUSED, "cli" for CLI, "pdb" to debug, "Enter" to continue: '
514 user
= input('PAUSED, "cli" for CLI, "pdb" to debug, "Enter" to continue: ')
518 cli(Mininet
.g_mnet_inst
)
520 pdb
.set_trace() # pylint: disable=forgotten-debug-statement
522 print('Unrecognized input: "%s"' % user
)
528 # Add common fixtures available to all tests as parameters
531 tgen
= pytest
.fixture(lib
.fixtures
.tgen
)
532 topo
= pytest
.fixture(lib
.fixtures
.topo
)