133 lines
7.3 KiB
Python
133 lines
7.3 KiB
Python
"""Regression checks for configuration, alert truth and stale snapshot handling."""
|
|
import ast
|
|
import concurrent.futures
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
from urllib.error import HTTPError
|
|
from urllib.request import Request, urlopen
|
|
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0,str(ROOT/'app'))
|
|
import pigway_pi_control_api as api
|
|
import plugin_api
|
|
from storage_cache import StorageCache
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory=tempfile.TemporaryDirectory();self.addCleanup(self.directory.cleanup)
|
|
self.config=Path(self.directory.name)/'agent.conf'
|
|
self.config.write_text((ROOT/'config/pigway-pi-control.conf').read_text())
|
|
self.override=patch.object(api,'CFG',self.config);self.override.start();self.addCleanup(self.override.stop)
|
|
|
|
def test_invalid_values_and_relationships_leave_configuration_unchanged(self):
|
|
original=self.config.read_text()
|
|
for updates in [
|
|
{'timing':{'sensor_interval':'-1'}},{'timing':{'sensor_interval':'0'}},
|
|
{'alerts':{'cpu_high':'nan'}},{'alerts':{'memory_high':'inf'}},
|
|
{'alerts':{'cpu_high':'101'}},{'alerts':{'cpu_high':'50'}},
|
|
{'alerts':{'temp_high':'80'}},{'alerts':{'disk_high':'99'}},
|
|
{'alerts':{'wifi_weak':'-60'}},{'alerts':{'cpu_high_seconds':'1.5'}},
|
|
{'services':{'BAD':'not-a-unit'}},
|
|
]:
|
|
with self.subTest(updates=updates):
|
|
with self.assertRaises(ValueError):api.update_ini(updates)
|
|
self.assertEqual(self.config.read_text(),original)
|
|
api.update_ini({'alerts':{'cpu_high':'60','cpu_recover':'50'}})
|
|
self.assertEqual(api.load_cfg().getfloat('alerts','cpu_high'),60)
|
|
|
|
def test_literal_percent_in_identity_and_process_matches(self):
|
|
api.update_ini({'device':{'name':'Office 100%'},'processes':{'APP':'worker --ratio=100%'}})
|
|
self.assertEqual(api.machine_identity()['name'],'Office 100%')
|
|
self.assertEqual(api.load_cfg().get('processes','app'),'worker --ratio=100%')
|
|
|
|
def test_parallel_writes_preserve_every_edit_and_backup(self):
|
|
def save(index):return api.update_ini({'processes':{f'APP{index}':f'worker-{index}'}})
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
backups=list(pool.map(save,range(20)))
|
|
self.assertEqual(len(set(backups)),20)
|
|
cfg=api.load_cfg()
|
|
for index in range(20):self.assertEqual(cfg.get('processes',f'app{index}'),f'worker-{index}')
|
|
self.assertFalse(list(self.config.parent.glob('agent.conf.*tmp*')))
|
|
|
|
|
|
class StateTests(unittest.TestCase):
|
|
def test_weekly_root_cache_survives_restart_after_first_day(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
reads=[]
|
|
def reader(_):reads.append(1);return SimpleNamespace(total=100,used=20,free=80)
|
|
path=Path(directory)/'cache'
|
|
StorageCache(path,reader,lambda:1000).sample()
|
|
restored=StorageCache(path,reader,lambda:101000)
|
|
restored.sample();self.assertEqual(len(reads),1)
|
|
self.assertEqual(restored.next_update_at,605800)
|
|
StorageCache(path,reader,lambda:605800).sample();self.assertEqual(len(reads),2)
|
|
|
|
def test_publisher_stops_renewing_stale_data_and_recovers(self):
|
|
now=[0];calls=[]
|
|
publisher=plugin_api.Publisher({},'test',lambda *a,**k:None)
|
|
class Event:
|
|
def is_set(self):return now[0]>=8
|
|
def wait(self,_):
|
|
now[0]+=1
|
|
if now[0]==6:publisher.publish(state)
|
|
publisher.stop_event=Event()
|
|
state={'system':{},'network':{},'display':{'alerts':[]}}
|
|
def request(path,method,endpoint,payload=None):
|
|
if method=='PUT':calls.append((now[0],payload['revision']))
|
|
return {'api_version':1}
|
|
with patch.object(plugin_api.time,'monotonic',lambda:now[0]),patch.object(plugin_api,'request',request):
|
|
publisher.publish(state);publisher.run('test','unused')
|
|
self.assertEqual(calls,[(0,1),(1,2),(2,3),(6,4),(7,5)])
|
|
|
|
def test_status_api_rejects_stale_agent_then_accepts_fresh_data(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path=Path(directory)/'status.json';path.write_text(json.dumps({'timestamp':time.time()-60}))
|
|
token=Path(directory)/'token';token.write_text('test-token')
|
|
with patch.object(api,'STATUS',path),patch.object(api,'TOKEN',token):
|
|
server=ThreadingHTTPServer(('127.0.0.1',0),api.Handler)
|
|
worker=threading.Thread(target=server.serve_forever);worker.start()
|
|
try:
|
|
req=Request(f'http://127.0.0.1:{server.server_port}/api/v1/status',headers={'Authorization':'Bearer test-token'})
|
|
with self.assertRaises(HTTPError) as caught:urlopen(req)
|
|
self.assertEqual(caught.exception.code,503)
|
|
caught.exception.close()
|
|
path.write_text(json.dumps({'timestamp':time.time()}))
|
|
with urlopen(req) as response:self.assertEqual(response.status,200)
|
|
finally:server.shutdown();worker.join();server.server_close()
|
|
|
|
def test_alert_severity_owner_changes_and_configured_disk_text(self):
|
|
tree=ast.parse((ROOT/'app/pigway_pi_control.py').read_text())
|
|
names={'AlertManager','alert_priority','update_active_alerts'}
|
|
nodes=[n for n in tree.body if isinstance(n,(ast.ClassDef,ast.FunctionDef)) and n.name in names]
|
|
logs=[]
|
|
env={'time':time,'log':lambda *args,**kwargs:logs.append((args,kwargs)),
|
|
'ALERT_PRIORITIES':{'CPU_HIGH':45,'DISK_CRITICAL':55,'DISK_HIGH':40,'TEMP_HIGH':50,'TEMP_CRITICAL':90},
|
|
'ALERT_CATEGORIES':{'CPU_HIGH':'CPU','DISK_CRITICAL':'STORAGE','DISK_HIGH':'STORAGE','TEMP_HIGH':'CPU','TEMP_CRITICAL':'CPU'}}
|
|
exec(compile(ast.Module(body=nodes,type_ignores=[]),'monitor-functions','exec'),env)
|
|
manager=env['AlertManager']()
|
|
manager.request('CPU_HIGH',1,'CPU');manager.request('DISK_CRITICAL',2,'DISK')
|
|
self.assertEqual(logs[-1][0][0],'CRITICAL')
|
|
manager.request('TEMP_HIGH',3,'TEMP')
|
|
self.assertEqual(manager.status_snapshot()['owner'],'TEMP_HIGH')
|
|
self.assertEqual(logs[-1][0][0],'EMERGENCY')
|
|
before=len(logs);manager.request('TEMP_HIGH',3,'TEMP');self.assertEqual(len(logs),before)
|
|
manager.request('TEMP_HIGH',1,'TEMP');self.assertEqual(logs[-1][0][1],'ALERT_CHANGED')
|
|
env.update(alert_manager=manager,cpu_hi=0,mem_hi=0,CPU_HIGH=90,CPU_RECOVER=75,CPU_HIGH_SEC=30,
|
|
MEM_HIGH=90,MEM_RECOVER=85,MEM_HIGH_SEC=30,TEMP_HIGH=68,TEMP_CRIT=75,
|
|
DISK_HIGH=70,DISK_CRIT=80,link_ok=True,net_ok=True,watches=[],watch_state=[],
|
|
net_kind='NET',health={},fs_readonly=False,ethernet_slow_active=False,
|
|
transition=lambda *a,**k:None,net_iface='',net_speed_mbps=None,ETHERNET_MIN_SPEED=1000)
|
|
env['update_active_alerts'](10,40,20,85)
|
|
self.assertEqual(manager.requests['DISK_CRITICAL']['l3'],'USED >= 80%')
|
|
env['update_active_alerts'](10,40,20,75)
|
|
self.assertEqual(manager.requests['DISK_HIGH']['l3'],'USED >= 70%')
|