115 lines
3.3 KiB
Python
Executable file
115 lines
3.3 KiB
Python
Executable file
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()
|