test/test_vapi.py

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