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

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2015 Ericsson AB 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
  19
  20
  21 @mock.patch('yardstick.benchmark.scenarios.networking.pktgen.ssh')
  22 class PktgenTestCase(unittest.TestCase):
  23
  24     def setUp(self):
  25         self.ctx = {
  26             'host': '172.16.0.137',
  27             'target': '172.16.0.138',
  28             'user': 'cirros',
  29             'key_filename': "mykey.key"
  30         }
  31
  32     def test_pktgen_successful_setup(self, mock_ssh):
  33
  34         p = pktgen.Pktgen(self.ctx)
  35         args = {
  36             'options': {'packetsize': 60},
  37             'ipaddr': '172.16.0.139'
  38         }
  39         p.setup()
  40
  41         mock_ssh.SSH().execute.return_value = (0, '', '')
  42         self.assertIsNotNone(p.server)
  43         self.assertIsNotNone(p.client)
  44         self.assertEqual(p.setup_done, True)
  45
  46     def test_pktgen_successful_iptables_setup(self, mock_ssh):
  47
  48         p = pktgen.Pktgen(self.ctx)
  49         args = {
  50             'options': {'packetsize': 60, 'number_of_ports': 10},
  51             'ipaddr': '172.16.0.139'
  52         }
  53         p.server = mock_ssh.SSH()
  54         p.number_of_ports = args['options']['number_of_ports']
  55
  56         mock_ssh.SSH().execute.return_value = (0, '', '')
  57
  58         p._iptables_setup()
  59
  60         mock_ssh.SSH().execute.assert_called_with(
  61             "sudo iptables -F; "
  62             "sudo iptables -A INPUT -p udp --dport 1000:%s -j DROP"
  63             % 1010)
  64
  65     def test_pktgen_unsuccessful_iptables_setup(self, mock_ssh):
  66
  67         p = pktgen.Pktgen(self.ctx)
  68         args = {
  69             'options': {'packetsize': 60, 'number_of_ports': 10},
  70             'ipaddr': '172.16.0.139'
  71         }
  72         p.server = mock_ssh.SSH()
  73         p.number_of_ports = args['options']['number_of_ports']
  74
  75         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
  76         self.assertRaises(RuntimeError, p._iptables_setup)
  77
  78     def test_pktgen_successful_iptables_get_result(self, mock_ssh):
  79
  80         p = pktgen.Pktgen(self.ctx)
  81         args = {
  82             'options': {'packetsize': 60, 'number_of_ports': 10},
  83             'ipaddr': '172.16.0.139'
  84         }
  85         p.server = mock_ssh.SSH()
  86         p.number_of_ports = args['options']['number_of_ports']
  87
  88         mock_ssh.SSH().execute.return_value = (0, '150000', '')
  89         p._iptables_get_result()
  90
  91         mock_ssh.SSH().execute.assert_called_with(
  92             "sudo iptables -L INPUT -vnx |"
  93             "awk '/dpts:1000:%s/ {{printf \"%%s\", $1}}'"
  94             % 1010)
  95
  96     def test_pktgen_unsuccessful_iptables_get_result(self, mock_ssh):
  97
  98         p = pktgen.Pktgen(self.ctx)
  99         args = {
 100             'options': {'packetsize': 60, 'number_of_ports': 10},
 101             'ipaddr': '172.16.0.139'
 102         }
 103         p.server = mock_ssh.SSH()
 104         p.number_of_ports = args['options']['number_of_ports']
 105
 106         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 107         self.assertRaises(RuntimeError, p._iptables_get_result)
 108
 109     def test_pktgen_successful_no_sla(self, mock_ssh):
 110
 111         p = pktgen.Pktgen(self.ctx)
 112         args = {
 113             'options': {'packetsize': 60, 'number_of_ports': 10},
 114             'ipaddr': '172.16.0.139'
 115         }
 116         result = {}
 117
 118         p.server = mock_ssh.SSH()
 119         p.client = mock_ssh.SSH()
 120
 121         mock_iptables_result = mock.Mock()
 122         mock_iptables_result.return_value = 149300
 123         p._iptables_get_result = mock_iptables_result
 124
 125         sample_output = '{"packets_per_second": 9753, "errors": 0, \
 126             "packets_sent": 149776, "flows": 110}'
 127         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 128
 129         p.run(args, result)
 130         expected_result = json.loads(sample_output)
 131         expected_result["packets_received"] = 149300
 132         self.assertEqual(result, expected_result)
 133
 134     def test_pktgen_successful_sla(self, mock_ssh):
 135
 136         p = pktgen.Pktgen(self.ctx)
 137         args = {
 138             'options': {'packetsize': 60, 'number_of_ports': 10},
 139             'ipaddr': '172.16.0.139',
 140             'sla': {'max_ppm': 10000}
 141         }
 142         result = {}
 143         p.server = mock_ssh.SSH()
 144         p.client = mock_ssh.SSH()
 145
 146         mock_iptables_result = mock.Mock()
 147         mock_iptables_result.return_value = 149300
 148         p._iptables_get_result = mock_iptables_result
 149
 150         sample_output = '{"packets_per_second": 9753, "errors": 0, \
 151             "packets_sent": 149776, "flows": 110}'
 152         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 153
 154         p.run(args, result)
 155         expected_result = json.loads(sample_output)
 156         expected_result["packets_received"] = 149300
 157         self.assertEqual(result, expected_result)
 158
 159     def test_pktgen_unsuccessful_sla(self, mock_ssh):
 160
 161         p = pktgen.Pktgen(self.ctx)
 162         args = {
 163             'options': {'packetsize': 60, 'number_of_ports': 10},
 164             'ipaddr': '172.16.0.139',
 165             'sla': {'max_ppm': 1000}
 166         }
 167         result = {}
 168
 169         p.server = mock_ssh.SSH()
 170         p.client = mock_ssh.SSH()
 171
 172         mock_iptables_result = mock.Mock()
 173         mock_iptables_result.return_value = 149300
 174         p._iptables_get_result = mock_iptables_result
 175
 176         sample_output = '{"packets_per_second": 9753, "errors": 0, \
 177             "packets_sent": 149776, "flows": 110}'
 178         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 179         self.assertRaises(AssertionError, p.run, args, result)
 180
 181     def test_pktgen_unsuccessful_script_error(self, mock_ssh):
 182
 183         p = pktgen.Pktgen(self.ctx)
 184         args = {
 185             'options': {'packetsize': 60, 'number_of_ports': 10},
 186             'ipaddr': '172.16.0.139',
 187             'sla': {'max_ppm': 1000}
 188         }
 189         result = {}
 190
 191         p.server = mock_ssh.SSH()
 192         p.client = mock_ssh.SSH()
 193
 194         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 195         self.assertRaises(RuntimeError, p.run, args, result)
 196
 197
 198 def main():
 199     unittest.main()
 200
 201 if __name__ == '__main__':
 202     main()