mirror of
https://github.com/home-assistant/supervisor.git
synced 2026-02-21 10:16:43 +00:00
31001280c8
* Add bus system for handling events hw/pulse/docker * give sound update back * register events * Add tests * Add debug logger * Update supervisor/coresys.py Co-authored-by: Joakim Sørensen <joasoe@gmail.com> Co-authored-by: Joakim Sørensen <joasoe@gmail.com>
45 lines
1.1 KiB
Python
45 lines
1.1 KiB
Python
"""Test bus backend."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from supervisor.const import BusEvent
|
|
from supervisor.coresys import CoreSys
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bus_event(coresys: CoreSys) -> None:
|
|
"""Test bus events over the backend."""
|
|
results = []
|
|
|
|
async def callback(data) -> None:
|
|
"""Test callback."""
|
|
results.append(data)
|
|
|
|
coresys.bus.register_event(BusEvent.HARDWARE_NEW_DEVICE, callback)
|
|
|
|
coresys.bus.fire_event(BusEvent.HARDWARE_NEW_DEVICE, None)
|
|
await asyncio.sleep(0)
|
|
assert results[-1] is None
|
|
|
|
coresys.bus.fire_event(BusEvent.HARDWARE_NEW_DEVICE, "test")
|
|
await asyncio.sleep(0)
|
|
assert results[-1] == "test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bus_event_not_called(coresys: CoreSys) -> None:
|
|
"""Test bus events over the backend."""
|
|
results = []
|
|
|
|
async def callback(data) -> None:
|
|
"""Test callback."""
|
|
results.append(data)
|
|
|
|
coresys.bus.register_event(BusEvent.HARDWARE_NEW_DEVICE, callback)
|
|
|
|
coresys.bus.fire_event(BusEvent.HARDWARE_REMOVE_DEVICE, None)
|
|
await asyncio.sleep(0)
|
|
assert len(results) == 0
|