38 lines
2.1 KiB
Python
38 lines
2.1 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': {'observed_at':120,'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.assertEqual(next(m['observed_at'] for m in result['metrics'] if m['key']=='network_tx_bytes_per_second'),120)
|
|
self.assertEqual(result['collection']['interval_seconds'],1)
|
|
self.assertTrue(result['alerts_complete'])
|
|
|
|
def test_collection_minimum_and_slower_device(self):
|
|
self.assertEqual(worker.collection_interval({'collection':{'interval_seconds':1}}),1)
|
|
self.assertEqual(worker.collection_interval({'collection':{'interval_seconds':5}}),5)
|
|
self.assertEqual(worker.collection_interval({'collection':{'interval_seconds':True}}),1)
|
|
|
|
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()
|