Lago
8151c530da
CI / lint-and-test (push) Has been cancelled
- Remove multi-marketplace from Phase 3 - Add FastAPI web UI on port 8766 with basic auth - Add 6 Jinja2 templates (dashboard, keywords, users, ads, stats) - Add pytest test suite (45 tests, 49% coverage) - Add GitHub Actions CI/CD workflow - Update docker-compose.yml to expose web UI port - Update Dockerfile to include tests
182 lines
7.0 KiB
Python
182 lines
7.0 KiB
Python
"""Tests for notifier module functions."""
|
|
|
|
from datetime import datetime, time, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from notifier import (
|
|
_format_text,
|
|
_build_keyboard,
|
|
is_user_muted,
|
|
buffer_for_digest,
|
|
)
|
|
|
|
|
|
class TestIsUserMuted:
|
|
"""Test the is_user_muted async function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_settings_not_muted(self, mock_pool):
|
|
"""User with no settings should not be muted."""
|
|
mock_pool.fetchrow.return_value = None
|
|
result = await is_user_muted(mock_pool, 12345)
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_mute_hours_not_muted(self, mock_pool):
|
|
"""User with settings but no mute hours should not be muted."""
|
|
mock_pool.fetchrow.return_value = {"mute_start": None, "mute_end": None}
|
|
result = await is_user_muted(mock_pool, 12345)
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normal_window_muted(self, mock_pool):
|
|
"""User should be muted when current time is within normal window."""
|
|
mock_pool.fetchrow.return_value = {
|
|
"mute_start": time(8, 0),
|
|
"mute_end": time(12, 0),
|
|
}
|
|
fake_dt = datetime(2024, 1, 15, 10, 0, 0, tzinfo=timezone.utc)
|
|
mock_dt = MagicMock()
|
|
mock_dt.now.return_value = fake_dt
|
|
with patch("notifier.datetime", mock_dt):
|
|
result = await is_user_muted(mock_pool, 12345)
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normal_window_not_muted(self, mock_pool):
|
|
"""User should not be muted when current time is outside normal window."""
|
|
mock_pool.fetchrow.return_value = {
|
|
"mute_start": time(8, 0),
|
|
"mute_end": time(12, 0),
|
|
}
|
|
fake_dt = datetime(2024, 1, 15, 14, 0, 0, tzinfo=timezone.utc)
|
|
mock_dt = MagicMock()
|
|
mock_dt.now.return_value = fake_dt
|
|
with patch("notifier.datetime", mock_dt):
|
|
result = await is_user_muted(mock_pool, 12345)
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_midnight_window_muted_after_start(self, mock_pool):
|
|
"""User should be muted when time is after start of cross-midnight window."""
|
|
mock_pool.fetchrow.return_value = {
|
|
"mute_start": time(22, 0),
|
|
"mute_end": time(6, 0),
|
|
}
|
|
fake_dt = datetime(2024, 1, 15, 23, 0, 0, tzinfo=timezone.utc)
|
|
mock_dt = MagicMock()
|
|
mock_dt.now.return_value = fake_dt
|
|
with patch("notifier.datetime", mock_dt):
|
|
result = await is_user_muted(mock_pool, 12345)
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_midnight_window_muted_before_end(self, mock_pool):
|
|
"""User should be muted when time is before end of cross-midnight window."""
|
|
mock_pool.fetchrow.return_value = {
|
|
"mute_start": time(22, 0),
|
|
"mute_end": time(6, 0),
|
|
}
|
|
fake_dt = datetime(2024, 1, 15, 3, 0, 0, tzinfo=timezone.utc)
|
|
mock_dt = MagicMock()
|
|
mock_dt.now.return_value = fake_dt
|
|
with patch("notifier.datetime", mock_dt):
|
|
result = await is_user_muted(mock_pool, 12345)
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_midnight_window_not_muted(self, mock_pool):
|
|
"""User should not be muted when time is outside cross-midnight window."""
|
|
mock_pool.fetchrow.return_value = {
|
|
"mute_start": time(22, 0),
|
|
"mute_end": time(6, 0),
|
|
}
|
|
fake_dt = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_dt = MagicMock()
|
|
mock_dt.now.return_value = fake_dt
|
|
with patch("notifier.datetime", mock_dt):
|
|
result = await is_user_muted(mock_pool, 12345)
|
|
assert result is False
|
|
|
|
|
|
class TestBufferForDigest:
|
|
"""Test the buffer_for_digest async function."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_mode_on_buffers(self, mock_pool):
|
|
"""Should buffer notification when digest mode is on."""
|
|
mock_pool.fetchrow.return_value = {"digest_mode": True}
|
|
ad = {"title": "Test Ad", "price": 100.0, "keyword": "bike", "url": "https://example.com"}
|
|
await buffer_for_digest(mock_pool, 12345, ad, "ad-uuid-1")
|
|
mock_pool.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_digest_mode_off_does_not_buffer(self, mock_pool):
|
|
"""Should not buffer notification when digest mode is off."""
|
|
mock_pool.fetchrow.return_value = {"digest_mode": False}
|
|
ad = {"title": "Test Ad", "price": 100.0, "keyword": "bike", "url": "https://example.com"}
|
|
await buffer_for_digest(mock_pool, 12345, ad, "ad-uuid-1")
|
|
mock_pool.execute.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_settings_does_not_buffer(self, mock_pool):
|
|
"""Should not buffer when user has no settings."""
|
|
mock_pool.fetchrow.return_value = None
|
|
ad = {"title": "Test Ad", "price": 100.0, "keyword": "bike", "url": "https://example.com"}
|
|
await buffer_for_digest(mock_pool, 12345, ad, "ad-uuid-1")
|
|
mock_pool.execute.assert_not_called()
|
|
|
|
|
|
class TestFormatText:
|
|
"""Test the _format_text function."""
|
|
|
|
def test_basic_format(self, sample_fields):
|
|
"""Should produce expected output with basic fields."""
|
|
text = _format_text("🆕 New listing found!", sample_fields)
|
|
assert "🆕 New listing found!" in text
|
|
assert "Mountain Bike 2024" in text
|
|
assert "250" in text
|
|
assert "Vienna" in text
|
|
assert "1010" in text
|
|
|
|
def test_format_with_no_price(self):
|
|
"""Should handle missing price gracefully."""
|
|
ad = {"title": "Free Item", "location": "Graz"}
|
|
text = _format_text("Header", ad)
|
|
assert "N/A" in text
|
|
|
|
def test_format_with_no_location(self):
|
|
"""Should handle missing location gracefully."""
|
|
ad = {"title": "Item", "price": 50.0}
|
|
text = _format_text("Header", ad)
|
|
assert "Item" in text
|
|
assert "50" in text
|
|
|
|
def test_format_with_postcode(self, sample_fields):
|
|
"""Should include postcode when present."""
|
|
text = _format_text("Header", sample_fields)
|
|
assert "1010" in text
|
|
|
|
def test_format_with_published_at(self, sample_fields):
|
|
"""Should include published date when present."""
|
|
text = _format_text("Header", sample_fields)
|
|
assert "15.01.2024" in text
|
|
|
|
|
|
class TestBuildKeyboard:
|
|
"""Test the _build_keyboard function."""
|
|
|
|
def test_with_url(self, sample_fields):
|
|
"""Should create keyboard with URL button when URL is present."""
|
|
keyboard = _build_keyboard(sample_fields)
|
|
assert keyboard is not None
|
|
assert len(keyboard.inline_keyboard) == 1
|
|
assert "View Ad" in keyboard.inline_keyboard[0][0].text
|
|
|
|
def test_without_url(self):
|
|
"""Should create keyboard with no buttons when URL is missing."""
|
|
keyboard = _build_keyboard({"title": "No URL Ad"})
|
|
assert keyboard is None
|