31 lines
1.6 KiB
Python
31 lines
1.6 KiB
Python
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'plugin'))
|
|
import worker
|
|
|
|
|
|
class MetricsTests(unittest.TestCase):
|
|
def test_existing_agent_fields_become_standard_history_metrics(self):
|
|
value = {'timestamp': 123, 'machine': {'name':'Pi'}, 'system': {
|
|
'cpu_percent': 10, 'memory_total_bytes': 8000, 'memory_used_bytes': 4000,
|
|
'disk_total_bytes': 16000, 'disk_used_bytes': 5000},
|
|
'uptime_seconds': 1000, 'network': {'link_speed_mbps':1000,'rx_bytes_per_second':0,'tx_bytes_per_second':1234},
|
|
'display': {'alerts': []}}
|
|
with patch.object(worker,'request',return_value=value): result=worker.collect({})
|
|
values = {m['key']:m['value'] for m in result['metrics']}
|
|
self.assertEqual(values['memory_used_bytes'],4000)
|
|
self.assertEqual(values['uptime_seconds'],1000)
|
|
self.assertEqual(values['network_link_speed_mbps'],1000)
|
|
self.assertEqual(values['network_rx_bytes_per_second'],0)
|
|
self.assertEqual(values['network_tx_bytes_per_second'],1234)
|
|
self.assertTrue(result['alerts_complete'])
|
|
|
|
def test_absent_or_boolean_fields_not_invented(self):
|
|
value={'system': {'cpu_percent': True}, 'machine':{}, 'network': {'link_speed_mbps': None, 'rx_bytes_per_second':float('nan'), 'tx_bytes_per_second':True}}
|
|
with patch.object(worker,'request',return_value=value): result=worker.collect({})
|
|
self.assertEqual(result['metrics'],[])
|
|
|
|
if __name__=='__main__': unittest.main()
|