test/test_vapi.py

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