from __future__ import unicode_literals import struct import sys import re import logging import argparse import subprocess import ipaddress logging.basicConfig() log = logging.getLogger("main") class TestStuff(object): def __init__(self): pass def test_ping6_switch(self): host = "h1" dst_ipv6 = ipaddress.ip_address("2001:db8::42") log.info("Trying to reach {} from {}".format(dst_ipv6, host)) cmd = "mx {} ping6 -c1 {}".format(host, dst_ipv6).split(" ") return cmd def test_ping4_switch(self): host = "h3" dst = ipaddress.ip_address("10.0.0.66") log.info("Trying to reach {} from {}".format(dst, host)) cmd = "mx {} ping -c1 {}".format(host, dst).split(" ") return cmd def test_v6_static_mapping(self): host = "h1" nat64_prefix = ipaddress.ip_network("2001:db8:1::/96") dst_ipv4 = ipaddress.ip_address("10.0.0.1") translated_ipv4 = nat64_prefix[int(dst_ipv4)] # '2001:db8:1::a00:1' log.info("Trying to reach {} ({}) from {}".format(dst_ipv4, translated_ipv4, host)) cmd = "mx {} ping6 -c1 {}".format(host, translated_ipv4).split(" ") return cmd def test_v4_static_mapping(self): host = "h3" dst_ipv4 = ipaddress.ip_address("10.1.1.1") log.info("Trying to reach {} from {}".format(dst_ipv4, host)) cmd = "mx {} ping -c1 {}".format(host, dst_ipv4).split(" ") return cmd def test_v6_udp_to_v4(self): print('mx h3 "echo V4-OK | socat - UDP-LISTEN:2342"') print('mx h1 "echo V6-OK | socat - UDP6:[2001:db8:1::a00:1]:2342"') return def test_v6_tcp_to_v4(self): print('mx h3 "echo V4-OK | socat - TCP-LISTEN:2342"') print('mx h1 "echo V6-OK | socat - TCP6:[2001:db8:1::a00:1]:2342"') return def test_v4_tcp_to_v6(self): print('mx h3 "echo V4-OK | socat - TCP:10.1.1.1:2342"') print('mx h1 "echo V6-OK | socat - TCP-LISTEN:2342"') return def test_v4_udp_to_v6(self): print('mx h3 "echo V4-OK | socat - UDP:10.1.1.1:2342"') print('mx h1 "echo V6-OK | socat - UDP-LISTEN:2342"') return def commandline(self): parser = argparse.ArgumentParser(description='controller++') methods_dir = dir(self) methods = [m for m in methods_dir if re.match("^test", m)] methods = [re.sub("^test_(.*)", r"\1", m) for m in methods] parser.add_argument('-m', '--method', help="which method?", choices=methods, required=True) parser.add_argument('--debug', help='Enable debug logging', action='store_true') parser.add_argument('--verbose', help='Enable verbose logging', action='store_true') parser.add_argument('--multicast-to-controller', help='Send debug multicast to controller', action='store_true') self.args = parser.parse_args() if self.args.debug: log.setLevel(logging.DEBUG) elif self.args.verbose: log.setLevel(logging.INFO) else: log.setLevel(logging.WARNING) f = getattr(self, "test_{}".format(self.args.method)) subprocess.call(f()) if __name__ == "__main__": import sys import os log.info("Booting...") log.debug("Debug enabled.") t = TestStuff() t.commandline()