"""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%')