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         p.server = mock_ssh.SSH()
 117         p.client = mock_ssh.SSH()
 118
 119         mock_iptables_result = mock.Mock()
 120         mock_iptables_result.return_value = 149300
 121         p._iptables_get_result = mock_iptables_result
 122
 123         sample_output = '{"packets_per_second": 9753, "errors": 0, \
 124             "packets_sent": 149776, "flows": 110}'
 125         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 126
 127         result = p.run(args)
 128         expected_result = json.loads(sample_output)
 129         expected_result["packets_received"] = 149300
 130         self.assertEqual(result, expected_result)
 131
 132     def test_pktgen_successful_sla(self, mock_ssh):
 133
 134         p = pktgen.Pktgen(self.ctx)
 135         args = {
 136             'options': {'packetsize': 60, 'number_of_ports': 10},
 137             'ipaddr': '172.16.0.139',
 138             'sla': {'max_ppm': 10000}
 139         }
 140         p.server = mock_ssh.SSH()
 141         p.client = mock_ssh.SSH()
 142
 143         mock_iptables_result = mock.Mock()
 144         mock_iptables_result.return_value = 149300
 145         p._iptables_get_result = mock_iptables_result
 146
 147         sample_output = '{"packets_per_second": 9753, "errors": 0, \
 148             "packets_sent": 149776, "flows": 110}'
 149         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 150
 151         result = p.run(args)
 152         expected_result = json.loads(sample_output)
 153         expected_result["packets_received"] = 149300
 154         self.assertEqual(result, expected_result)
 155
 156     def test_pktgen_unsuccessful_sla(self, mock_ssh):
 157
 158         p = pktgen.Pktgen(self.ctx)
 159         args = {
 160             'options': {'packetsize': 60, 'number_of_ports': 10},
 161             'ipaddr': '172.16.0.139',
 162             'sla': {'max_ppm': 1000}
 163         }
 164         p.server = mock_ssh.SSH()
 165         p.client = mock_ssh.SSH()
 166
 167         mock_iptables_result = mock.Mock()
 168         mock_iptables_result.return_value = 149300
 169         p._iptables_get_result = mock_iptables_result
 170
 171         sample_output = '{"packets_per_second": 9753, "errors": 0, \
 172             "packets_sent": 149776, "flows": 110}'
 173         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 174         self.assertRaises(AssertionError, p.run, args)
 175
 176     def test_pktgen_unsuccessful_script_error(self, mock_ssh):
 177
 178         p = pktgen.Pktgen(self.ctx)
 179         args = {
 180             'options': {'packetsize': 60, 'number_of_ports': 10},
 181             'ipaddr': '172.16.0.139',
 182             'sla': {'max_ppm': 1000}
 183         }
 184         p.server = mock_ssh.SSH()
 185         p.client = mock_ssh.SSH()
 186
 187         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 188         self.assertRaises(RuntimeError, p.run, args)
 189
 190
 191 def main():
 192     unittest.main()
 193
 194 if __name__ == '__main__':
 195     main()