def setUp(self):
self.ctx = {
- 'host': '172.16.0.137',
- 'target': '172.16.0.138',
- 'user': 'cirros',
- 'key_filename': "mykey.key"
+ 'host': {
+ 'ip': '172.16.0.137',
+ 'user': 'root',
+ 'key_filename': 'mykey.key'
+ },
+ 'target': {
+ 'ip': '172.16.0.138',
+ 'user': 'root',
+ 'key_filename': 'mykey.key',
+ 'ipaddr': '172.16.0.138'
+ }
}
def test_pktgen_successful_setup(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60},
- 'ipaddr': '172.16.0.139'
}
+ p = pktgen.Pktgen(args, self.ctx)
p.setup()
mock_ssh.SSH().execute.return_value = (0, '', '')
def test_pktgen_successful_iptables_setup(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139'
}
+ p = pktgen.Pktgen(args, self.ctx)
p.server = mock_ssh.SSH()
p.number_of_ports = args['options']['number_of_ports']
def test_pktgen_unsuccessful_iptables_setup(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139'
}
+
+ p = pktgen.Pktgen(args, self.ctx)
p.server = mock_ssh.SSH()
p.number_of_ports = args['options']['number_of_ports']
def test_pktgen_successful_iptables_get_result(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139'
}
+
+ p = pktgen.Pktgen(args, self.ctx)
p.server = mock_ssh.SSH()
p.number_of_ports = args['options']['number_of_ports']
def test_pktgen_unsuccessful_iptables_get_result(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139'
}
+
+ p = pktgen.Pktgen(args, self.ctx)
+
p.server = mock_ssh.SSH()
p.number_of_ports = args['options']['number_of_ports']
def test_pktgen_successful_no_sla(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139'
}
+ result = {}
+
+ p = pktgen.Pktgen(args, self.ctx)
+
p.server = mock_ssh.SSH()
p.client = mock_ssh.SSH()
"packets_sent": 149776, "flows": 110}'
mock_ssh.SSH().execute.return_value = (0, sample_output, '')
- result = p.run(args)
+ p.run(result)
expected_result = json.loads(sample_output)
expected_result["packets_received"] = 149300
self.assertEqual(result, expected_result)
def test_pktgen_successful_sla(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139',
'sla': {'max_ppm': 10000}
}
+ result = {}
+
+ p = pktgen.Pktgen(args, self.ctx)
+
p.server = mock_ssh.SSH()
p.client = mock_ssh.SSH()
"packets_sent": 149776, "flows": 110}'
mock_ssh.SSH().execute.return_value = (0, sample_output, '')
- result = p.run(args)
+ p.run(result)
expected_result = json.loads(sample_output)
expected_result["packets_received"] = 149300
self.assertEqual(result, expected_result)
def test_pktgen_unsuccessful_sla(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139',
'sla': {'max_ppm': 1000}
}
+ result = {}
+
+ p = pktgen.Pktgen(args, self.ctx)
+
p.server = mock_ssh.SSH()
p.client = mock_ssh.SSH()
sample_output = '{"packets_per_second": 9753, "errors": 0, \
"packets_sent": 149776, "flows": 110}'
mock_ssh.SSH().execute.return_value = (0, sample_output, '')
- self.assertRaises(AssertionError, p.run, args)
+ self.assertRaises(AssertionError, p.run, result)
def test_pktgen_unsuccessful_script_error(self, mock_ssh):
- p = pktgen.Pktgen(self.ctx)
args = {
'options': {'packetsize': 60, 'number_of_ports': 10},
- 'ipaddr': '172.16.0.139',
'sla': {'max_ppm': 1000}
}
+ result = {}
+
+ p = pktgen.Pktgen(args, self.ctx)
+
p.server = mock_ssh.SSH()
p.client = mock_ssh.SSH()
mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
- self.assertRaises(RuntimeError, p.run, args)
+ self.assertRaises(RuntimeError, p.run, result)
def main():