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
  13
  14 class VCLTestCase(VppTestCase):
  15     """ VPP Communications Library Test """
  16
  17     server_addr = "127.0.0.1"
  18     server_port = "22000"
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         super(VCLTestCase, cls).setUpClass()
  23
  24         cls.vapi.session_enable_disable(is_enabled=1)
  25
  26     def setUp(self):
  27         super(VCLTestCase, self).setUp()
  28
  29     def test_vcl_cutthru(self):
  30         """ run VCL cut-thru test """
  31         timeout = 5
  32         var = "VPP_TEST_BUILD_DIR"
  33         build_dir = os.getenv(var, None)
  34         self.assertIsNotNone(build_dir,
  35                              "Environment variable `%s' not set" % var)
  36         vcl_exe_dir = "%s/vpp/.libs" % build_dir
  37         executable = "%s/vcl_test_server" % vcl_exe_dir
  38         worker_server = Worker([executable, self.server_port], self.logger)
  39         worker_server.env["VCL_API_PREFIX"] = self.shm_prefix
  40 #        worker_server.env["VCL_DEBUG"] = "2"
  41         worker_server.env["VCL_APP_SCOPE_LOCAL"] = "true"
  42         worker_server.start()
  43         executable = "%s/vcl_test_client" % vcl_exe_dir
  44         worker_client = Worker(
  45             [executable, self.server_addr, self.server_port,
  46              "-E", "Hello, world!", "-X"], self.logger)
  47         worker_client.env["VCL_API_PREFIX"] = self.shm_prefix
  48 #        worker_client.env["VCL_DEBUG"] = "2"
  49         worker_client.env["VCL_APP_SCOPE_LOCAL"] = "true"
  50         worker_client.start()
  51         worker_client.join(timeout)
  52         self.logger.info("Client worker result is `%s'" % worker_client.result)
  53         error = False
  54         if worker_client.result is None:
  55             try:
  56                 error = True
  57                 self.logger.error(
  58                     "Timeout! Client worker did not finish in %ss" % timeout)
  59                 os.killpg(os.getpgid(worker_client.process.pid),
  60                           signal.SIGTERM)
  61                 worker_client.join()
  62             except:
  63                 raise Exception("Couldn't kill client worker-spawned process")
  64         if error:
  65             os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM)
  66             worker_server.join()
  67             raise Exception(
  68                 "Timeout! Client worker did not finish in %ss" % timeout)
  69         self.assert_equal(worker_client.result, 0, "Binary test return code")
  70
  71
  72 if __name__ == '__main__':
  73     unittest.main(testRunner=VppTestRunner)