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