26 KiB
26 KiB
Task: Test suite with pytest (≥80% coverage)
Description
The project currently has zero automated tests. Every change is verified manually by watching logs or sending test messages to the bot. This makes refactoring risky and prevents CI/CD automation.
This task introduces a comprehensive pytest test suite covering all critical paths: scraper parsing, notification logic, price/postcode filters, retry queue behavior, and scheduler flow. Coverage threshold is set to 80% minimum.
Architecture
┌───────────────────────────────────────┐
│ Tests structure │
│ │
│ worker/tests/ │
│ ├── conftest.py │
│ │ (fixtures: mock_pool, mock_bot)│
│ ├── test_scraper.py │
│ │ (parsing, pagination) │
│ ├── test_notifier.py │
│ │ (send, retry queue, digest) │
│ ├── test_filters.py │
│ │ (price, postcode, mute hours) │
│ ├── test_scheduler.py │
│ │ (cycle flow, shutdown) │
│ └── test_health.py │
│ (healthcheck endpoint) │
│ │
└──────────┬────────────────────────────┘
│
▼
┌───────────────────────────────────────┐
│ CI/CD Pipeline (.github/workflows/ci.yml)
│ │
│ on: push to main, feat/*; pull_request│
│ │
│ jobs: │
│ lint-and-test: │
│ └─ python 3.12 │
│ ├─ flake8 (error-only) │
│ ├─ pytest --cov=src tests/ │
│ └─ coverage >= 80% │
└───────────────────────────────────────┘
Test strategy:
Unit tests (majority):
- Isolate each function/method with mocks
- Test edge cases: missing fields, NULL prices, empty results
- Fast (<1s per test)
Integration tests (minority):
- Real HTTP to willhaben API (cached responses only)
- PostgreSQL test container via docker-compose
Mocking strategy:
- Telegram Bot: mock `bot.send_message()` → verify call count + content
- Asyncpg pool: mock fetchval/fetch/execute → return canned data
- httpx client: use pytest-httpx to intercept and return fixtures
Key design decisions
- pytest over unittest. Cleaner syntax, better fixture system, easier async support.
- pytest-asyncio for testing async functions directly without wrapping in
loop.run_until_complete().- Adding as test dependency:
pip install pytest pytest-asyncio pytest-cov httpx[socks]
- Adding as test dependency:
- Coverage threshold at 80% enforced via
pyproject.toml. Failures are actionable (which files/functions need coverage). - Snapshot testing for HTTP responses. Save real willhaben API responses as JSON fixtures to avoid live network calls in CI.
Implementation Details
1. Add test dependencies and configuration
In worker/requirements-test.txt:
pytest>=8.0
pytest-asyncio>=0.24
pytest-cov>=6.0
aioresponses>=0.7 # mock aiohttp responses for health endpoint tests
Create pyproject.toml in the project root:
[tool.pytest.ini_options]
testpaths = ["worker/tests"]
asyncio_mode = "auto"
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
[tool.coverage.run]
source = ["worker/src"]
omit = [
"*/tests/*",
"*/migrate.py", # migration runner — tested manually against real DB
]
[tool.coverage.report]
fail_under = 80.0
show_missing = true
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
]
2. Create worker/tests/conftest.py (fixtures)
import pytest
from unittest.mock import AsyncMock, MagicMock
@pytest.fixture
def mock_pool():
"""Mock asyncpg.Pool with fetchval/fetch/execute."""
pool = AsyncMock()
pool.fetchval = AsyncMock(return_value=None)
pool.fetch = AsyncMock(return_value=[])
pool.execute = AsyncMock(return_value="DONE 1")
return pool
@pytest.fixture
def mock_bot():
"""Mock Telegram Bot instance."""
bot = MagicMock()
bot.send_message = AsyncMock(return_value=True)
bot.get_me = AsyncMock(return_value={"id": "bot_user", "is_bot": True})
return bot
@pytest.fixture
def sample_willhaben_response():
"""Realistic willhaben API response for testing."""
return {
"rowsFound": 45,
"advertSummaryList": {
"advertSummary": [
{
"id": "123456789",
"title": {"Value": "RTX 3090 Gaming X - Top Zustand"},
"linkUrl": "https://www.willhaben.at/iad/markt/123456789-rtx-3090-gaming-x",
"attributes": [
{
"name": "PRICE",
"items": [{"name": "priceString", "valueString": "750.00"}]
},
{
"name": "PUBLISHED",
"items": [{"name": "publishedString", "valueString": "2026-07-04T12:30:00+02:00"}]
},
{
"name": "LOCATION",
"items": [
{"name": "CityName", "valueString": "Wien"},
{"name": "ZIP", "valueString": "1010"},
]
}
]
},
]
}
}
@pytest.fixture
def sample_ad_normalized():
"""Expected normalized ad dict from willhaben response."""
return {
"id": "123456789",
"marketplace": "willhaben",
"title": "RTX 3090 Gaming X - Top Zustand",
"price": 75000, # in cents
"currency": "EUR",
"url": "https://www.willhaben.at/iad/markt/123456789-rtx-3090-gaming-x",
"published_at": ..., # will be datetime object
"location": {
"city": "Wien",
"postcode": "1010",
"region": None,
},
"attributes": {...},
}
@pytest.fixture
def sample_keyword_row():
"""Sample keyword DB row."""
return {
"id": "kw-uuid-here",
"keyword_name": "rtx 3090",
"telegram_id": "298181113",
"is_active": True,
"price_min": 50000, # €500 minimum
"price_max": 1000000, # €10000 maximum
"allowed_postcodes": ["1010", "1020"],
"last_scraped_at": None,
"ads_cursor": None,
}
3. Create worker/tests/test_scraper.py
import pytest
from unittest.mock import AsyncMock, patch
from scrapers.willhaben import WillhabenScraper
from datetime import datetime, timezone
class TestWillhabenScraper:
def test_build_query(self):
scraper = WillhabenScraper()
url = scraper.build_query("rtx 3090", offset=30)
assert "keyword=rtx+3090" in url
assert "offset=30" in url
assert "rows=30" in url
def test_build_query_default_offset(self):
scraper = WillhabenScraper()
url = scraper.build_query("gtx 1660")
assert "offset=0" in url
def test_parse_page_empty_response(self, sample_willhaben_response):
scraper = WillhabenScraper()
# Empty response
empty = {"rowsFound": 0, "advertSummaryList": {"advertSummary": []}}
result = scraper.parse_page(empty)
assert result == []
def test_normalize_ad(self, sample_willhaben_response):
scraper = WillhabenScraper()
raw_ad = sample_willhaben_response["advertSummaryList"]["advertSummary"][0]
ad = scraper.normalize_ad(raw_ad)
assert ad["id"] == "123456789"
assert ad["marketplace"] == "willhaben"
assert ad["price"] == 75000 # cents
assert ad["location"]["postcode"] == "1010"
def test_normalize_ad_missing_price(self):
scraper = WillhabenScraper()
raw_ad = {
"id": "no-price",
"title": {"Value": "Free RTX"},
"linkUrl": "https://example.com",
"attributes": [], # no price
}
ad = scraper.normalize_ad(raw_ad)
assert ad["price"] is None
@pytest.mark.asyncio
async def test_fetch_ads_pagination(self):
scraper = WillhabenScraper()
with patch.object(scraper, '_fetch_with_retry') as mock_fetch:
# Simulate 2 pages of results
page1 = {
"rowsFound": 45,
"advertSummaryList": {"advertSummary": [
{"id": f"ad{i}", "title": {"Value": f"Ad {i}"},
"linkUrl": f"https://example.com/{i}",
"attributes": [{"name": "PUBLISHED", "items": [
{"name": "publishedString",
"valueString": "2026-07-04T15:30:00+02:00"}]}]},
] for i in range(3)}]
}
page2 = {
"rowsFound": 45,
"advertSummaryList": {"advertSummary": [
{"id": f"ad{i}", "title": {"Value": f"Ad {i}"},
"linkUrl": f"https://example.com/{i}",
"attributes": [{"name": "PUBLISHED", "items": [
{"name": "publishedString",
"valueString": "2026-07-04T15:30:00+02:00"}]}]},
] for i in range(3, 6)}]
}
mock_fetch.side_effect = [page1, page2]
ads, total = await scraper.fetch_ads("test keyword", max_pages=2)
assert len(ads) == 6 # 3 from each page
class TestScraperBase:
def test_headers_default(self):
scraper = WillhabenScraper()
headers = scraper.headers
assert "Accept" in headers
assert "User-Agent" in headers
@pytest.mark.asyncio
async def test_fetch_with_retry_exhausts_retries(self):
import httpx
scraper = WillhabenScraper()
client = AsyncMock()
client.get = AsyncMock(side_effect=httpx.ConnectError("connection refused"))
with pytest.raises(httpx.ConnectError):
await scraper._fetch_with_retry(client, "http://example.com", max_retries=2)
4. Create worker/tests/test_filters.py
import pytest
from unittest.mock import AsyncMock
class TestPriceFilters:
async def test_pass_when_no_min(self):
"""Ad passes when no price minimum is set."""
# Import the actual function being tested
from notifier import _extract_price # or wherever it lives
kw_row = {"price_min": None, "price_max": None}
ad_dict = {
"attributes": [{"name": "PRICE", "items": [
{"name": "priceString", "valueString": "50.00"}]}]
}
# The check function (to be implemented in main.py)
from ..main import _check_price_filters # actual implementation
result = await _check_price_filters(ad_dict, kw_row)
assert result is True
async def test_pass_when_no_max(self):
kw_row = {"price_min": 1000, "price_max": None}
ad_dict = {
"attributes": [{"name": "PRICE", "items": [
{"name": "priceString", "valueString": "100.00"}]}]
}
from ..main import _check_price_filters
result = await _check_price_filters(ad_dict, kw_row)
assert result is True
async def test_fail_below_min(self):
kw_row = {"price_min": 50000, "price_max": None} # €500 min
ad_dict = {
"attributes": [{"name": "PRICE", "items": [
{"name": "priceString", "valueString": "10.00"}]}] # €10 — too cheap
}
from ..main import _check_price_filters
result = await _check_price_filters(ad_dict, kw_row)
assert result is False
async def test_fail_above_max(self):
kw_row = {"price_min": None, "price_max": 1000} # €10 max
ad_dict = {
"attributes": [{"name": "PRICE", "items": [
{"name": "priceString", "valueString": "500.00"}]}] # €500 — too expensive
}
from ..main import _check_price_filters
result = await _check_price_filters(ad_dict, kw_row)
assert result is False
class TestPostcodeFilters:
async def test_pass_when_no_filter(self):
kw_row = {"allowed_postcodes": None}
ad_dict = {
"attributes": [{"name": "LOCATION", "items": [
{"name": "ZIP", "valueString": "1010"}]}]
}
from ..main import _check_postcode_filter
result = await _check_postcode_filter(ad_dict, kw_row)
assert result is True
async def test_pass_when_matching(self):
kw_row = {"allowed_postcodes": ["1010", "1020"]}
ad_dict = {
"attributes": [{"name": "LOCATION", "items": [
{"name": "ZIP", "valueString": "1010"}]}]
}
from ..main import _check_postcode_filter
result = await _check_postcode_filter(ad_dict, kw_row)
assert result is True
async def test_fail_when_not_matching(self):
kw_row = {"allowed_postcodes": ["1010", "1020"]}
ad_dict = {
"attributes": [{"name": "LOCATION", "items": [
{"name": "ZIP", "valueString": "8010"}]}] # Graz — not in allowed list
}
from ..main import _check_postcode_filter
result = await _check_postcode_filter(ad_dict, kw_row)
assert result is False
async def test_fail_when_no_postcode_in_ad(self):
kw_row = {"allowed_postcodes": ["1010", "1020"]}
ad_dict = {
"attributes": [] # no location info
}
from ..main import _check_postcode_filter
result = await _check_postcode_filter(ad_dict, kw_row)
assert result is False
class TestMuteHours:
async def test_no_mute_when_not_configured(self, mock_pool):
mock_pool.fetchrow.return_value = None
from ..notifier import _is_in_mute_hours
result = await _is_in_mute_hours("298181113", mock_pool)
assert result is False
async def test_no_mute_outside_window(self, mock_pool):
from datetime import time
mock_pool.fetchrow.return_value = {
"mute_start": time(22, 0), # 10 PM
"mute_end": time(7, 0), # 7 AM
}
# Mock current time to noon UTC (outside mute window)
with patch("notifier.datetime") as mock_dt:
from datetime import datetime, timezone
mock_dt.now.return_value = datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc)
from ..notifier import _is_in_mute_hours
result = await _is_in_mute_hours("298181113", mock_pool)
assert result is False
async def test_muted_during_window(self, mock_pool):
from datetime import time
mock_pool.fetchrow.return_value = {
"mute_start": time(22, 0),
"mute_end": time(7, 0),
}
with patch("notifier.datetime") as mock_dt:
from datetime import datetime, timezone
mock_dt.now.return_value = datetime(2026, 1, 1, 3, 0, tzinfo=timezone.utc)
from ..notifier import _is_in_mute_hours
result = await _is_in_mute_hours("298181113", mock_pool)
assert result is True
class TestNotificationQueue:
async def test_enqueue_on_failure(self, mock_pool):
from ..notifier import _enqueue_retry
await _enqueue_retry(
ad_id="ad-uuid-here",
telegram_id="298181113",
message_text="Test notification",
notif_type="new",
error_msg="Telegram API timeout",
)
mock_pool.execute.assert_called_once()
async def test_no_duplicate_enqueue(self, mock_pool):
"""Same ad+user should not create duplicate queue entries."""
mock_pool.fetchval.return_value = "already-exists-uuid"
from ..notifier import _enqueue_retry
await _enqueue_retry(
ad_id="ad-uuid-here",
telegram_id="298181113",
message_text="Test",
notif_type="new",
error_msg="timeout",
)
# fetchval called (to check), but execute NOT called (no insert)
mock_pool.fetchval.assert_called_once()
mock_pool.execute.assert_not_called()
5. Create worker/tests/test_scheduler.py
import pytest
from unittest.mock import AsyncMock, patch
class TestSchedulerFlow:
@pytest.mark.asyncio
async def test_process_notification_queue_retries_pending(self, mock_pool):
"""Pending items should be retried if backoff period has elapsed."""
from datetime import datetime, timezone
mock_pool.fetch.return_value = [
{
"id": "queue-item-1",
"ad_id": "ad-uuid",
"telegram_id": "298181113",
"message_text": "Retry this ad",
"type": "new",
"attempts": 0,
"max_attempts": 5,
"last_error": "timeout",
"updated_at": datetime(2026, 7, 4, 10, 0, tzinfo=timezone.utc),
},
]
with patch("main.get_application_bot") as mock_get_bot:
mock_bot = AsyncMock()
mock_bot.send_message = AsyncMock()
mock_get_bot.return_value = mock_bot
from ..main import process_notification_queue
result = await process_notification_queue()
assert result == 1
@pytest.mark.asyncio
async def test_process_notification_queue_dead_after_max_attempts(self, mock_pool):
"""Items exceeding max_attempts should be marked as dead."""
from datetime import datetime, timezone
mock_pool.fetch.return_value = [
{
"id": "queue-item-2",
"ad_id": "ad-uuid",
"telegram_id": "298181113",
"message_text": "Will fail again",
"type": "new",
"attempts": 5, # already at max
"max_attempts": 5,
"last_error": "user blocked bot",
"updated_at": datetime(2026, 7, 4, 10, 0, tzinfo=timezone.utc),
},
]
with patch("main.get_application_bot") as mock_get_bot:
import telegram.error
mock_bot = AsyncMock()
mock_bot.send_message = AsyncMock(
side_effect=telegram.error.TelegramError("blocked")
)
mock_get_bot.return_value = mock_bot
from ..main import process_notification_queue
await process_notification_queue()
# Should have updated to 'dead' status
calls = [c[0] for c in mock_pool.execute.call_args_list]
assert any("status = 'dead'" in str(c) or "status=$2" in str(c)
for c in calls), "Item should be marked as dead"
class TestDigestFlushing:
@pytest.mark.asyncio
async def test_flush_digest_buffers(self, mock_pool):
"""Buffered items older than interval should be flushed."""
from datetime import datetime, timezone
mock_pool.fetch.side_effect = [
# First fetch: get digest-enabled users
[{"telegram_id": "298181113", "digest_interval": 60}],
# Second fetch: get buffered items
[
{"id": "buf-1", "keyword": "rtx 3090",
"title": "Ad 1", "price": 75000,
"url": "https://...", "ad_id": "ad-uuid"},
{"id": "buf-2", "keyword": "rtx 3090",
"title": "Ad 2", "price": 68000,
"url": "https://...", "ad_id": "ad-uuid-2"},
],
]
with patch("main.get_application_bot") as mock_get_bot:
mock_bot = AsyncMock()
mock_bot.send_message = AsyncMock()
mock_get_bot.return_value = mock_bot
from ..main import flush_digest_buffers
result = await flush_digest_buffers()
assert result == 1 # one digest sent
@pytest.mark.asyncio
async def test_flush_empty_buffer(self, mock_pool):
"""No buffered items should result in no action."""
mock_pool.fetch.side_effect = [
[{"telegram_id": "298181113", "digest_interval": 60}],
[], # empty buffer
]
from ..main import flush_digest_buffers
result = await flush_digest_buffers()
assert result == 0
class TestGracefulShutdown:
@pytest.mark.asyncio
async def test_cleanup_stops_scheduler_and_closes_pool(self):
"""Cleanup should cancel scheduler, stop bot, close DB pool."""
from telegram.ext import Application
app = AsyncMock(spec=Application)
app.updater.running = True
with patch("main._scheduler_task") as mock_task:
mock_task.done.return_value = False
from ..main import cleanup
await cleanup(app)
mock_task.cancel.assert_called_once()
app.updater.stop_polling.assert_called_once()
6. Create worker/tests/test_health.py
import pytest
@pytest.mark.asyncio
async def test_health_endpoint_ok():
"""Health endpoint should return 200 when system is healthy."""
from health import create_health_app
app = create_health_app()
with patch("health.get_pool") as mock_pool_get:
mock_pool = AsyncMock()
mock_pool.fetchval = AsyncMock(return_value=1)
mock_pool_get.return_value = mock_pool
runner = web.AppRunner(app)
await runner.setup()
try:
from aiohttp.test_utils import TestClient, TestServer
client = TestClient(TestServer(runner))
async with client:
resp = await client.get("/health")
assert resp.status == 200
data = await resp.json()
assert data["status"] == "ok"
assert "db_connected" in data
finally:
await runner.cleanup()
@pytest.mark.asyncio
async def test_health_endpoint_unhealthy_db():
"""Health endpoint should return 503 when DB is unreachable."""
from health import create_health_app
app = create_health_app()
with patch("health.get_pool") as mock_pool_get:
mock_pool_get.side_effect = Exception("connection refused")
runner = web.AppRunner(app)
await runner.setup()
try:
from aiohttp.test_utils import TestClient, TestServer
client = TestClient(TestServer(runner))
async with client:
resp = await client.get("/health")
assert resp.status == 503
data = await resp.json()
assert data["status"] == "unhealthy"
finally:
await runner.cleanup()
7. Create GitHub Actions workflow .github/workflows/ci.yml
name: CI — Lint & Test
on:
push:
branches: [main, 'feat/*']
pull_request:
branches: [main]
jobs:
lint-and-test:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python 3.12
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install dependencies
run: |
pip install --upgrade pip
pip install -r worker/requirements.txt
pip install -r worker/requirements-test.txt
- name: Lint with flake8
run: |
flake8 worker/src/ --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 worker/tests/ --count --select=E9,F63,F7,F82 --show-source --statistics
- name: Test with pytest + coverage
run: |
cd worker
python -m pytest tests/ --cov=src --cov-report=term-missing --cov-fail-under=80
build-docker:
runs-on: ubuntu-latest
needs: lint-and-test
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Build Docker image
run: docker compose -f docker-compose.yml build worker
- name: Test healthcheck
run: |
docker compose up -d worker
sleep 5
docker inspect --format='{{.State.Health.Status}}' willhaben-tracker-worker-1 || true
docker compose down
Acceptance Criteria
pytestruns with 0 failures and ≥80% coverage on all source files- GitHub Actions pipeline passes on every push to
mainand feature branches - Tests cover: scraper parsing, pagination, price filters, postcode filters, mute hours, notification retry queue, digest flushing, graceful shutdown, healthcheck endpoint
- Flake8 linting (error-level checks) passes in CI
- Docker image builds successfully after all tests pass
- Adding a new test file automatically includes it in the coverage report