summaryrefslogtreecommitdiffstats
path: root/test/test_ct6.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_ct6.py')
-rw-r--r--test/test_ct6.py72
1 files changed, 72 insertions, 0 deletions
diff --git a/test/test_ct6.py b/test/test_ct6.py
new file mode 100644
index 00000000000..f4d32856947
--- /dev/null
+++ b/test/test_ct6.py
@@ -0,0 +1,72 @@
+from config import config
+from framework import VppTestCase
+from asfframework import VppTestRunner
+import unittest
+from scapy.layers.l2 import Ether
+from scapy.layers.inet import UDP
+from scapy.layers.inet6 import IPv6
+from random import randint
+
+
+# TODO: get an actual output from "show ip6 connection-tracker"
+@unittest.skipIf("ct6" in config.excluded_plugins, "Exclude CT6 plugin tests")
+class TestCt6(VppTestCase):
+ """CT6 plugin tests"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestCt6, cls).setUpClass()
+ try:
+ cls.create_pg_interfaces(range(2))
+ for i in cls.pg_interfaces:
+ i.config_ip6().resolve_ndp()
+ i.admin_up()
+ except Exception:
+ cls.tearDownClass()
+ raise
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestCt6, cls).tearDownClass()
+
+ def create_stream(self, src_if, dst_if, count):
+ packets = []
+ for i in range(count):
+ p = (
+ Ether(src=src_if.remote_mac, dst=src_if.local_mac)
+ / IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
+ / UDP(sport=randint(49152, 65535), dport=5678, chksum=0)
+ )
+ packets.append(p)
+ return packets
+
+ def test_ct6_vapi(self):
+ self.vapi.ct6_enable_disable(enable_disable=True, is_inside=True, sw_if_index=1)
+ self.vapi.ct6_enable_disable(
+ enable_disable=True, is_inside=False, sw_if_index=1
+ )
+
+ packets = self.create_stream(self.pg0, self.pg1, 5)
+ self.pg0.add_stream(packets)
+ self.pg_start()
+
+ self.vapi.ct6_enable_disable(1234, 567, True, True, "pg0", "disable")
+
+ def test_ct6_cli(self):
+ self.vapi.cli("set ct6 outside pg1")
+ self.vapi.cli("set ct6 inside pg1")
+
+ packets = self.create_stream(self.pg0, self.pg1, 5)
+ self.pg0.add_stream(packets)
+ self.pg_start()
+
+ reply = self.vapi.cli("show ip6 connection-tracker")
+ self.assertIn("Thread 0", reply)
+ reply = self.vapi.cli("test ip6 connection-tracker")
+ self.assertIn("End state", reply)
+
+ self.vapi.cli("set ct6 inside pg1 disable")
+
+
+if __name__ == "__main__":
+ unittest.main(testRunner=VppTestRunner)