test/asf/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 asfframework import VppAsfTestCase, VppTestRunner, Worker
   9
  10
  11 class VAPITestCase(VppAsfTestCase):
  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 run_vapi_c(self, path, transport):
  23         executable = f"{config.vpp_build_dir}/vpp/bin/vapi_c_test"
  24         worker = Worker([executable, "vapi client", path, transport], self.logger)
  25         worker.start()
  26         timeout = 60
  27         worker.join(timeout)
  28         self.logger.info("Worker result is `%s'" % worker.result)
  29         error = False
  30         if worker.result is None:
  31             try:
  32                 error = True
  33                 self.logger.error("Timeout! Worker did not finish in %ss" % timeout)
  34                 os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM)
  35                 worker.join()
  36             except:
  37                 self.logger.debug("Couldn't kill worker-spawned process")
  38                 raise
  39         if error:
  40             raise Exception("Timeout! Worker did not finish in %ss" % timeout)
  41         self.assert_equal(worker.result, 0, "Binary test return code")
  42
  43     def test_vapi_c_shm(self):
  44         """run C VAPI tests (over shared memory)"""
  45         self.run_vapi_c(self.get_api_segment_prefix(), "shm")
  46
  47     def test_vapi_c_uds(self):
  48         """run C VAPI tests (over unix domain socket)"""
  49         self.run_vapi_c(self.get_api_sock_path(), "uds")
  50
  51     def run_vapi_cpp(self, path, transport):
  52         """run C++ VAPI tests"""
  53         executable = f"{config.vpp_build_dir}/vpp/bin/vapi_cpp_test"
  54         worker = Worker([executable, "vapi client", path, transport], self.logger)
  55         worker.start()
  56         timeout = 120
  57         worker.join(timeout)
  58         self.logger.info("Worker result is `%s'" % worker.result)
  59         error = False
  60         if worker.result is None:
  61             try:
  62                 error = True
  63                 self.logger.error("Timeout! Worker did not finish in %ss" % timeout)
  64                 os.killpg(os.getpgid(worker.process.pid), signal.SIGTERM)
  65                 worker.join()
  66             except:
  67                 raise Exception("Couldn't kill worker-spawned process")
  68         if error:
  69             raise Exception("Timeout! Worker did not finish in %ss" % timeout)
  70         self.assert_equal(worker.result, 0, "Binary test return code")
  71
  72     def test_vapi_cpp_shm(self):
  73         """run C++ VAPI tests (over shared memory)"""
  74         self.run_vapi_cpp(self.get_api_segment_prefix(), "shm")
  75
  76     def test_vapi_cpp_uds(self):
  77         """run C++ VAPI tests (over unix domain socket)"""
  78         self.run_vapi_cpp(self.get_api_sock_path(), "uds")
  79
  80
  81 if __name__ == "__main__":
  82     unittest.main(testRunner=VppTestRunner)