test/test_vapi.py

   1 #!/usr/bin/env python
   2 """ VAPI test """
   3
   4 from __future__ import division
   5 import unittest
   6 import os
   7 import signal
   8 import subprocess
   9 from threading import Thread
  10 from log import single_line_delim
  11 from framework import VppTestCase, running_extended_tests, \
  12     running_on_centos, VppTestRunner
  13
  14
  15 class Worker(Thread):
  16     def __init__(self, args, logger):
  17         self.logger = logger
  18         self.args = args
  19         self.result = None
  20         super(Worker, self).__init__()
  21
  22     def run(self):
  23         executable = self.args[0]
  24         self.logger.debug("Running executable w/args `%s'" % self.args)
  25         env = os.environ.copy()
  26         env["CK_LOG_FILE_NAME"] = "-"
  27         self.process = subprocess.Popen(
  28             self.args, shell=False, env=env, preexec_fn=os.setpgrp,
  29             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  30         out, err = self.process.communicate()
  31         self.logger.debug("Finished running `%s'" % executable)
  32         self.logger.info("Return code is `%s'" % self.process.returncode)
  33         self.logger.info(single_line_delim)
  34         self.logger.info("Executable `%s' wrote to stdout:" % executable)
  35         self.logger.info(single_line_delim)
  36         self.logger.info(out)
  37         self.logger.info(single_line_delim)
  38         self.logger.info("Executable `%s' wrote to stderr:" % executable)
  39         self.logger.info(single_line_delim)
  40         self.logger.error(err)
  41         self.logger.info(single_line_delim)
  42         self.result = self.process.returncode
  43
  44
  45 @unittest.skipUnless(running_extended_tests(), "part of extended tests")
  46 class VAPITestCase(VppTestCase):
  47     """ VAPI test """
  48
  49     def test_vapi_c(self):
  50         """ run C VAPI tests """
  51         var = "BR"
  52         built_root = os.getenv(var, None)
  53         self.assertIsNotNone(built_root,
  54                              "Environment variable `%s' not set" % var)
  55         executable = "%s/vapi_test/vapi_c_test" % built_root
  56         worker = Worker(
  57             [executable, "vapi client", self.shm_prefix], self.logger)
  58         worker.start()
  59         timeout = 60
  60         worker.join(timeout)
  61         self.logger.info("Worker result is `%s'" % worker.result)
  62         error = False
  63         if worker.result is None:
  64             try:
  65                 error = True
  66                 self.logger.error(
  67                     "Timeout! Worker did not finish in %ss" % timeout)
  68                 os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM)
  69                 worker.join()
  70             except:
  71                 raise Exception("Couldn't kill worker-spawned process")
  72         if error:
  73             raise Exception(
  74                 "Timeout! Worker did not finish in %ss" % timeout)
  75         self.assert_equal(worker.result, 0, "Binary test return code")
  76
  77     @unittest.skipIf(running_on_centos(), "Centos's gcc can't compile our C++")
  78     def test_vapi_cpp(self):
  79         """ run C++ VAPI tests """
  80         var = "BR"
  81         built_root = os.getenv(var, None)
  82         self.assertIsNotNone(built_root,
  83                              "Environment variable `%s' not set" % var)
  84         executable = "%s/vapi_test/vapi_cpp_test" % built_root
  85         worker = Worker(
  86             [executable, "vapi client", self.shm_prefix], self.logger)
  87         worker.start()
  88         timeout = 120
  89         worker.join(timeout)
  90         self.logger.info("Worker result is `%s'" % worker.result)
  91         error = False
  92         if worker.result is None:
  93             try:
  94                 error = True
  95                 self.logger.error(
  96                     "Timeout! Worker did not finish in %ss" % timeout)
  97                 os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM)
  98                 worker.join()
  99             except:
 100                 raise Exception("Couldn't kill worker-spawned process")
 101         if error:
 102             raise Exception(
 103                 "Timeout! Worker did not finish in %ss" % timeout)
 104         self.assert_equal(worker.result, 0, "Binary test return code")
 105
 106
 107 if __name__ == '__main__':
 108     unittest.main(testRunner=VppTestRunner)