forked from genewildish/Mainline
71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
"""
|
|
Tests for engine.ntfy module.
|
|
"""
|
|
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from engine.ntfy import NtfyPoller
|
|
|
|
|
|
class TestNtfyPollerInit:
|
|
"""Tests for NtfyPoller initialization."""
|
|
|
|
def test_init_sets_defaults(self):
|
|
"""Default values are set correctly."""
|
|
poller = NtfyPoller("http://example.com/topic")
|
|
assert poller.topic_url == "http://example.com/topic"
|
|
assert poller.reconnect_delay == 5
|
|
assert poller.display_secs == 30
|
|
|
|
def test_init_custom_values(self):
|
|
"""Custom values are set correctly."""
|
|
poller = NtfyPoller(
|
|
"http://example.com/topic", reconnect_delay=10, display_secs=60
|
|
)
|
|
assert poller.reconnect_delay == 10
|
|
assert poller.display_secs == 60
|
|
|
|
|
|
class TestNtfyPollerStart:
|
|
"""Tests for NtfyPoller.start method."""
|
|
|
|
@patch("engine.ntfy.threading.Thread")
|
|
def test_start_creates_daemon_thread(self, mock_thread):
|
|
"""start() creates and starts a daemon thread."""
|
|
mock_thread_instance = MagicMock()
|
|
mock_thread.return_value = mock_thread_instance
|
|
|
|
poller = NtfyPoller("http://example.com/topic")
|
|
result = poller.start()
|
|
|
|
assert result is True
|
|
mock_thread.assert_called_once()
|
|
args, kwargs = mock_thread.call_args
|
|
assert kwargs.get("daemon") is True
|
|
mock_thread_instance.start.assert_called_once()
|
|
|
|
|
|
class TestNtfyPollerGetActiveMessage:
|
|
"""Tests for NtfyPoller.get_active_message method."""
|
|
|
|
def test_returns_none_when_no_message(self):
|
|
"""Returns None when no message has been received."""
|
|
poller = NtfyPoller("http://example.com/topic")
|
|
result = poller.get_active_message()
|
|
assert result is None
|
|
|
|
|
|
class TestNtfyPollerDismiss:
|
|
"""Tests for NtfyPoller.dismiss method."""
|
|
|
|
def test_dismiss_clears_message(self):
|
|
"""dismiss() clears the current message."""
|
|
poller = NtfyPoller("http://example.com/topic")
|
|
|
|
with patch.object(poller, "_lock"):
|
|
poller._message = ("Title", "Body", time.monotonic())
|
|
poller.dismiss()
|
|
|
|
assert poller._message is None
|