'"%(class_name)" class.')
+class InvalidType(YardstickException):
+ message = 'Type "%(type_to_convert)s" is not valid'
+
+
class InfluxDBConfigurationMissing(YardstickException):
message = ('InfluxDB configuration is not available. Add "influxdb" as '
'a dispatcher and the configuration section')
import ipaddress
import logging
import os
+import pydoc
import random
import re
import signal
finally:
sock.close()
return ret
+
+
+def safe_cast(value, type_to_convert, default_value):
+ """Convert value to type, in case of error return default_value
+
+ :param value: value to convert
+ :param type_to_convert: type to convert, could be "type" or "string"
+ :param default_value: default value to return
+ :return: converted value or default_value
+ """
+ if isinstance(type_to_convert, type):
+ _type = type_to_convert
+ else:
+ _type = pydoc.locate(type_to_convert)
+ if not _type:
+ raise exceptions.InvalidType(type_to_convert=type_to_convert)
+
+ try:
+ return _type(value)
+ except ValueError:
+ return default_value
'out_packets': int(stats['Frames_Tx'][port_num]),
'RxThroughput': float(stats['Valid_Frames_Rx'][port_num]) / duration,
'TxThroughput': float(stats['Frames_Tx'][port_num]) / duration,
- 'Store-Forward_Avg_latency_ns': avg_latency,
- 'Store-Forward_Min_latency_ns': min_latency,
- 'Store-Forward_Max_latency_ns': max_latency
+ 'Store-Forward_Avg_latency_ns': utils.safe_cast(avg_latency, int, 0),
+ 'Store-Forward_Min_latency_ns': utils.safe_cast(min_latency, int, 0),
+ 'Store-Forward_Max_latency_ns': utils.safe_cast(max_latency, int, 0)
}
except IndexError:
pass
def test_return_value(self):
self.assertEqual('foo', utils.get_port_ip(self.ssh_client, 99))
+
+
+class SafeCaseTestCase(unittest.TestCase):
+
+ def test_correct_type_int(self):
+ self.assertEqual(35, utils.safe_cast('35', int, 0))
+
+ def test_correct_int_as_string(self):
+ self.assertEqual(25, utils.safe_cast('25', 'int', 0))
+
+ def test_incorrect_type_as_string(self):
+ with self.assertRaises(exceptions.InvalidType):
+ utils.safe_cast('100', 'intt', 0)
+
+ def test_default_value(self):
+ self.assertEqual(0, utils.safe_cast('', 'int', 0))