test/test_vapi.py

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