mirror of
https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git
synced 2026-07-09 12:41:05 +02:00
7803872422
## ℹ️ Description > **Why one PR?** The four extractions are mechanically identical: extract a method body into a module function, replace `self.*` with explicit parameters, update call sites in `run()`. Splitting into 4 sequential PRs would produce repetitive diffs against the same `__init__.py`, compound rebase friction, and add no independent review signal — each PR would be reviewed as "same pattern, different function." Combined, the mechanical nature is visible at a glance and the 17 files are mostly new modules + focused test files. Reduce KleinanzeigenBot.__init__.py from 2865 to 2391 lines by extracting four browser workflow modules with explicit dependencies. ## 📋 Changes Summary - New `published_ads.py` — `fetch_published_ads()` and `PublishedAdsFetchIncompleteError`, shared by all workflows - New `delete_flow.py` — `delete_ads()` orchestrator and `delete_ad()` single-ad deletion - New `extend_flow.py` — `extend_ads()` orchestrator and `_extend_ad()` single-ad extension - New `download_flow.py` — `download_ads()`, `resolve_download_dir()`, and `_download_ad_with_resolved_state()` - Promote `_timeout` and `_navigate_paginated_ad_overview` to public in WebScrapingMixin - Restore `PublishedAdsFetchIncompleteError` base class to KleinanzeigenBotError - New focused test files: `test_published_ads.py` (17), `test_delete_flow.py` (13), `test_extend_flow.py` (17), `test_download_flow.py` (16) - Removed: `test_json_pagination.py`, `test_extend_command.py` - Flow modules receive explicit dependencies (`web`, `root_url`, `config`, `workspace`, `load_ads_func`) — no `do_work(bot)` signatures - Temporary `_fetch_published_ads` delegator kept on bot (still used by publish/update) ### ⚙️ Type of Change - [ ] 🐞 Bug fix (non-breaking change which fixes an issue) - [x] ✨ 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. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Documentation** * Clarified timeout configuration guidance and examples (use public timeout API and when to use effective timeout). * **New Features** * Added workflows to delete, download (all/new/IDs) and extend ads; added API-backed published-ads retrieval and safe ID matching. * **Refactor** * Command routing moved to workflow modules; unified timeout and pagination helpers for consistent UI/navigation behavior. * **Tests** * Expanded/updated unit tests covering deletion, download, extension, pagination, timeout and related edge cases. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
167 lines
7.6 KiB
Python
167 lines
7.6 KiB
Python
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
|
|
"""Tests for the navigate_paginated_ad_overview helper method."""
|
|
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from kleinanzeigen_bot.utils.web_scraping_mixin import By, Element, WebScrapingMixin
|
|
|
|
|
|
class TestNavigatePaginatedAdOverview:
|
|
"""Tests for navigate_paginated_ad_overview method."""
|
|
|
|
@staticmethod
|
|
async def _single_page_find_side_effect(selector_type:By, selector_value:str, **kwargs:Any) -> Element: # noqa: ARG004
|
|
if selector_type == By.ID and selector_value == "my-manageitems-adlist":
|
|
return MagicMock()
|
|
raise TimeoutError("Unexpected find")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_page_action_succeeds(self) -> None:
|
|
"""Test pagination on single page where action succeeds."""
|
|
mixin = WebScrapingMixin()
|
|
|
|
# Mock callback that succeeds
|
|
callback = AsyncMock(return_value = True)
|
|
|
|
with (
|
|
patch.object(mixin, "web_open", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_sleep", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_find", new_callable = AsyncMock, side_effect = self._single_page_find_side_effect),
|
|
patch.object(mixin, "web_find_all", new_callable = AsyncMock, side_effect = TimeoutError("No pagination")),
|
|
patch.object(mixin, "web_scroll_page_down", new_callable = AsyncMock),
|
|
patch.object(mixin, "timeout", return_value = 10),
|
|
):
|
|
result = await mixin.navigate_paginated_ad_overview(callback)
|
|
|
|
assert result is True
|
|
callback.assert_awaited_once_with(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_page_action_returns_false(self) -> None:
|
|
"""Test pagination on single page where action returns False."""
|
|
mixin = WebScrapingMixin()
|
|
|
|
# Mock callback that returns False (doesn't find what it's looking for)
|
|
callback = AsyncMock(return_value = False)
|
|
|
|
with (
|
|
patch.object(mixin, "web_open", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_sleep", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_find", new_callable = AsyncMock, side_effect = self._single_page_find_side_effect),
|
|
patch.object(mixin, "web_find_all", new_callable = AsyncMock, side_effect = TimeoutError("No pagination")),
|
|
patch.object(mixin, "web_scroll_page_down", new_callable = AsyncMock),
|
|
patch.object(mixin, "timeout", return_value = 10),
|
|
):
|
|
result = await mixin.navigate_paginated_ad_overview(callback)
|
|
|
|
assert result is False
|
|
callback.assert_awaited_once_with(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_page_action_succeeds_on_page_2(self) -> None:
|
|
"""Test pagination across multiple pages where action succeeds on page 2."""
|
|
mixin = WebScrapingMixin()
|
|
|
|
# Mock callback that returns False on page 1, True on page 2
|
|
callback_results = [False, True]
|
|
callback = AsyncMock(side_effect = callback_results)
|
|
|
|
next_button_enabled = MagicMock()
|
|
next_button_enabled.attrs = {} # No "disabled" attribute = enabled
|
|
next_button_enabled.click = AsyncMock()
|
|
|
|
async def mock_find_all_side_effect(selector_type:By, selector_value:str, **kwargs:Any) -> list[Element]: # noqa: ARG004
|
|
if selector_type == By.CSS_SELECTOR and selector_value == 'button[aria-label="Nächste"]':
|
|
return [next_button_enabled]
|
|
return []
|
|
|
|
with (
|
|
patch.object(mixin, "web_open", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_sleep", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_find", new_callable = AsyncMock, side_effect = self._single_page_find_side_effect),
|
|
patch.object(mixin, "web_find_all", new_callable = AsyncMock, side_effect = mock_find_all_side_effect),
|
|
patch.object(mixin, "web_scroll_page_down", new_callable = AsyncMock),
|
|
patch.object(mixin, "timeout", return_value = 10),
|
|
):
|
|
result = await mixin.navigate_paginated_ad_overview(callback)
|
|
|
|
assert result is True
|
|
assert callback.await_count == 2
|
|
next_button_enabled.click.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_open_raises_timeout(self) -> None:
|
|
"""Test that TimeoutError on web_open is caught and returns False."""
|
|
mixin = WebScrapingMixin()
|
|
|
|
callback = AsyncMock()
|
|
|
|
with patch.object(mixin, "web_open", new_callable = AsyncMock, side_effect = TimeoutError("Page load timeout")):
|
|
result = await mixin.navigate_paginated_ad_overview(callback)
|
|
|
|
assert result is False
|
|
callback.assert_not_awaited() # Callback should not be called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ad_list_container_not_found(self) -> None:
|
|
"""Test that missing ad list container returns False."""
|
|
mixin = WebScrapingMixin()
|
|
|
|
callback = AsyncMock()
|
|
|
|
with (
|
|
patch.object(mixin, "web_open", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_sleep", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_find", new_callable = AsyncMock, side_effect = TimeoutError("Container not found")),
|
|
):
|
|
result = await mixin.navigate_paginated_ad_overview(callback)
|
|
|
|
assert result is False
|
|
callback.assert_not_awaited()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_web_scroll_timeout_continues(self) -> None:
|
|
"""Test that TimeoutError on web_scroll_page_down is non-fatal and pagination continues."""
|
|
mixin = WebScrapingMixin()
|
|
|
|
callback = AsyncMock(return_value = True)
|
|
|
|
with (
|
|
patch.object(mixin, "web_open", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_sleep", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_find", new_callable = AsyncMock, side_effect = self._single_page_find_side_effect),
|
|
patch.object(mixin, "web_find_all", new_callable = AsyncMock, side_effect = TimeoutError("No pagination")),
|
|
patch.object(mixin, "web_scroll_page_down", new_callable = AsyncMock, side_effect = TimeoutError("Scroll timeout")),
|
|
patch.object(mixin, "timeout", return_value = 10),
|
|
):
|
|
result = await mixin.navigate_paginated_ad_overview(callback)
|
|
|
|
# Should continue and call callback despite scroll timeout
|
|
assert result is True
|
|
callback.assert_awaited_once_with(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_page_action_raises_timeout(self) -> None:
|
|
"""Test that TimeoutError from page_action is caught and returns False."""
|
|
mixin = WebScrapingMixin()
|
|
|
|
callback = AsyncMock(side_effect = TimeoutError("Action timeout"))
|
|
|
|
with (
|
|
patch.object(mixin, "web_open", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_sleep", new_callable = AsyncMock),
|
|
patch.object(mixin, "web_find", new_callable = AsyncMock, side_effect = self._single_page_find_side_effect),
|
|
patch.object(mixin, "web_find_all", new_callable = AsyncMock, side_effect = TimeoutError("No pagination")),
|
|
patch.object(mixin, "web_scroll_page_down", new_callable = AsyncMock),
|
|
patch.object(mixin, "timeout", return_value = 10),
|
|
):
|
|
result = await mixin.navigate_paginated_ad_overview(callback)
|
|
|
|
assert result is False
|
|
callback.assert_awaited_once_with(1)
|