test/test_container.py

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