Files
sunnypilot/openpilot/selfdrive/selfdrived/tests/test_alertmanager.py
Jason Wen 7461f70fdb Merge commit 'b7c333cf3fee117779515c9ebfd7b2beb164fa81' into sync-20260813
# Conflicts:
#	README.md
#	SConstruct
#	conftest.py
#	docs/CARS.md
#	msgq_repo
#	opendbc_repo
#	openpilot/common/params_keys.h
#	openpilot/common/params_pyx.pyx
#	openpilot/common/tests/test_swaglog.cc
#	openpilot/selfdrive/car/card.py
#	openpilot/selfdrive/car/tests/test_car_interfaces.py
#	openpilot/selfdrive/car/tests/test_cruise_speed.py
#	openpilot/selfdrive/car/tests/test_models.py
#	openpilot/selfdrive/controls/controlsd.py
#	openpilot/selfdrive/controls/lib/latcontrol_torque.py
#	openpilot/selfdrive/controls/lib/longitudinal_planner.py
#	openpilot/selfdrive/controls/plannerd.py
#	openpilot/selfdrive/controls/radard.py
#	openpilot/selfdrive/controls/tests/test_longcontrol.py
#	openpilot/selfdrive/locationd/torqued.py
#	openpilot/selfdrive/modeld/modeld.py
#	openpilot/selfdrive/monitoring/dmonitoringd.py
#	openpilot/selfdrive/monitoring/test_monitoring.py
#	openpilot/selfdrive/selfdrived/selfdrived.py
#	openpilot/selfdrive/selfdrived/tests/test_alertmanager.py
#	openpilot/selfdrive/test/longitudinal_maneuvers/plant.py
#	openpilot/selfdrive/test/process_replay/migration.py
#	openpilot/selfdrive/test/process_replay/process_replay.py
#	openpilot/selfdrive/ui/feedback/feedbackd.py
#	openpilot/selfdrive/ui/layouts/settings/device.py
#	openpilot/selfdrive/ui/layouts/settings/toggles.py
#	openpilot/selfdrive/ui/mici/layouts/onboarding.py
#	openpilot/selfdrive/ui/onroad/augmented_road_view.py
#	openpilot/selfdrive/ui/tests/test_soundd.py
#	openpilot/selfdrive/ui/translations/app.pot
#	openpilot/selfdrive/ui/translations/app_de.po
#	openpilot/selfdrive/ui/translations/app_en.po
#	openpilot/selfdrive/ui/translations/app_es.po
#	openpilot/selfdrive/ui/translations/app_fr.po
#	openpilot/selfdrive/ui/translations/app_ja.po
#	openpilot/selfdrive/ui/translations/app_ko.po
#	openpilot/selfdrive/ui/translations/app_pt-BR.po
#	openpilot/selfdrive/ui/translations/app_th.po
#	openpilot/selfdrive/ui/translations/app_tr.po
#	openpilot/selfdrive/ui/translations/app_uk.po
#	openpilot/selfdrive/ui/translations/app_zh-CHS.po
#	openpilot/selfdrive/ui/translations/app_zh-CHT.po
#	openpilot/system/athena/athenad.py
#	openpilot/system/hardware/hardwared.py
#	openpilot/system/loggerd/deleter.py
#	openpilot/system/manager/process_config.py
#	openpilot/system/ui/lib/application.py
#	panda
#	pyproject.toml
#	tinygrad_repo
#	uv.lock
2026-08-14 15:35:10 -04:00

62 lines
2.1 KiB
Python

import random
from openpilot.common.test import OpenpilotTestCase
from openpilot.selfdrive.selfdrived.events import Alert, EVENTS
from openpilot.selfdrive.selfdrived.alertmanager import AlertManager
from openpilot.sunnypilot.selfdrive.selfdrived.events_base import EmptyAlert
class TestAlertManager(OpenpilotTestCase):
def test_duration(self):
"""
Enforce that an alert lasts for max(alert duration, duration the alert is added)
"""
for duration in range(1, 100):
alert = None
while not isinstance(alert, Alert):
event = random.choice([e for e in EVENTS.values() if len(e)])
alert = random.choice(list(event.values()))
alert.duration = duration
# check two cases:
# - alert is added to AM for <= the alert's duration
# - alert is added to AM for > alert's duration
for greater in (True, False):
if greater:
add_duration = duration + random.randint(1, 10)
else:
add_duration = random.randint(1, duration)
show_duration = max(duration, add_duration)
AM = AlertManager()
for frame in range(duration+10):
if frame < add_duration:
AM.add_many(frame, [alert, ])
AM.process_alerts(frame, set())
shown = AM.current_alert != EmptyAlert
should_show = frame <= show_duration
assert shown == should_show, f"{frame=} {add_duration=} {duration=}"
# check one case:
# - if alert is re-added to AM before it ends the duration is extended
if duration > 1:
AM = AlertManager()
show_duration = duration * 2
for frame in range(duration * 2 + 10):
if frame == 0:
AM.add_many(frame, [alert, ])
if frame == duration:
# add alert one frame before it ends
assert AM.current_alert == alert
AM.add_many(frame, [alert, ])
AM.process_alerts(frame, set())
shown = AM.current_alert != EmptyAlert
should_show = frame <= show_duration
assert shown == should_show, f"{frame=} {duration=}"