test/test_container.py

   1 #!/usr/bin/env python3
   2 """ Container integration tests """
   3
   4 import unittest
   5 from framework import VppTestCase, VppTestRunner, running_extended_tests
   6 from scapy.layers.l2 import Ether
   7 from scapy.packet import Raw
   8 from scapy.layers.inet import IP, UDP, TCP
   9 from scapy.packet import Packet
  10 from socket import inet_pton, AF_INET, AF_INET6
  11 from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest
  12 from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting
  13 from scapy.layers.inet6 import IPv6ExtHdrFragment
  14 from pprint import pprint
  15 from random import randint
  16 from util import L4_Conn
  17
  18
  19 class Conn(L4_Conn):
  20     # for now same as L4_Conn
  21     pass
  22
  23
  24 @unittest.skipUnless(running_extended_tests, "part of extended tests")
  25 class ContainerIntegrationTestCase(VppTestCase):
  26     """ Container integration extended testcases """
  27
  28     @classmethod
  29     def setUpClass(cls):
  30         super(ContainerIntegrationTestCase, cls).setUpClass()
  31         # create pg0 and pg1
  32         cls.create_pg_interfaces(range(2))
  33         for i in cls.pg_interfaces:
  34             i.admin_up()
  35             i.config_ip4()
  36             i.config_ip6()
  37             i.resolve_arp()
  38             i.resolve_ndp()
  39
  40     @classmethod
  41     def tearDownClass(cls):
  42         super(ContainerIntegrationTestCase, cls).tearDownClass()
  43
  44     def tearDown(self):
  45         """Run standard test teardown and log various show commands
  46         """
  47         super(ContainerIntegrationTestCase, self).tearDown()
  48
  49     def show_commands_at_teardown(self):
  50         self.logger.info(self.vapi.cli("show ip neighbors"))
  51
  52     def run_basic_conn_test(self, af, acl_side):
  53         """ Basic connectivity test """
  54         conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242)
  55         conn1.send_through(0)
  56         # the return packets should pass
  57         conn1.send_through(1)
  58
  59     def run_negative_conn_test(self, af, acl_side):
  60         """ Packets with local spoofed address """
  61         conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242)
  62         try:
  63             p2 = conn1.send_through(0).command()
  64         except:
  65             # If we asserted while waiting, it's good.
  66             # the conn should have timed out.
  67             p2 = None
  68         self.assert_equal(p2, None, ": packet should have been dropped")
  69
  70     def test_0010_basic_conn_test(self):
  71         """ IPv4 basic connectivity test """
  72         self.run_basic_conn_test(AF_INET, 0)
  73
  74     def test_0011_basic_conn_test(self):
  75         """ IPv6 basic connectivity test """
  76         self.run_basic_conn_test(AF_INET6, 0)
  77
  78     def test_0050_loopback_prepare_test(self):
  79         """ Create loopbacks overlapping with remote addresses """
  80         self.create_loopback_interfaces(2)
  81         for i in range(2):
  82             intf = self.lo_interfaces[i]
  83             intf.admin_up()
  84             intf.local_ip4 = self.pg_interfaces[i].remote_ip4
  85             intf.local_ip4_prefix_len = 32
  86             intf.config_ip4()
  87             intf.local_ip6 = self.pg_interfaces[i].remote_ip6
  88             intf.local_ip6_prefix_len = 128
  89             intf.config_ip6()
  90
  91     def test_0110_basic_conn_test(self):
  92         """ IPv4 local-spoof connectivity test """
  93         self.run_negative_conn_test(AF_INET, 0)
  94
  95     def test_0111_basic_conn_test(self):
  96         """ IPv6 local-spoof connectivity test """
  97         self.run_negative_conn_test(AF_INET, 1)
  98
  99     def test_0200_basic_conn_test(self):
 100         """ Configure container commands """
 101         for i in range(2):
 102             for addr in [self.pg_interfaces[i].remote_ip4,
 103                          self.pg_interfaces[i].remote_ip6]:
 104                 self.vapi.ppcli("ip container " + addr + " " +
 105                                 self.pg_interfaces[i].name)
 106                 self.vapi.ppcli("stn rule address " + addr +
 107                                 " interface " + self.pg_interfaces[i].name)
 108
 109     def test_0210_basic_conn_test(self):
 110         """ IPv4 test after configuring container """
 111         self.run_basic_conn_test(AF_INET, 0)
 112
 113     def test_0211_basic_conn_test(self):
 114         """ IPv6 test after configuring container """
 115         self.run_basic_conn_test(AF_INET, 1)
 116
 117     def test_0300_unconfigure_commands(self):
 118         """ Unconfigure container commands """
 119         for i in range(2):
 120             for addr in [self.pg_interfaces[i].remote_ip4,
 121                          self.pg_interfaces[i].remote_ip6]:
 122                 self.vapi.ppcli("ip container " + addr + " " +
 123                                 self.pg_interfaces[i].name +
 124                                 " del")
 125                 self.vapi.ppcli("stn rule address " + addr +
 126                                 " interface " + self.pg_interfaces[i].name +
 127                                 " del")
 128
 129     def test_0410_spoof_test(self):
 130         """ IPv4 local-spoof after unconfig test """
 131         self.run_negative_conn_test(AF_INET, 0)
 132
 133     def test_0411_spoof_test(self):
 134         """ IPv6 local-spoof after unconfig test """
 135         self.run_negative_conn_test(AF_INET, 1)