test/test_vapi.py

   1 #!/usr/bin/env python
   2 """ VAPI test """
   3
   4 import unittest
   5 import os
   6 import signal
   7 from framework import VppTestCase, running_extended_tests, \
   8     running_on_centos, VppTestRunner, Worker
   9
  10
  11 @unittest.skipUnless(running_extended_tests(), "part of extended tests")
  12 class VAPITestCase(VppTestCase):
  13     """ VAPI test """
  14
  15     def test_vapi_c(self):
  16         """ run C VAPI tests """
  17         var = "BR"
  18         built_root = os.getenv(var, None)
  19         self.assertIsNotNone(built_root,
  20                              "Environment variable `%s' not set" % var)
  21         executable = "%s/vapi_test/vapi_c_test" % built_root
  22         worker = Worker(
  23             [executable, "vapi client", self.shm_prefix], self.logger)
  24         worker.start()
  25         timeout = 60
  26         worker.join(timeout)
  27         self.logger.info("Worker result is `%s'" % worker.result)
  28         error = False
  29         if worker.result is None:
  30             try:
  31                 error = True
  32                 self.logger.error(
  33                     "Timeout! Worker did not finish in %ss" % timeout)
  34                 os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM)
  35                 worker.join()
  36             except:
  37                 self.logger.debug("Couldn't kill worker-spawned process")
  38                 raise
  39         if error:
  40             raise Exception(
  41                 "Timeout! Worker did not finish in %ss" % timeout)
  42         self.assert_equal(worker.result, 0, "Binary test return code")
  43
  44     @unittest.skipIf(running_on_centos(), "Centos's gcc can't compile our C++")
  45     def test_vapi_cpp(self):
  46         """ run C++ VAPI tests """
  47         var = "BR"
  48         built_root = os.getenv(var, None)
  49         self.assertIsNotNone(built_root,
  50                              "Environment variable `%s' not set" % var)
  51         executable = "%s/vapi_test/vapi_cpp_test" % built_root
  52         worker = Worker(
  53             [executable, "vapi client", self.shm_prefix], self.logger)
  54         worker.start()
  55         timeout = 120
  56         worker.join(timeout)
  57         self.logger.info("Worker result is `%s'" % worker.result)
  58         error = False
  59         if worker.result is None:
  60             try:
  61                 error = True
  62                 self.logger.error(
  63                     "Timeout! Worker did not finish in %ss" % timeout)
  64                 os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM)
  65                 worker.join()
  66             except:
  67                 raise Exception("Couldn't kill worker-spawned process")
  68         if error:
  69             raise Exception(
  70                 "Timeout! Worker did not finish in %ss" % timeout)
  71         self.assert_equal(worker.result, 0, "Binary test return code")
  72
  73
  74 if __name__ == '__main__':
  75     unittest.main(testRunner=VppTestRunner)