vstf/vstf/agent/unittest/perf/test_netperf.py

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