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(
  26             [executable, "vapi client", self.get_api_segment_prefix()], self.logger
  27         )
  28         worker.start()
  29         timeout = 60
  30         worker.join(timeout)
  31         self.logger.info("Worker result is `%s'" % worker.result)
  32         error = False
  33         if worker.result is None:
  34             try:
  35                 error = True
  36                 self.logger.error("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("Timeout! Worker did not finish in %ss" % timeout)
  44         self.assert_equal(worker.result, 0, "Binary test return code")
  45
  46     def test_vapi_cpp(self):
  47         """run C++ VAPI tests"""
  48         executable = f"{config.vpp_build_dir}/vpp/bin/vapi_cpp_test"
  49         worker = Worker(
  50             [executable, "vapi client", self.get_api_segment_prefix()], self.logger
  51         )
  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("Timeout! Worker did not finish in %ss" % timeout)
  61                 os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM)
  62                 worker.join()
  63             except:
  64                 raise Exception("Couldn't kill worker-spawned process")
  65         if error:
  66             raise Exception("Timeout! Worker did not finish in %ss" % timeout)
  67         self.assert_equal(worker.result, 0, "Binary test return code")
  68
  69
  70 if __name__ == "__main__":
  71     unittest.main(testRunner=VppTestRunner)