From 9b83937d6035c9019876e9795cb6702768963343 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B8=A1=E5=8F=A3=E6=B5=AA=E4=BA=BA?= Date: Sun, 18 Oct 2026 18:15:22 +0800 Subject: [PATCH] fix: validate atomic configuration updates and expire stale monitoring snapshots --- README.md | 2 +- README.zh-CN.md | 2 +- RELEASE_NOTES.md | 12 +-- SHA256SUMS | 32 ++++---- app/pigway_pi_control.py | 13 ++-- app/pigway_pi_control_api.py | 119 +++++++++++++++------------- app/plugin_api.py | 13 +++- app/storage_cache.py | 2 +- install.sh | 2 +- tests/test_audit_regressions.py | 132 ++++++++++++++++++++++++++++++++ 10 files changed, 244 insertions(+), 85 deletions(-) create mode 100644 tests/test_audit_regressions.py diff --git a/README.md b/README.md index fdde8e1..aa1236b 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# PIGWay Raspberry Pi Agent v3.11.2 +# PIGWay Raspberry Pi Agent v3.11.3 [简体中文](README.zh-CN.md) | English diff --git a/README.zh-CN.md b/README.zh-CN.md index f1a68ca..86a4c1d 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -1,4 +1,4 @@ -# PIGWay Raspberry Pi Agent v3.11.2 +# PIGWay Raspberry Pi Agent v3.11.3 简体中文 | [English](README.md) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index edc399f..e7bef35 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,7 +1,9 @@ -# Raspberry Pi Agent v3.11.2 +# Raspberry Pi Agent v3.11.3 -新增磁盘只由用户手动扫描登记，已知磁盘容量及根分区兼容指标每周更新；服务重启复用运行缓存。保留现有CPU、内存、网络、硬件插件及监控协议。 +- 监控快照超过三秒停止续期硬件插件租约，状态 API 对超过十秒的状态返回 503，避免采集停滞仍显示为新状态。 +- 配置 API 校验有限数值、范围、阈值关系和服务目标；保留名称及进程表达式中的字面百分号。 +- 配置写入使用进程内事务锁、唯一备份及原子替换，避免并发覆盖。 +- 告警日志按严重程度记录，严重程度变化单独记事件；主要告警优先按严重程度排序，磁盘提示使用实际阈值。 +- 根分区缓存恢复期限与七天采集周期一致。 -磁盘清单持久化，支持名称/显示设置；整机重启后容量缓存首次初始化不自动发现新盘。没有调整风扇、RGB或OLED控制协议。 - -验证：10项后端测试通过；pigway-gateway生产升级后原两个服务正常，容量周期为604800秒，配置和令牌校验和保留。未执行整机重启或七天自然周期测试。 +本地 17 项回归测试通过，覆盖并发配置、旧状态拒绝与恢复、快照停滞、告警和缓存。本记录不作为当前生产状态的证明。 diff --git a/SHA256SUMS b/SHA256SUMS index 6ec3c29..1ca59dc 100644 --- a/SHA256SUMS +++ b/SHA256SUMS @@ -1,20 +1,24 @@ -ecf1884bb89b86ed4573d4d922a971abde43772bf7b2ffa6a999a12fbfb72caf .gitignore -a16429e43b2876ac4587479c3c3b5c7e3b5c666d7e889c2178a34a4ae9de8deb README.md -5a81a035871415f88fee6f12e20a42c8ff606f90e13fd383b8dd9407fe8aa909 README.zh-CN.md +516ed5bf52a34f3a842dcc050e015106f33cd856987cff1e8005ddeee73452ea .gitignore +9c90aa203d27dea1ff6493ff401ce252daa50b340672aa02b84e6729387ffe78 README.md +cf797200392a367fbe2f4fc79deabcd430249299947bd2d1d8d8cace49bfb97b README.zh-CN.md +dd98abfbdfc25c58783084b286d8324a49e5fb0410935f25be04ae9f75f7a2fc RELEASE_NOTES.md c5853f283c0951b43f2bacbc49e66c26d3a605a184d427828e042eb9236d2740 app/network_rates.py -69877b4ad2b8307db69495d629988ff4bca838dea9e36d839165029fdb6ed4a8 app/pigway_pi_control.py -757eb6a83ca07740a2389af9b51c6fd5d940b4eb86abbc603f231409416d45db app/pigway_pi_control_api.py -3a8a368bbf299eb2d6cccc968a4aacf070058bddebdc5176083a97ac579d720d app/plugin_api.py -81ca44b8bdc384c7f27a39da8e785af2450241c65200edee61db95f48d67f058 app/storage_cache.py -d5a783b64a10fa43191a813ef73ef8f762e359b0bd42ec155724fa46216298b0 config/pigway-pi-control.conf +5f553c03839a2acb86265b24db656b358b4cde6b3c617eef97c9316868578e1b app/pigway_pi_control.py +c3278a0c09f7529b50460755eed5470893d6598002ac0e0d47bf6247b100aa77 app/pigway_pi_control_api.py +9e011761ddb9583c26505fd85d6b484c440b649cb6169922512020f95a5a7d42 app/plugin_api.py +6429ba4e65baae379e892b8c2f15a5a158924b72060a692f0a3b447f8bebfc80 app/storage_cache.py +14e0e3337dde66309418658d14d1c61f1742c0f327b76a57126c72f006448165 config/pigway-pi-control.conf e3bf3e7070ea62c0526791bed922688c3fb49fee36ca9c375d080119968feb72 docs/HARDWARE_PLUGIN_API.md b54ec1f3e9fb15473910686c54142a85d387e44ea5df427a4cf1b06052e789b3 docs/RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.md 9222a2b6a38f6b6d7d88b30c05413c68c05f69ee499c8479eca05ee7c62e87c3 docs/RASPBERRY_PI_OS_TRIXIE_FRESH_INSTALL.zh-CN.md -1b0aeaf4997bdb02b967d538cf276ad7462a3e22ec35fc0d046d779ef56cd1f8 install.sh -8a593c290da5bd0d037aee86c35264512afd574a2a0f60f2c29e9db63e0109bd systemd/pigway-pi-control-api.service -1cb55de6b104a233066495e9786319305fa4f1b6f25483c108ab4f5ac9aa5700 systemd/pigway-pi-control.service -8b97229abfb31c68464eb5754bc3a8fec7a6a78e216d74a86b7cc89486700740 tests/test_network_rates.py -f78e31e0eeee4e02532885ace5c8f6f7011d084e4c7440b704408bc88495e40e tests/test_storage_cache.py -9f59e7313ee58e687ae6dd29d175c67082674254ac5c5f54168b2735b64c20a4 uninstall.sh 67f56bdb128d64f434ca52e45245fd2f3548da3852b3e06cc8def8c804502bf0 docs/README.md 6552741bb5639520c331a8c38d5d99c90f0ab3d9b1f11e571cb789c386bd0577 docs/STORAGE.md +0eddd7adb18bfdf801e93fde28aa06ef28db3011b4e5f60015f0c641d62c6de1 install.sh +8a593c290da5bd0d037aee86c35264512afd574a2a0f60f2c29e9db63e0109bd systemd/pigway-pi-control-api.service +1cb55de6b104a233066495e9786319305fa4f1b6f25483c108ab4f5ac9aa5700 systemd/pigway-pi-control.service +a58ff2e2ae49f481618cbb8d83b7cfbaa08ed4361eecbcab54755c886a6163eb tests/test_audit_regressions.py +8b97229abfb31c68464eb5754bc3a8fec7a6a78e216d74a86b7cc89486700740 tests/test_network_rates.py +56a7b171f7b6d93721b4888908929b36095a4cb73700a041e37f9bc796309c5e tests/test_storage_cache.py +ab555730f3f3955a1338d5d4282addfb2ea1703f6a50541146dc49adbca5694b tests/test_storage_config.py +7ef40a31988bc258eb490d39732a1790e6e533b045c3f3ff277d2e0fd274dfa1 tests/test_storage_inventory.py +9f59e7313ee58e687ae6dd29d175c67082674254ac5c5f54168b2735b64c20a4 uninstall.sh diff --git a/app/pigway_pi_control.py b/app/pigway_pi_control.py index 31e31ac..4b4500d 100755 --- a/app/pigway_pi_control.py +++ b/app/pigway_pi_control.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 import time,socket,subprocess,configparser,signal,re,argparse,json,os,platform from pathlib import Path -APP_VERSION="3.11.2" +APP_VERSION="3.11.3" import sys sys.path.append("/usr/local/share/pigway-pi-control") from plugin_api import Publisher,registered,discover @@ -185,7 +185,7 @@ def internet(): except: return False def load_cfg(): - c=configparser.ConfigParser(); c.read(CFG) + c=configparser.ConfigParser(interpolation=None); c.read(CFG) watches=[] if c.has_section("services"): for label,target in c["services"].items(): @@ -331,10 +331,10 @@ def update_active_alerts(cpu,temp,mem,disk,sample_elapsed=0.0): else: alert_manager.clear("MEMORY_HIGH") if disk is not None: if disk>=DISK_CRIT: - alert_manager.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%","FREE < 5%","CHECK STORAGE") + alert_manager.request("DISK_CRITICAL",2,"DISK CRITICAL",f"USED {disk:.1f}%",f"USED >= {DISK_CRIT:g}%","CHECK STORAGE") alert_manager.clear("DISK_HIGH") elif disk>=DISK_HIGH: - alert_manager.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%","FREE < 10%","CHECK STORAGE") + alert_manager.request("DISK_HIGH",1,"DISK SPACE LOW",f"USED {disk:.1f}%",f"USED >= {DISK_HIGH:g}%","CHECK STORAGE") alert_manager.clear("DISK_CRITICAL") else: alert_manager.clear("DISK_HIGH"); alert_manager.clear("DISK_CRITICAL") @@ -408,13 +408,14 @@ class AlertManager: "first_seen":old["first_seen"] if old else now,"metadata":metadata or {}} value["last_changed"]=old["last_changed"] if old and all(old[k]==v for k,v in value.items()) else now self.requests[request_id]=value - if old is None:log("WARN","ALERT_ACTIVE",alert=request_id,severity=severity,detail=l2) + if old is None or old["severity"]!=severity: + log({1:"WARN",2:"CRITICAL",3:"EMERGENCY"}[severity],"ALERT_ACTIVE" if old is None else "ALERT_CHANGED",alert=request_id,severity=severity,detail=l2) def clear(self,request_id): if self.requests.pop(request_id,None) is not None:log("INFO","ALERT_RECOVERED",alert=request_id) def status_snapshot(self): - values=sorted(self.requests.values(),key=lambda a:(-a["priority"],a["first_seen"],a["id"])) + values=sorted(self.requests.values(),key=lambda a:(-a["severity"],-a["priority"],a["first_seen"],a["id"])) return {"owner":values[0]["id"] if values else "NORMAL_HOME","alerts":values} diff --git a/app/pigway_pi_control_api.py b/app/pigway_pi_control_api.py index 2d69ada..efebfe4 100644 --- a/app/pigway_pi_control_api.py +++ b/app/pigway_pi_control_api.py @@ -2,26 +2,30 @@ import sys sys.path.append("/usr/local/share/pigway-pi-control") from plugin_api import discover,registered,request as plugin_request,socket_path +import math,functools,tempfile import configparser,ipaddress,json,re,secrets,shutil,subprocess,threading,time,socket -from concurrent.futures import ThreadPoolExecutor,as_completed -from http import HTTPStatus from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer from pathlib import Path -STORAGE_SCAN_LOCK=threading.Lock() -from urllib.parse import parse_qs,urlencode,urlparse -from urllib.request import Request,urlopen +from urllib.parse import parse_qs,urlparse -APP_VERSION="3.11.2" +STORAGE_SCAN_LOCK=threading.Lock() +CONFIG_LOCK=threading.RLock() + + +def config_transaction(function): + @functools.wraps(function) + def locked(*args,**kwargs): + with CONFIG_LOCK: + return function(*args,**kwargs) + return locked + +APP_VERSION="3.11.3" CFG=Path("/etc/pigway-pi-control.conf") STATUS=Path("/run/pigway-pi-control/status.json") TOKEN=Path("/etc/pigway-pi-control-api.token") ALLOWED_SECTIONS={"device","services","processes","alerts","timing","dark_mode","api","storage_names","storage_visible"} NAME_RE=re.compile(r"^[A-Za-z0-9_.-]{1,64}$") UNIT_RE=re.compile(r"^[A-Za-z0-9_.@:-]{1,128}\.service$") -RGB_KEY_RE=re.compile(r"^(cpu|power|memory|storage|network|service|normal)_[rgb]$") -FLASH_KEY_RE=re.compile(r"^(warning|critical|emergency|normal)_flash_(on|off)_ms$") -NORMAL_MODES={"off","solid","flash","flow","breathe","marquee","rainbow","colorful"} -EFFECT_COLORS={"red","green","blue","yellow","purple","cyan","white"} PROTECTED_UNITS={"pigway-pi-control-api.service"} def plugin_inventory(): @@ -36,6 +40,7 @@ def find_plugin(plugin_id): raise ValueError("unknown plugin") +@config_transaction def plugin_operation(body): plugin_id=str(body.get("id","")) if not NAME_RE.fullmatch(plugin_id):raise ValueError("invalid plugin id") @@ -66,7 +71,7 @@ def machine_identity(): return {"id":identifier,"name":name,"hostname":socket.gethostname(),"local":True,"url":""} def load_cfg(): - cfg=configparser.ConfigParser(); cfg.read(CFG) + cfg=configparser.ConfigParser(interpolation=None); cfg.read(CFG) return cfg def api_setting(key,default,cast=str): @@ -91,36 +96,39 @@ def validate_value(section,key,value,existing): if section=="timing" and key in {"oled_refresh_interval","page_interval"}:raise ValueError("display timing is managed by the hardware plugin") if section not in {"services","processes"} and key not in existing.get(section,set()): raise ValueError(f"unsupported option: {section}.{key}") - if section=="hardware" and key=="cooling_hat_enabled": - if value.lower() not in ("true","false"): raise ValueError("cooling_hat_enabled must be true or false") - elif section=="led" and key=="alert_mode": - if value.lower() not in ("breathe","flash"): raise ValueError("alert_mode must be breathe or flash") - elif section=="led" and key=="custom_min_hold_ms": - if not 1500<=int(value)<=60000: raise ValueError("custom_min_hold_ms must be 1500..60000") - elif RGB_KEY_RE.fullmatch(key): - number=int(value) - if not 0<=number<=255: raise ValueError(f"{key} must be 0..255") - elif FLASH_KEY_RE.fullmatch(key): - number=int(value) - if not 50<=number<=60000: raise ValueError(f"{key} must be 50..60000") - elif section=="led" and key=="normal_mode": - if value.lower() not in NORMAL_MODES: raise ValueError("normal_mode must be off, solid, flash, flow, breathe, marquee, rainbow or colorful") - elif section=="led" and key=="normal_effect_speed": - if int(value) not in (1,2,3): raise ValueError("normal_effect_speed must be 1, 2 or 3") - elif section=="led" and key=="normal_effect_color": - if value.lower() not in EFFECT_COLORS: raise ValueError("normal_effect_color must be red, green, blue, yellow, purple, cyan or white") - elif section=="api" and key=="port": - number=int(value) - if not 1<=number<=65535: raise ValueError("api.port must be 1..65535") - elif section=="api" and key=="log_limit": - number=int(value) - if not 1<=number<=1000: raise ValueError("api.log_limit must be 1..1000") - elif section=="api" and key=="bind": - ipaddress.ip_address(value) + if section in {"alerts","timing"}: + number=float(value) + if not math.isfinite(number):raise ValueError(f"{section}.{key} must be finite") + low,high=(0,86400) + if section=="timing": + low=0 if key in {"startup_grace_seconds","network_failure_seconds"} else 0.2 + high=3600 + elif key in {"cpu_high","cpu_recover","memory_high","memory_recover","disk_high","disk_critical"}:low,high=0,100 + elif key in {"temp_high","temp_critical"}:low,high=-50,150 + elif key in {"wifi_weak","wifi_recover"}:low,high=-120,0 + elif key=="ethernet_min_speed_mbps":low,high=0,1000000 + if not low<=number<=high:raise ValueError(f"{section}.{key} must be {low}..{high}") + if key in {"cpu_high_seconds","memory_high_seconds","ethernet_min_speed_mbps"} and not value.isdecimal(): + raise ValueError(f"{key} must be a nonnegative integer") + elif section=="dark_mode" and value.lower() not in {"on","off","true","false","0","1","yes","no"}: + raise ValueError("LED setting must be on or off") + elif section=="services" and not UNIT_RE.fullmatch(value):raise ValueError("invalid systemd service name") + elif section=="api" and key in {"port","log_limit"}: + limit=65535 if key=="port" else 1000 + if not 1<=int(value)<=limit:raise ValueError(f"api.{key} must be 1..{limit}") + elif section=="api" and key=="bind":ipaddress.ip_address(value) elif section=="device" and key=="identifier" and value.lower()!="auto" and not NAME_RE.fullmatch(value): raise ValueError("device.identifier must be auto or 1..64 letters, numbers, dot, underscore or dash") +def validate_candidate(parsed): + for lower,upper in (("cpu_recover","cpu_high"),("memory_recover","memory_high"), + ("temp_high","temp_critical"),("disk_high","disk_critical"),("wifi_weak","wifi_recover")): + if parsed.has_option("alerts",lower) and parsed.has_option("alerts",upper): + if not parsed.getfloat("alerts",lower)= 0 for k in ('total', 'used', 'free')) and values['total'] > 0 and values['used'] <= values['total']: at = data['observed_at'] - if isinstance(at, (int, float)) and 0 <= self.clock() - at < 90000: + if isinstance(at, (int, float)) and 0 <= self.clock() - at < 604800: self.values, self.observed_at = values, at if not self.error:self.next_update_at = at + 604800 except (OSError, ValueError, KeyError, TypeError): diff --git a/install.sh b/install.sh index 9897d4c..d3d9235 100755 --- a/install.sh +++ b/install.sh @@ -35,7 +35,7 @@ fi python3 - <<'PY' from pathlib import Path import configparser,re -p=Path('/etc/pigway-pi-control.conf');text=p.read_text();c=configparser.ConfigParser();c.read_string(text) +p=Path('/etc/pigway-pi-control.conf');text=p.read_text();c=configparser.ConfigParser(interpolation=None);c.read_string(text) for section,values in {'device':{'identifier':'auto','name':'auto'},'plugins':{}, 'alerts':{'ethernet_min_speed_mbps':'1000','ethernet_speed_seconds':'15'}, 'api':{'bind':'0.0.0.0','port':'6001','log_limit':'200'}}.items(): diff --git a/tests/test_audit_regressions.py b/tests/test_audit_regressions.py new file mode 100644 index 0000000..0780645 --- /dev/null +++ b/tests/test_audit_regressions.py @@ -0,0 +1,132 @@ +"""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%')