test/test_vcl.py

   1 #!/usr/bin/env python
   2 """ Vpp VCL tests """
   3
   4 import unittest
   5 import os
   6 import signal
   7 import subprocess
   8 from threading import Thread
   9 from log import single_line_delim
  10 from framework import VppTestCase, running_extended_tests, \
  11     running_on_centos, VppTestRunner, Worker
  12 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
  13
  14
  15 class VCLCUTTHRUTestCase(VppTestCase):
  16     """ VCL Cut Thru Test """
  17
  18     server_addr = "127.0.0.1"
  19     server_port = "22000"
  20
  21     @classmethod
  22     def setUpClass(cls):
  23         super(VCLCUTTHRUTestCase, cls).setUpClass()
  24
  25     def setUp(self):
  26         super(VCLCUTTHRUTestCase, self).setUp()
  27
  28         self.vapi.session_enable_disable(is_enabled=1)
  29
  30     def tearDown(self):
  31         self.vapi.session_enable_disable(is_enabled=0)
  32
  33         super(VCLCUTTHRUTestCase, self).tearDown()
  34
  35     def test_vcl_cutthru(self):
  36         """ run VCL cut-thru test """
  37         timeout = 5
  38         var = "VPP_TEST_BUILD_DIR"
  39         build_dir = os.getenv(var, None)
  40         self.assertIsNotNone(build_dir,
  41                              "Environment variable `%s' not set" % var)
  42         vcl_exe_dir = "%s/vpp/.libs" % build_dir
  43         executable = "%s/vcl_test_server" % vcl_exe_dir
  44         worker_server = Worker([executable, self.server_port], self.logger)
  45         worker_server.env["VCL_API_PREFIX"] = self.shm_prefix
  46         worker_server.env["VCL_APP_SCOPE_LOCAL"] = "true"
  47         worker_server.start()
  48         executable = "%s/vcl_test_client" % vcl_exe_dir
  49         worker_client = Worker(
  50             [executable, self.server_addr, self.server_port,
  51              "-E", "Hello, world!", "-X"], self.logger)
  52         worker_client.env["VCL_API_PREFIX"] = self.shm_prefix
  53         worker_client.env["VCL_APP_SCOPE_LOCAL"] = "true"
  54         worker_client.start()
  55         worker_client.join(timeout)
  56         self.logger.info("Client worker result is `%s'" % worker_client.result)
  57         error = False
  58         if worker_client.result is None:
  59             try:
  60                 error = True
  61                 self.logger.error(
  62                     "Timeout! Client worker did not finish in %ss" % timeout)
  63                 os.killpg(os.getpgid(worker_client.process.pid),
  64                           signal.SIGTERM)
  65                 worker_client.join()
  66             except:
  67                 raise Exception("Couldn't kill client worker-spawned process")
  68         if error:
  69             os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM)
  70             worker_server.join()
  71             raise Exception(
  72                 "Timeout! Client worker did not finish in %ss" % timeout)
  73         self.assert_equal(worker_client.result, 0, "Binary test return code")
  74
  75
  76 class VCLTHRUHSTestcase(VppTestCase):
  77     """ VCL Thru Hoststack Test """
  78
  79     server_port = "22000"
  80
  81     @classmethod
  82     def setUpClass(cls):
  83         super(VCLTHRUHSTestcase, cls).setUpClass()
  84
  85     def setUp(self):
  86         super(VCLTHRUHSTestcase, self).setUp()
  87
  88         self.vapi.session_enable_disable(is_enabled=1)
  89         self.create_loopback_interfaces(range(2))
  90
  91         table_id = 0
  92
  93         for i in self.lo_interfaces:
  94             i.admin_up()
  95
  96             if table_id != 0:
  97                 tbl = VppIpTable(self, table_id)
  98                 tbl.add_vpp_config()
  99
 100             i.set_table_ip4(table_id)
 101             i.config_ip4()
 102             table_id += 1
 103
 104         # Configure namespaces
 105         self.vapi.app_namespace_add(namespace_id="0",
 106                                     sw_if_index=self.loop0.sw_if_index)
 107         self.vapi.app_namespace_add(namespace_id="1",
 108                                     sw_if_index=self.loop1.sw_if_index)
 109
 110     def tearDown(self):
 111         for i in self.lo_interfaces:
 112             i.unconfig_ip4()
 113             i.set_table_ip4(0)
 114             i.admin_down()
 115
 116         self.vapi.session_enable_disable(is_enabled=0)
 117         super(VCLTHRUHSTestcase, self).tearDown()
 118
 119     def test_vcl_thru_hoststack(self):
 120         """ run VCL thru hoststack test """
 121         # Add inter-table routes
 122         ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
 123                             [VppRoutePath("0.0.0.0",
 124                                           0xffffffff,
 125                                           nh_table_id=1)])
 126         ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
 127                             [VppRoutePath("0.0.0.0",
 128                                           0xffffffff,
 129                                           nh_table_id=0)], table_id=1)
 130         ip_t01.add_vpp_config()
 131         ip_t10.add_vpp_config()
 132
 133         timeout = 20
 134         var = "VPP_TEST_BUILD_DIR"
 135         build_dir = os.getenv(var, None)
 136         self.assertIsNotNone(build_dir,
 137                              "Environment variable `%s' not set" % var)
 138         vcl_exe_dir = "%s/vpp/.libs" % build_dir
 139         executable = "%s/vcl_test_server" % vcl_exe_dir
 140         worker_server = Worker([executable, self.server_port], self.logger)
 141         worker_server.env["VCL_API_PREFIX"] = self.shm_prefix
 142         worker_server.env["VCL_APP_SCOPE_GLOBAL"] = "true"
 143         worker_server.env["VCL_APP_NAMESPACE_ID"] = "0"
 144         worker_server.start()
 145         executable = "%s/vcl_test_client" % vcl_exe_dir
 146         worker_client = Worker(
 147             [executable, self.loop0.local_ip4, self.server_port,
 148              "-E", "Hello, world!", "-X"], self.logger)
 149         worker_client.env["VCL_API_PREFIX"] = self.shm_prefix
 150         worker_client.env["VCL_APP_SCOPE_GLOBAL"] = "true"
 151         worker_client.env["VCL_APP_NAMESPACE_ID"] = "1"
 152         worker_client.start()
 153         worker_client.join(timeout)
 154         self.logger.info("Client worker result is `%s'" % worker_client.result)
 155         error = False
 156         if worker_client.result is None:
 157             try:
 158                 error = True
 159                 self.logger.error(
 160                     "Timeout! Client worker did not finish in %ss" % timeout)
 161                 os.killpg(os.getpgid(worker_client.process.pid),
 162                           signal.SIGTERM)
 163                 worker_client.join()
 164             except:
 165                 raise Exception("Couldn't kill client worker-spawned process")
 166         if error:
 167             os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM)
 168             worker_server.join()
 169             raise Exception(
 170                 "Timeout! Client worker did not finish in %ss" % timeout)
 171         self.assert_equal(worker_client.result, 0, "Binary test return code")
 172
 173 if __name__ == '__main__':
 174     unittest.main(testRunner=VppTestRunner)