help="Pause after each test",
)
+ parser.addoption(
+ "--pause-at-end",
+ action="store_true",
+ help="Pause before taking munet down",
+ )
+
parser.addoption(
"--pause-on-error",
action="store_true",
help="Do not pause after (disables default when --shell or -vtysh given)",
)
+ parser.addoption(
+ "--pcap",
+ default="",
+ metavar="NET[,NET...]",
+ help="Comma-separated list of networks to capture packets on, or 'all'",
+ )
+
rundir_help = "directory for running in and log files"
parser.addini("rundir", rundir_help, default="/tmp/topotests")
parser.addoption("--rundir", metavar="DIR", help=rundir_help)
def start(self):
"""Start the micronet topology."""
+ pcapopt = self.cfgopt.get_option_list("--pcap")
+ if "all" in pcapopt:
+ pcapopt = self.switches.keys()
+ for pcap in pcapopt:
+ if ":" in pcap:
+ host, intf = pcap.split(":")
+ pcap = f"{host}-{intf}"
+ host = self.hosts[host]
+ else:
+ host = self
+ intf = pcap
+ pcapfile = f"{self.rundir}/capture-{pcap}.pcap"
+ host.run_in_window(
+ f"tshark -s 9200 -i {intf} -P -w {pcapfile}",
+ background=True,
+ title=f"cap:{pcap}",
+ )
+
self.logger.debug("%s: Starting (no-op).", self)
def stop(self):
from lib.micronet import Commander
from lib.micronet_compat import Mininet
from lib.topolog import logger
+from munet.testing.util import pause_test
from lib import topotest
first is a simple kill with no sleep, the second will sleep if not
killed and try with a different signal.
"""
+ pause = bool(self.net.cfgopt.get_option("--pause-at-end"))
+ pause = pause or bool(self.net.cfgopt.get_option("--pause"))
+ if pause:
+ try:
+ pause_test("Before MUNET delete")
+ except KeyboardInterrupt:
+ print("^C...continuing")
+ except Exception as error:
+ self.logger.error("\n...continuing after error: %s", error)
+
logger.info("stopping topology: {}".format(self.modname))
+
errors = ""
for gear in self.gears.values():
errors += gear.stop()