vstf/vstf/agent/unittest/perf/test_qperf.py

   1 '''
   2 Created on 2015-9-24
   3
   4 @author: y00228926
   5 '''
   6 import unittest
   7 import subprocess
   8 import time
   9
  10 from vstf.agent.unittest.perf import model
  11 from vstf.agent.perf import qperf
  12 from vstf.agent.perf.utils import get_pid_by_name
  13
  14
  15 class testQperf(model.Model):
  16     def setUp(self):
  17         super(testQperf, self).setUp()
  18         subprocess.call("killall qperf", shell = True)
  19         for ns, dev, ip_setting in zip(self.ns_list, self.device_list, self.ip_setting_list):
  20             netdev = {
  21                   "namespace":ns,
  22                   "iface":dev,
  23                   'ip_setting':ip_setting
  24             }
  25             self.mgr.config_dev(netdev)
  26         self.send_cfg = {
  27             "namespace": self.ns_list[0],
  28             "time":1,
  29             "protocol": "udp_lat",
  30             "dst":[
  31                     {"ip": self.ip_list[1]}
  32                 ],
  33             "size": 64,
  34         }
  35
  36     def tearDown(self):
  37         super(testQperf, self).tearDown()
  38
  39     def test_qperf_quick(self):
  40         perf = qperf.Qperf()
  41         ret = perf.receive_start(namespace=self.ns_list[1])
  42         exp = (0, "start qperf receive success")
  43         self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
  44
  45         ret = perf.send_start(**self.send_cfg)
  46         exp = (0,"start qperf send success")
  47         self.assertEqual(ret, exp, "send_start failed")
  48
  49         time.sleep(3)
  50
  51         ret = perf.send_stop()
  52         for r in ret:
  53             self.assertEqual(r[0], 0, "send_stop failed, ret = %s" % str(ret))
  54             for key in ('MaximumLatency', 'AverageLatency', 'MinimumLatency'):
  55                 self.assertIn(key, r[1], "send_stop failed, ret = %s" % str(ret))
  56
  57         ret = perf.receive_stop()
  58         exp = (0, "stop qperf receive success")
  59         self.assertEqual(ret, exp, "receive_stop failed, ret = %s" % str(ret))
  60
  61     def test_qperf_quick_3s(self):
  62         perf = qperf.Qperf()
  63         self.send_cfg.update({"time":3})
  64         ret = perf.receive_start(namespace='receive')
  65         exp = (0, 'start qperf receive success')
  66         self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
  67
  68         self.send_cfg.update({"threads":3})
  69         exp = (0,"start qperf send success")
  70         ret = perf.send_start(**self.send_cfg)
  71         self.assertEqual(ret, exp, "send_start failed %s" % str(ret))
  72
  73         ret = perf.send_stop()
  74         for r in ret:
  75             self.assertEqual(r[0], 0, "send_stop failed, ret = %s" % str(ret))
  76             for key in ('MaximumLatency', 'AverageLatency', 'MinimumLatency'):
  77                 self.assertIn(key, r[1], "send_stop failed, ret = %s" % str(ret))
  78
  79         ret = perf.receive_stop()
  80         self.assertEqual(ret, (0, 'stop qperf receive success'), "receive_stop failedf, ret = %s" % str(ret))
  81
  82     def test_clean(self):
  83         perf = qperf.Qperf()
  84         self.send_cfg.update({"time":10})
  85         ret = perf.receive_start(namespace=self.ns_list[1])
  86         exp = (0, "start qperf receive success")
  87         self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
  88
  89         ret = perf.send_start(**self.send_cfg)
  90         exp = (0,"start qperf send success")
  91         self.assertEqual(ret, exp, "send_start failed")
  92
  93         perf.clean()
  94         ret = get_pid_by_name('qperf')
  95         self.assertEqual(ret, [], "clean qperf failed")
  96
  97
  98 if __name__ == "__main__":
  99     import logging
 100     logging.getLogger(__name__)
 101     logging.basicConfig(level = logging.DEBUG)
 102     unittest.main()