pigway-console-adapter-raspberry-pi/tests/test_metrics.py

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()