anti
741e6bb0d3
- Rename project to stealergram throughout - Add pyproject.toml (replaces requirements.txt split, folds pytest.ini) - Replace all em-dashes with hyphens across all source files Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
116 lines
3.0 KiB
Python
116 lines
3.0 KiB
Python
"""
|
|
Tests for tui/events.py - subscribe/unsubscribe broadcast, signal_channel_changed.
|
|
"""
|
|
|
|
import queue
|
|
import pytest
|
|
from tui import events as bus
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_bus():
|
|
"""Reset all bus state between tests."""
|
|
bus._queue = None
|
|
bus.tui_active = False
|
|
bus._subscribers.clear()
|
|
bus._bot_loop = None
|
|
bus._bot_ch_ev = None
|
|
yield
|
|
bus._queue = None
|
|
bus.tui_active = False
|
|
bus._subscribers.clear()
|
|
bus._bot_loop = None
|
|
bus._bot_ch_ev = None
|
|
|
|
|
|
class TestInitBus:
|
|
def test_init_creates_queue(self):
|
|
q = bus.init_bus()
|
|
assert q is not None
|
|
assert bus.tui_active is True
|
|
|
|
def test_get_bus_returns_same_queue(self):
|
|
q = bus.init_bus()
|
|
assert bus.get_bus() is q
|
|
|
|
|
|
class TestPost:
|
|
def test_post_before_init_is_silent(self):
|
|
bus.post("event") # should not raise
|
|
|
|
def test_post_reaches_tui_queue(self):
|
|
q = bus.init_bus()
|
|
bus.post("hello")
|
|
assert q.get_nowait() == "hello"
|
|
|
|
def test_post_reaches_subscriber(self):
|
|
bus.init_bus()
|
|
sub = bus.subscribe()
|
|
bus.post("world")
|
|
assert sub.get_nowait() == "world"
|
|
|
|
def test_post_reaches_multiple_subscribers(self):
|
|
bus.init_bus()
|
|
s1 = bus.subscribe()
|
|
s2 = bus.subscribe()
|
|
bus.post(42)
|
|
assert s1.get_nowait() == 42
|
|
assert s2.get_nowait() == 42
|
|
|
|
|
|
class TestSubscribeUnsubscribe:
|
|
def test_subscribe_returns_queue(self):
|
|
q = bus.subscribe()
|
|
assert isinstance(q, queue.Queue)
|
|
assert q in bus._subscribers
|
|
|
|
def test_unsubscribe_removes_queue(self):
|
|
q = bus.subscribe()
|
|
bus.unsubscribe(q)
|
|
assert q not in bus._subscribers
|
|
|
|
def test_unsubscribe_twice_is_safe(self):
|
|
q = bus.subscribe()
|
|
bus.unsubscribe(q)
|
|
bus.unsubscribe(q) # should not raise
|
|
|
|
def test_unsubscribed_does_not_receive(self):
|
|
bus.init_bus()
|
|
sub = bus.subscribe()
|
|
bus.unsubscribe(sub)
|
|
bus.post("gone")
|
|
with pytest.raises(queue.Empty):
|
|
sub.get_nowait()
|
|
|
|
|
|
class TestSignalChannelChanged:
|
|
def test_signal_without_context_is_safe(self):
|
|
bus.signal_channel_changed() # should not raise
|
|
|
|
def test_set_bot_context_stores_refs(self):
|
|
import asyncio
|
|
|
|
async def _inner():
|
|
loop = asyncio.get_event_loop()
|
|
ev = asyncio.Event()
|
|
bus.set_bot_context(loop, ev)
|
|
assert bus._bot_loop is loop
|
|
assert bus._bot_ch_ev is ev
|
|
|
|
asyncio.run(_inner())
|
|
|
|
def test_signal_sets_event(self):
|
|
import asyncio
|
|
|
|
async def _inner():
|
|
loop = asyncio.get_event_loop()
|
|
ev = asyncio.Event()
|
|
bus.set_bot_context(loop, ev)
|
|
assert not ev.is_set()
|
|
bus.signal_channel_changed()
|
|
# give the call_soon_threadsafe a chance to fire
|
|
await asyncio.sleep(0)
|
|
assert ev.is_set()
|
|
|
|
asyncio.run(_inner())
|