tests/unit/benchmark/scenarios/networking/test_pktgen_dpdk.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2015 ZTE and others.
   5 #
   6 # All rights reserved. This program and the accompanying materials
   7 # are made available under the terms of the Apache License, Version 2.0
   8 # which accompanies this distribution, and is available at
   9 # http://www.apache.org/licenses/LICENSE-2.0
  10 ##############################################################################
  11
  12 # Unittest for yardstick.benchmark.scenarios.networking.pktgen.Pktgen
  13
  14 import mock
  15 import unittest
  16 import json
  17
  18 from yardstick.benchmark.scenarios.networking import pktgen_dpdk
  19
  20 @mock.patch('yardstick.benchmark.scenarios.networking.pktgen_dpdk.ssh')
  21 class PktgenDPDKLatencyTestCase(unittest.TestCase):
  22
  23     def setUp(self):
  24         self.ctx = {
  25             'host': {
  26                 'ip': '172.16.0.137',
  27                 'user': 'root',
  28                 'key_filename': 'mykey.key'
  29             },
  30             'target': {
  31                 'ip': '172.16.0.138',
  32                 'user': 'root',
  33                 'key_filename': 'mykey.key',
  34                 'ipaddr': '172.16.0.138'
  35             }
  36         }
  37
  38     def test_pktgen_dpdk_successful_setup(self, mock_ssh):
  39
  40         args = {
  41             'options': {'packetsize': 60},
  42         }
  43         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
  44         p.setup()
  45
  46         mock_ssh.SSH().execute.return_value = (0, '', '')
  47         self.assertIsNotNone(p.server)
  48         self.assertIsNotNone(p.client)
  49         self.assertEqual(p.setup_done, True)
  50
  51     def test_pktgen_dpdk_successful_get_port_ip(self, mock_ssh):
  52
  53         args = {
  54             'options': {'packetsize': 60},
  55         }
  56         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
  57         p.server = mock_ssh.SSH()
  58
  59         mock_ssh.SSH().execute.return_value = (0, '', '')
  60
  61         p.get_port_ip(p.server, "eth1")
  62
  63         mock_ssh.SSH().execute.assert_called_with(
  64             "ifconfig eth1 |grep 'inet addr' |awk '{print $2}' \
  65             |cut -d ':' -f2 ")
  66
  67     def test_pktgen_dpdk_unsuccessful_get_port_ip(self, mock_ssh):
  68
  69         args = {
  70             'options': {'packetsize': 60},
  71         }
  72
  73         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
  74         p.server = mock_ssh.SSH()
  75
  76         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
  77         self.assertRaises(RuntimeError, p.get_port_ip, p.server, "eth1")
  78
  79     def test_pktgen_dpdk_successful_get_port_mac(self, mock_ssh):
  80
  81         args = {
  82             'options': {'packetsize': 60},
  83         }
  84         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
  85         p.server = mock_ssh.SSH()
  86
  87         mock_ssh.SSH().execute.return_value = (0, '', '')
  88
  89         p.get_port_mac(p.server, "eth1")
  90
  91         mock_ssh.SSH().execute.assert_called_with(
  92             "ifconfig |grep HWaddr |grep eth1 |awk '{print $5}' ")
  93
  94     def test_pktgen_dpdk_unsuccessful_get_port_mac(self, mock_ssh):
  95
  96         args = {
  97             'options': {'packetsize': 60},
  98         }
  99
 100         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
 101         p.server = mock_ssh.SSH()
 102
 103         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 104         self.assertRaises(RuntimeError, p.get_port_mac, p.server, "eth1")
 105
 106     def test_pktgen_dpdk_successful_no_sla(self, mock_ssh):
 107
 108         args = {
 109             'options': {'packetsize': 60},
 110         }
 111
 112         result = {}
 113         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
 114
 115         sample_output = '100\n110\n112\n130\n149\n150\n90\n150\n200\n162\n'
 116         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 117
 118         p.run(result)
 119         self.assertEqual(result, {"avg_latency": 132})
 120
 121     def test_pktgen_dpdk_successful_sla(self, mock_ssh):
 122
 123         args = {
 124             'options': {'packetsize': 60},
 125             'sla': {'max_latency': 100}
 126         }
 127         result = {}
 128
 129         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
 130
 131         sample_output = '100\n100\n100\n100\n100\n100\n100\n100\n100\n100\n'
 132         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 133
 134         p.run(result)
 135
 136         self.assertEqual(result, {"avg_latency": 100})
 137
 138     def test_pktgen_dpdk_unsuccessful_sla(self, mock_ssh):
 139
 140         args = {
 141             'options': {'packetsize': 60},
 142             'sla': {'max_latency': 100}
 143         }
 144         result = {}
 145
 146         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
 147
 148         p.server = mock_ssh.SSH()
 149         p.client = mock_ssh.SSH()
 150
 151         sample_output = '100\n110\n112\n130\n149\n150\n90\n150\n200\n162\n'
 152         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 153         self.assertRaises(AssertionError, p.run, result)
 154
 155     def test_pktgen_dpdk_unsuccessful_script_error(self, mock_ssh):
 156
 157         args = {
 158             'options': {'packetsize': 60},
 159             'sla': {'max_latency': 100}
 160         }
 161         result = {}
 162
 163         p = pktgen_dpdk.PktgenDPDKLatency(args, self.ctx)
 164
 165         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 166         self.assertRaises(RuntimeError, p.run, result)
 167
 168
 169 def main():
 170     unittest.main()
 171
 172 if __name__ == '__main__':
 173     main()