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