test/test_stats_client.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4 import psutil
   5 from vpp_papi.vpp_stats import VPPStats
   6
   7 from framework import tag_fixme_vpp_workers
   8 from framework import VppTestCase, VppTestRunner
   9
  10
  11 @tag_fixme_vpp_workers
  12 class StatsClientTestCase(VppTestCase):
  13     """Test Stats Client"""
  14
  15     @classmethod
  16     def setUpClass(cls):
  17         super(StatsClientTestCase, cls).setUpClass()
  18
  19     @classmethod
  20     def tearDownClass(cls):
  21         super(StatsClientTestCase, cls).tearDownClass()
  22
  23     def test_set_errors(self):
  24         """Test set errors"""
  25         self.assertEqual(self.statistics.set_errors(), {})
  26         self.assertEqual(self.statistics.get_counter('/err/ethernet-input/no'),
  27                          [0])
  28
  29     def test_client_fd_leak(self):
  30         """Test file descriptor count - VPP-1486"""
  31
  32         cls = self.__class__
  33         p = psutil.Process()
  34         initial_fds = p.num_fds()
  35
  36         for _ in range(100):
  37             stats = VPPStats(socketname=cls.stats_sock)
  38             stats.disconnect()
  39
  40         ending_fds = p.num_fds()
  41         self.assertEqual(initial_fds, ending_fds,
  42                          "initial client side file descriptor count: %s "
  43                          "is not equal to "
  44                          "ending client side file descriptor count: %s" % (
  45                              initial_fds, ending_fds))
  46
  47     @unittest.skip("Manual only")
  48     def test_mem_leak(self):
  49         def loop():
  50             print('Running loop')
  51             for i in range(50):
  52                 rv = self.vapi.papi.tap_create_v2(id=i, use_random_mac=1)
  53                 self.assertEqual(rv.retval, 0)
  54                 rv = self.vapi.papi.tap_delete_v2(sw_if_index=rv.sw_if_index)
  55                 self.assertEqual(rv.retval, 0)
  56
  57         before = self.statistics.get_counter('/mem/statseg/used')
  58         loop()
  59         self.vapi.cli("memory-trace on stats-segment")
  60         for j in range(100):
  61             loop()
  62         print(self.vapi.cli("show memory stats-segment verbose"))
  63         print('AFTER', before,
  64               self.statistics.get_counter('/mem/statseg/used'))
  65
  66
  67 if __name__ == '__main__':
  68     unittest.main(testRunner=VppTestRunner)