Upload the contribution of vstf as bottleneck network framework.

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

diff --git a/vstf/vstf/agent/unittest/perf/test_pktgen.py b/vstf/vstf/agent/unittest/perf/test_pktgen.py
new file mode 100755 (executable)
index 0000000..5e6cdf7
--- /dev/null
+++ b/vstf/vstf/agent/unittest/perf/test_pktgen.py
@@ -0,0 +1,90 @@
+"""
+Created on 2015-9-24
+
+@author: y00228926
+"""
+import unittest
+import time
+
+from vstf.agent.unittest.perf import model
+from vstf.agent.perf import pktgen
+
+
+class TestPktgen(model.Model):
+    def setUp(self):
+        super(TestPktgen, self).setUp()
+
+    def tearDown(self):
+        super(TestPktgen, self).tearDown()
+
+    def test_single_thread(self):
+        perf = pktgen.Pktgen()
+        print perf.receive_start()
+        send = {
+            "src": [
+                {"iface": self.device_list[0], "mac": self.mac_list[0]}
+            ],
+            "dst": [
+                {"mac": self.mac_list[1]}
+            ],
+            "size": 64,
+            "threads": 1,
+            'ratep': 0
+        }
+        ret = perf.send_start(**send)
+        self.assertEqual((0, 'start pktgen send success'), ret, "send_start failed, ret=%s" % str(ret))
+        time.sleep(5)
+        ret = perf.send_stop()
+        self.assertEqual([(0, '')], ret, "send_start failed, ret=%s" % ret)
+        ret = perf.receive_stop()
+        self.assertEqual((0, 'pktgen neednt receive stop'), ret, "send_stop failed, ret=%s" % str(ret))
+
+    def test_single_thread_bidirectional(self):
+        perf = pktgen.Pktgen()
+        print perf.receive_start()
+        send = {
+            "src": [
+                {"iface": self.device_list[0], "mac": self.mac_list[0]},
+                {"iface": self.device_list[1], "mac": self.mac_list[1]}
+            ],
+            "dst": [
+                {"mac": self.mac_list[1]},
+                {"mac": self.mac_list[0]}
+            ],
+            "size": 64,
+            "threads": 1,
+            'ratep': 0
+        }
+        ret = perf.send_start(**send)
+        self.assertEqual((0, 'start pktgen send success'), ret, "send_start failed, ret=%s" % str(ret))
+        time.sleep(5)
+        ret = perf.send_stop()
+        self.assertEqual([(0, '')], ret, "send_start failed, ret=%s" % ret)
+        ret = perf.receive_stop()
+        self.assertEqual((0, 'pktgen neednt receive stop'), ret, "send_stop failed, ret=%s" % str(ret))
+
+    def test_clean(self):
+        perf = pktgen.Pktgen()
+        print perf.receive_start()
+        send = {
+            "src": [
+                {"iface": self.device_list[0], "mac": self.mac_list[0]}
+            ],
+            "dst": [
+                {"mac": self.mac_list[1]}
+            ],
+            "size": 64,
+            "threads": 1,
+            'ratep': 0
+        }
+        ret = perf.send_start(**send)
+        self.assertEqual((0, 'start pktgen send success'), ret, "send_start failed, ret=%s" % str(ret))
+        perf.clean()
+
+
+if __name__ == "__main__":
+    import logging
+
+    logging.getLogger(__name__)
+    logging.basicConfig(level=logging.DEBUG)
+    unittest.main()