Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_init(self):
zr = ZabbixResponse()
self.assertEqual(zr.failed, 0)
self.assertEqual(zr.processed, 0)
self.assertEqual(zr.time, 0)
self.assertEqual(zr.total, 0)
self.assertEqual(zr.chunk, 0)
def test_send(self, mock_socket):
mock_data = b'\x01\\\x00\x00\x00\x00\x00\x00\x00'
mock_socket.return_value = mock_socket
mock_socket.recv.side_effect = (b'ZBXD', mock_data, self.resp_body)
zm = ZabbixMetric('host1', 'key1', 100500, 1457358608)
zs = ZabbixSender()
result = zs.send([zm])
self.assertIsInstance(result, ZabbixResponse)
self.assertEqual(result.chunk, 1)
self.assertEqual(result.total, 10)
self.assertEqual(result.failed, 10)
def test_FunctionalSenderOld(self):
cur_date_unix = int(now())
m = [
ZabbixMetric('host2', 'key3', 'IDDQD'),
ZabbixMetric('host1', 'key1', 33.1, cur_date_unix)
]
z = ZabbixSender('127.0.0.1', 10051).send(m)
self.assertIsInstance(z, ZabbixResponse)
self.assertEqual(z.total, 2)
self.assertEqual(z.processed, 2)
self.assertEqual(z.failed, 0)
self.assertEqual(z.chunk, 1)
def test_repr(self):
zr = ZabbixResponse()
result = json.dumps({'processed': zr._processed,
'failed': zr._failed,
'total': zr._total,
'time': str(zr._time),
'chunk': zr._chunk})
self.assertEqual(zr.__repr__(), result)