Upload the contribution of vstf as bottleneck network framework.

[bottlenecks.git] / vstf / vstf / agent / unittest / perf / test_netperf.py

diff --git a/vstf/vstf/agent/unittest/perf/test_netperf.py b/vstf/vstf/agent/unittest/perf/test_netperf.py
new file mode 100755 (executable)
index 0000000..b5f8cb2
--- /dev/null
+++ b/vstf/vstf/agent/unittest/perf/test_netperf.py
@@ -0,0 +1,105 @@
+'''
+Created on 2015-9-24
+
+@author: y00228926
+'''
+import unittest
+import time
+import subprocess
+
+from vstf.agent.unittest.perf import model
+from vstf.agent.perf import netperf
+from vstf.agent.perf.utils import get_pid_by_name
+
+
+class TestNetperf(model.Model):
+    '''
+    please make sure 'Tn' network on 'Target Host' is created.
+    '''
+    def setUp(self):
+        super(TestNetperf, self).setUp()
+        subprocess.call("killall netperf", shell = True)
+        subprocess.call("killall netserver",shell = True)
+        for ns, dev, ip_setting in zip(self.ns_list, self.device_list, self.ip_setting_list):
+            netdev = {
+                  "namespace":ns,
+                  "iface":dev,
+                  'ip_setting':ip_setting
+            }
+            self.mgr.config_dev(netdev)
+        self.send_cfg = {
+            "namespace": "send",
+            "protocol": "udp_bw",
+            "dst":[
+                    {"ip": "192.168.1.2"}
+                ],
+            "size": 64,
+            "threads": 1,
+            "time": 10, 
+        }
+  
+    def tearDown(self):
+        super(TestNetperf, self).tearDown()
+    
+    def test_netperf_start_success(self):
+        perf = netperf.Netperf()
+        ret = perf.receive_start(namespace='receive')
+        exp = (0, 'start netserver success')
+        self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
+        
+        ret = perf.send_start(**self.send_cfg)
+        exp = (0,"start netperf send success")
+        self.assertEqual(ret, exp, "failed to start netperf")
+        
+        time.sleep(3)
+        
+        ret = perf.send_stop()
+        exp = [(0, "process is stopped by killed")]
+        self.assertEqual(ret, exp, "send_stop failed, ret = %s" % str(ret)) 
+          
+        ret = perf.receive_stop()
+        exp = (0, "stop netserver success")
+        self.assertEqual(ret, exp, "receive_stop failedf, ret = %s" % str(ret)) 
+    
+    def test_netperf_start_success_mutil_threads(self):
+        perf = netperf.Netperf()
+        ret = perf.receive_start(namespace='receive')
+        exp = (0, 'start netserver success')
+        self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
+        
+        self.send_cfg.update({"threads":3})
+        exp = (0,"start netperf send success")
+        ret = perf.send_start(**self.send_cfg)
+        self.assertEqual(ret, exp, "failed to start netperf")
+        
+        time.sleep(3)
+        
+        rets = perf.send_stop()
+        exp = [(0, 'process is stopped by killed'), (0, 'process is stopped by killed'), (0, 'process is stopped by killed')]
+        self.assertEqual(rets, exp, "send_stop failed, rets = %s" % str(rets)) 
+          
+        rets = perf.receive_stop()
+        self.assertEqual(rets, (0, "stop netserver success"), "receive_stop failedf, rets = %s" % str(rets))
+    
+    def test_clean(self):
+        perf = netperf.Netperf()
+        ret = perf.receive_start(namespace='receive')
+        exp = (0, 'start netserver success')
+        self.assertEqual(ret, exp, "receive_start failed %s" % str(ret))
+        
+        self.send_cfg.update({"threads":3})
+        exp = (0,"start netperf send success")
+        ret = perf.send_start(**self.send_cfg)
+        self.assertEqual(ret, exp, "failed to start netperf")
+        perf.clean()
+        ret = get_pid_by_name('netperf')
+        self.assertEqual(ret, [], "failed to clean netperf")
+        ret = get_pid_by_name('netserver')
+        self.assertEqual(ret, [], "failed to clean netserver")
+            
+if __name__ == "__main__":
+    import logging
+    logging.getLogger(__name__)
+    logging.basicConfig(level = logging.DEBUG)
+    #import sys;sys.argv = ['', 'Test.testName']
+    unittest.main()
\ No newline at end of file