Mirrors/kleinanzeigen-bot
1
0
Fork 0
mirror of https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git synced 2026-03-12 02:31:45 +01:00
Code Issues Packages Projects Releases Wiki Activity

test: strengthen coverage for sessions, logging, and update check (#686)

Browse Source 
## ℹ️ Description
* Strengthen the session/logging/update-check tests to exercise real
resources and guards while bringing the update-check docs in line with
the supported interval units.
- Link to the related issue(s): Issue #N/A

## 📋 Changes Summary
- Reworked the `WebScrapingMixin` session tests so they capture each
`stop` handler before the browser reference is nulled, ensuring cleanup
logic is exercised without crashing.
- Added targeted publish and update-check tests that patch the async
helpers, guard logic, and logging handlers while confirming
`requests.get` is skipped when the state gate is closed.
- Updated `docs/update-check.md` to list only the actually supported
interval units (up to 30 days) and noted the new guard coverage in the
changelog.

### ⚙️ Type of Change
- [x] 🐞 Bug fix (non-breaking change which fixes an issue)
- [ ] ✨ New feature (adds new functionality without breaking existing
usage)
- [ ] 💥 Breaking change (changes that might break existing user setups,
scripts, or configurations)

## ✅ Checklist
- [x] I have reviewed my changes to ensure they meet the project's
standards.
- [x] I have tested my changes and ensured that all tests pass (`pdm run
test`).
- [x] I have formatted the code (`pdm run format`).
- [x] I have verified that linting passes (`pdm run lint`).
- [x] I have updated documentation where necessary.

By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Tests**
* Expanded test coverage for publish workflow orchestration and update
checking interval behavior.
* Added comprehensive browser session cleanup tests, including
idempotent operations and edge case handling.
* Consolidated logging configuration tests with improved handler
management validation.
  * Refined test fixtures and assertions for better test reliability.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Jens
2025-11-17 11:02:18 +01:00
committed by GitHub
parent b99966817c
commit 89df56bf8b
4 changed files with 120 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
docs/update-check.md
View File

@@ -22,16 +22,15 @@ The interval is specified as a number followed by a unit:
- `m`: minutes
- `h`: hours
- `d`: days
- `w`: weeks
Examples:
- `7d`: Check every 7 days
- `12h`: Check every 12 hours
- `1w`: Check every week
- `30d`: Check every 30 days
Validation rules:
- Minimum interval: 1 day (`1d`)
- Maximum interval: 4 weeks (`4w`)
- Maximum interval: 30 days (`30d`, roughly 4 weeks)
- Value must be positive
- Only supported units are allowed

124
tests/unit/test_init.py
View File

@@ -1,7 +1,7 @@
# SPDX-FileCopyrightText: © Jens Bergmann and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import copy, io, logging, os, tempfile # isort: skip
import copy, io, json, logging, os, tempfile # isort: skip
from collections.abc import Generator
from contextlib import redirect_stdout
from datetime import timedelta
@@ -12,7 +12,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from pydantic import ValidationError
from kleinanzeigen_bot import LOG, KleinanzeigenBot, misc
from kleinanzeigen_bot import LOG, AdUpdateStrategy, KleinanzeigenBot, misc
from kleinanzeigen_bot._version import __version__
from kleinanzeigen_bot.model.ad_model import Ad
from kleinanzeigen_bot.model.config_model import AdDefaults, Config, PublishingConfig
@@ -24,7 +24,6 @@ from kleinanzeigen_bot.utils.web_scraping_mixin import By, Element
def mock_page() -> MagicMock:
"""Provide a mock page object for testing."""
mock = MagicMock()
# Mock async methods
mock.sleep = AsyncMock()
mock.evaluate = AsyncMock()
mock.click = AsyncMock()
@@ -39,26 +38,6 @@ def mock_page() -> MagicMock:
return mock
@pytest.mark.asyncio
async def test_mock_page_fixture(mock_page:MagicMock) -> None:
"""Test that the mock_page fixture is properly configured."""
# Test that all required async methods are present
assert hasattr(mock_page, "sleep")
assert hasattr(mock_page, "evaluate")
assert hasattr(mock_page, "click")
assert hasattr(mock_page, "type")
assert hasattr(mock_page, "select")
assert hasattr(mock_page, "wait_for_selector")
assert hasattr(mock_page, "wait_for_navigation")
assert hasattr(mock_page, "wait_for_load_state")
assert hasattr(mock_page, "content")
assert hasattr(mock_page, "goto")
assert hasattr(mock_page, "close")
# Test that content returns expected value
assert await mock_page.content() == "<html></html>"
@pytest.fixture
def base_ad_config() -> dict[str, Any]:
"""Provide a base ad configuration that can be used across tests."""
@@ -114,30 +93,6 @@ def remove_fields(config:dict[str, Any], *fields:str) -> dict[str, Any]:
return result
def test_remove_fields() -> None:
"""Test the remove_fields helper function."""
test_config = {
"field1": "value1",
"field2": "value2",
"nested": {
"field3": "value3"
}
}
# Test removing top-level field
result = remove_fields(test_config, "field1")
assert "field1" not in result
assert "field2" in result
# Test removing nested field
result = remove_fields(test_config, "nested.field3")
assert "field3" not in result["nested"]
# Test removing non-existent field
result = remove_fields(test_config, "nonexistent")
assert result == test_config
@pytest.fixture
def minimal_ad_config(base_ad_config:dict[str, Any]) -> dict[str, Any]:
"""Provide a minimal ad configuration with only required fields."""
@@ -187,24 +142,37 @@ class TestKleinanzeigenBotInitialization:
class TestKleinanzeigenBotLogging:
"""Tests for logging functionality."""
def test_configure_file_logging_creates_log_file(self, test_bot:KleinanzeigenBot, log_file_path:str) -> None:
"""Verify that file logging configuration creates the log file."""
test_bot.log_file_path = log_file_path
def test_configure_file_logging_adds_and_removes_handlers(
self,
test_bot:KleinanzeigenBot,
tmp_path:Path
) -> None:
"""Ensure file logging registers a handler and cleans it up afterward."""
log_path = tmp_path / "bot.log"
test_bot.log_file_path = str(log_path)
root_logger = logging.getLogger()
initial_handlers = list(root_logger.handlers)
test_bot.configure_file_logging()
assert test_bot.file_log is not None
assert os.path.exists(log_file_path)
assert log_path.exists()
assert len(root_logger.handlers) == len(initial_handlers) + 1
# Test that calling again doesn't recreate logger
original_file_log = test_bot.file_log
test_bot.configure_file_logging()
assert test_bot.file_log is original_file_log
test_bot.file_log.close()
assert test_bot.file_log.is_closed()
assert len(root_logger.handlers) == len(initial_handlers)
def test_configure_file_logging_skips_when_path_missing(self, test_bot:KleinanzeigenBot) -> None:
"""Ensure no handler is added when no log path is configured."""
root_logger = logging.getLogger()
initial_handlers = list(root_logger.handlers)
def test_configure_file_logging_disabled_when_no_path(self, test_bot:KleinanzeigenBot) -> None:
"""Verify that logging is disabled when no path is provided."""
test_bot.log_file_path = None
test_bot.configure_file_logging()
assert test_bot.file_log is None
assert list(root_logger.handlers) == initial_handlers
class TestKleinanzeigenBotCommandLine:
@@ -510,26 +478,32 @@ class TestKleinanzeigenBotBasics:
"""Test version retrieval."""
assert test_bot.get_version() == __version__
def test_configure_file_logging(self, test_bot:KleinanzeigenBot, log_file_path:str) -> None:
"""Test file logging configuration."""
test_bot.log_file_path = log_file_path
test_bot.configure_file_logging()
assert test_bot.file_log is not None
assert os.path.exists(log_file_path)
@pytest.mark.asyncio
async def test_publish_ads_triggers_publish_and_cleanup(
self,
test_bot:KleinanzeigenBot,
base_ad_config:dict[str, Any],
mock_page:MagicMock,
) -> None:
"""Simulate publish job wiring without hitting the live site."""
test_bot.page = mock_page
test_bot.config.publishing.delete_old_ads = "AFTER_PUBLISH"
test_bot.keep_old_ads = False
def test_configure_file_logging_no_path(self, test_bot:KleinanzeigenBot) -> None:
"""Test file logging configuration with no path."""
test_bot.log_file_path = None
test_bot.configure_file_logging()
assert test_bot.file_log is None
payload:dict[str, list[Any]] = {"ads": []}
ad_cfgs:list[tuple[str, Ad, dict[str, Any]]] = [("ad.yaml", Ad.model_validate(base_ad_config), {})]
def test_close_browser_session(self, test_bot:KleinanzeigenBot) -> None:
"""Test closing browser session."""
mock_close = MagicMock()
test_bot.page = MagicMock() # Ensure page exists to trigger cleanup
with patch.object(test_bot, "close_browser_session", new = mock_close):
test_bot.close_browser_session() # Call directly instead of relying on __del__
mock_close.assert_called_once()
with patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = {"content": json.dumps(payload)}) as web_request_mock, \
patch.object(test_bot, "publish_ad", new_callable = AsyncMock) as publish_ad_mock, \
patch.object(test_bot, "web_await", new_callable = AsyncMock, return_value = True) as web_await_mock, \
patch.object(test_bot, "delete_ad", new_callable = AsyncMock) as delete_ad_mock:
await test_bot.publish_ads(ad_cfgs)
web_request_mock.assert_awaited_once_with(f"{test_bot.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT")
publish_ad_mock.assert_awaited_once_with("ad.yaml", ad_cfgs[0][1], {}, [], AdUpdateStrategy.REPLACE)
web_await_mock.assert_awaited_once()
delete_ad_mock.assert_awaited_once_with(ad_cfgs[0][1], [], delete_old_ads_by_title = False)
def test_get_root_url(self, test_bot:KleinanzeigenBot) -> None:
"""Test root URL retrieval."""

20
tests/unit/test_update_checker.py
View File

@@ -5,6 +5,7 @@
from __future__ import annotations
import json
import logging
from datetime import datetime, timedelta, timezone, tzinfo
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
@@ -274,6 +275,25 @@ class TestUpdateChecker:
expected = "You are on the latest version: 2025+fb00f11 (compared to fb00f11 in channel latest)"
assert any(expected in r.getMessage() for r in caplog.records)
def test_check_for_updates_respects_interval_gate(
self,
config:Config,
caplog:pytest.LogCaptureFixture
) -> None:
"""Ensure the interval guard short-circuits update checks without touching the network."""
caplog.set_level(logging.WARNING)
with patch.object(UpdateCheckState, "should_check", return_value = False) as should_check_mock, \
patch.object(UpdateCheckState, "update_last_check") as update_last_check_mock, \
patch("requests.get") as mock_get:
checker = UpdateChecker(config)
checker.check_for_updates()
should_check_mock.assert_called_once()
mock_get.assert_not_called()
update_last_check_mock.assert_not_called()
assert all("Could not determine local version" not in message for message in caplog.messages)
def test_update_check_state_empty_file(self, state_file:Path) -> None:
"""Test that loading an empty state file returns a new state."""
state_file.touch() # Create empty file

81
tests/unit/test_web_scraping_mixin.py
View File

@@ -431,51 +431,70 @@ class TestSelectorTimeoutMessages:
class TestWebScrapingSessionManagement:
"""Test session management edge cases in WebScrapingMixin."""
def test_close_browser_session_cleans_up(self, mock_browser:AsyncMock) -> None:
"""Test that close_browser_session cleans up browser and page references and kills child processes."""
def test_close_browser_session_cleans_up_resources(self) -> None:
"""Ensure browser and page references are cleared and child processes are killed."""
scraper = WebScrapingMixin()
scraper.browser = MagicMock()
scraper.page = MagicMock()
scraper.browser._process_pid = 12345
scraper.browser._process_pid = 42
stop_mock = scraper.browser.stop = MagicMock()
# Patch psutil.Process and its children
scraper.page = MagicMock(spec = Page)
with patch("psutil.Process") as mock_proc:
mock_child = MagicMock()
mock_child.is_running.return_value = True
mock_proc.return_value.children.return_value = [mock_child]
scraper.close_browser_session()
# Browser stop should be called
mock_proc.assert_called_once_with(42)
stop_mock.assert_called_once()
# Child process kill should be called
mock_child.kill.assert_called_once()
# Browser and page references should be cleared
assert scraper.browser is None
assert scraper.page is None
def test_close_browser_session_double_close(self) -> None:
"""Test that calling close_browser_session twice does not raise and is idempotent."""
def test_close_browser_session_idempotent(self) -> None:
"""Repeated calls should leave the state clean without re-running cleanup logic."""
scraper = WebScrapingMixin()
scraper.browser = MagicMock()
scraper.page = MagicMock()
scraper.browser._process_pid = 12345
scraper.browser.stop = MagicMock()
scraper.browser._process_pid = 99
stop_mock = scraper.browser.stop = MagicMock()
scraper.page = MagicMock(spec = Page)
with patch("psutil.Process") as mock_proc:
mock_child = MagicMock()
mock_child.is_running.return_value = True
mock_proc.return_value.children.return_value = [mock_child]
mock_proc.return_value.children.return_value = []
scraper.close_browser_session()
# Second call should not raise
scraper.close_browser_session()
def test_close_browser_session_no_browser(self) -> None:
"""Test that close_browser_session is a no-op if browser is None."""
mock_proc.assert_called_once()
stop_mock.assert_called_once()
def test_close_browser_session_without_browser_skips_inspection(self) -> None:
"""When no browser exists, no process inspection should run and the page should stay untouched."""
scraper = WebScrapingMixin()
scraper.browser = None # type: ignore[unused-ignore,reportAttributeAccessIssue]
scraper.page = MagicMock()
# Should not raise
preserved_page = MagicMock(spec = Page)
scraper.page = preserved_page
with patch("psutil.Process") as mock_proc:
scraper.close_browser_session()
# Page should remain unchanged
assert scraper.page is not None
mock_proc.assert_not_called()
assert scraper.page is preserved_page
def test_close_browser_session_handles_missing_children(self) -> None:
"""Child-less browsers should still stop cleanly without raising."""
scraper = WebScrapingMixin()
scraper.browser = MagicMock()
scraper.browser._process_pid = 123
stop_mock = scraper.browser.stop = MagicMock()
scraper.page = MagicMock(spec = Page)
with patch("psutil.Process") as mock_proc:
mock_proc.return_value.children.return_value = []
scraper.close_browser_session()
mock_proc.assert_called_once()
stop_mock.assert_called_once()
def test_get_compatible_browser_raises_on_unknown_os(self) -> None:
"""Test get_compatible_browser raises AssertionError on unknown OS."""
@@ -494,22 +513,6 @@ class TestWebScrapingSessionManagement:
):
scraper.get_compatible_browser()
def test_close_browser_session_no_children(self) -> None:
"""Test that close_browser_session handles case when browser has no child processes."""
scraper = WebScrapingMixin()
scraper.browser = MagicMock()
scraper.page = MagicMock()
scraper.browser._process_pid = 12345
stop_mock = scraper.browser.stop = MagicMock()
# Mock Process to return no children
with patch("psutil.Process") as mock_proc:
mock_proc.return_value.children.return_value = []
scraper.close_browser_session()
stop_mock.assert_called_once()
assert scraper.browser is None
assert scraper.page is None
class TestWebScrolling:
"""Test scrolling helpers."""