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