test/test_ct6.py

   1 from config import config
   2 from framework import VppTestCase
   3 from asfframework import VppTestRunner
   4 import unittest
   5 from scapy.layers.l2 import Ether
   6 from scapy.layers.inet import UDP
   7 from scapy.layers.inet6 import IPv6
   8 from random import randint
   9
  10
  11 # TODO: get an actual output from "show ip6 connection-tracker"
  12 @unittest.skipIf("ct6" in config.excluded_plugins, "Exclude CT6 plugin tests")
  13 class TestCt6(VppTestCase):
  14     """CT6 plugin tests"""
  15
  16     @classmethod
  17     def setUpClass(cls):
  18         super(TestCt6, cls).setUpClass()
  19         try:
  20             cls.create_pg_interfaces(range(2))
  21             for i in cls.pg_interfaces:
  22                 i.config_ip6().resolve_ndp()
  23                 i.admin_up()
  24         except Exception:
  25             cls.tearDownClass()
  26             raise
  27
  28     @classmethod
  29     def tearDownClass(cls):
  30         super(TestCt6, cls).tearDownClass()
  31
  32     def create_stream(self, src_if, dst_if, count):
  33         packets = []
  34         for i in range(count):
  35             p = (
  36                 Ether(src=src_if.remote_mac, dst=src_if.local_mac)
  37                 / IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)
  38                 / UDP(sport=randint(49152, 65535), dport=5678, chksum=0)
  39             )
  40             packets.append(p)
  41         return packets
  42
  43     def test_ct6_vapi(self):
  44         self.vapi.ct6_enable_disable(enable_disable=True, is_inside=True, sw_if_index=1)
  45         self.vapi.ct6_enable_disable(
  46             enable_disable=True, is_inside=False, sw_if_index=1
  47         )
  48
  49         packets = self.create_stream(self.pg0, self.pg1, 5)
  50         self.pg0.add_stream(packets)
  51         self.pg_start()
  52
  53         self.vapi.ct6_enable_disable(1234, 567, True, True, "pg0", "disable")
  54
  55     def test_ct6_cli(self):
  56         self.vapi.cli("set ct6 outside pg1")
  57         self.vapi.cli("set ct6 inside pg1")
  58
  59         packets = self.create_stream(self.pg0, self.pg1, 5)
  60         self.pg0.add_stream(packets)
  61         self.pg_start()
  62
  63         reply = self.vapi.cli("show ip6 connection-tracker")
  64         self.assertIn("Thread 0", reply)
  65         reply = self.vapi.cli("test ip6 connection-tracker")
  66         self.assertIn("End state", reply)
  67
  68         self.vapi.cli("set ct6 inside pg1 disable")
  69
  70
  71 if __name__ == "__main__":
  72     unittest.main(testRunner=VppTestRunner)