mirror of
https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git
synced 2026-03-12 10:31:50 +01:00
96f465d5bc
## ℹ️ Description *Provide a concise summary of the changes introduced in this pull request.* - Link to the related issue(s): Closes #789 (completes the fix started in #793) - **Motivation**: Fix JSON API pagination for accounts with >25 ads. Aligns pagination logic with weidi’s approach (starts at page 1), while hardening error handling and tests. Based on https://github.com/weidi/kleinanzeigen-bot/pull/1. ## 📋 Changes Summary - Added pagination helper to fetch all published ads and use it in delete/extend/publish/update flows - Added robust handling for malformed JSON payloads and unexpected ads types (with translated warnings) - Improved sell_directly extraction with pagination, bounds checks, and shared coercion helper - Added/updated tests for pagination and edge cases; updated assertions to pytest.fail style ### ⚙️ Type of Change Select the type(s) of change(s) included in this pull request: - [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 Before requesting a review, confirm the following: - [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:cov:unified`). - [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 * **New Features** * Reliable multi-page fetching for published ads and buy-now eligibility checks. * **Bug Fixes** * Safer pagination with per-page JSON handling, limits and improved termination diagnostics; ensures pageNum is used when needed. * **Tests** * New comprehensive pagination tests and updates to existing tests to reflect multi-page behavior. * **Chores** * Added a utility to safely coerce page numbers; minor utility signature cleanup. <sub>✏️ Tip: You can customize this high-level summary in your review settings.</sub> <!-- end of auto-generated comment: release notes by coderabbit.ai -->
232 lines
11 KiB
Python
232 lines
11 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 JSON API pagination helper methods."""
|
|
|
|
import json
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from kleinanzeigen_bot import KleinanzeigenBot
|
|
from kleinanzeigen_bot.utils import misc
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestJSONPagination:
|
|
"""Tests for _coerce_page_number and _fetch_published_ads methods."""
|
|
|
|
@pytest.fixture
|
|
def bot(self) -> KleinanzeigenBot:
|
|
return KleinanzeigenBot()
|
|
|
|
def test_coerce_page_number_with_valid_int(self) -> None:
|
|
"""Test that valid integers are returned as-is."""
|
|
result = misc.coerce_page_number(1)
|
|
if result != 1:
|
|
pytest.fail(f"_coerce_page_number(1) expected 1, got {result}")
|
|
|
|
result = misc.coerce_page_number(0)
|
|
if result != 0:
|
|
pytest.fail(f"_coerce_page_number(0) expected 0, got {result}")
|
|
|
|
result = misc.coerce_page_number(42)
|
|
if result != 42:
|
|
pytest.fail(f"_coerce_page_number(42) expected 42, got {result}")
|
|
|
|
def test_coerce_page_number_with_string_int(self) -> None:
|
|
"""Test that string integers are converted to int."""
|
|
result = misc.coerce_page_number("1")
|
|
if result != 1:
|
|
pytest.fail(f"_coerce_page_number('1') expected 1, got {result}")
|
|
|
|
result = misc.coerce_page_number("0")
|
|
if result != 0:
|
|
pytest.fail(f"_coerce_page_number('0') expected 0, got {result}")
|
|
|
|
result = misc.coerce_page_number("42")
|
|
if result != 42:
|
|
pytest.fail(f"_coerce_page_number('42') expected 42, got {result}")
|
|
|
|
def test_coerce_page_number_with_none(self) -> None:
|
|
"""Test that None returns None."""
|
|
result = misc.coerce_page_number(None)
|
|
if result is not None:
|
|
pytest.fail(f"_coerce_page_number(None) expected None, got {result}")
|
|
|
|
def test_coerce_page_number_with_invalid_types(self) -> None:
|
|
"""Test that invalid types return None."""
|
|
result = misc.coerce_page_number("invalid")
|
|
if result is not None:
|
|
pytest.fail(f'_coerce_page_number("invalid") expected None, got {result}')
|
|
|
|
result = misc.coerce_page_number("")
|
|
if result is not None:
|
|
pytest.fail(f'_coerce_page_number("") expected None, got {result}')
|
|
|
|
result = misc.coerce_page_number([])
|
|
if result is not None:
|
|
pytest.fail(f"_coerce_page_number([]) expected None, got {result}")
|
|
|
|
result = misc.coerce_page_number({})
|
|
if result is not None:
|
|
pytest.fail(f"_coerce_page_number({{}}) expected None, got {result}")
|
|
|
|
result = misc.coerce_page_number(3.14)
|
|
if result is not None:
|
|
pytest.fail(f"_coerce_page_number(3.14) expected None, got {result}")
|
|
|
|
def test_coerce_page_number_with_whole_number_float(self) -> None:
|
|
"""Test that whole-number floats are accepted and converted to int."""
|
|
result = misc.coerce_page_number(2.0)
|
|
if result != 2:
|
|
pytest.fail(f"_coerce_page_number(2.0) expected 2, got {result}")
|
|
|
|
result = misc.coerce_page_number(0.0)
|
|
if result != 0:
|
|
pytest.fail(f"_coerce_page_number(0.0) expected 0, got {result}")
|
|
|
|
result = misc.coerce_page_number(42.0)
|
|
if result != 42:
|
|
pytest.fail(f"_coerce_page_number(42.0) expected 42, got {result}")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_single_page_no_paging(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test fetching ads from single page with no paging info."""
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.return_value = {"content": '{"ads": [{"id": 1, "title": "Ad 1"}, {"id": 2, "title": "Ad 2"}]}'}
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
if len(result) != 2:
|
|
pytest.fail(f"Expected 2 results, got {len(result)}")
|
|
if result[0]["id"] != 1:
|
|
pytest.fail(f"Expected result[0]['id'] == 1, got {result[0]['id']}")
|
|
if result[1]["id"] != 2:
|
|
pytest.fail(f"Expected result[1]['id'] == 2, got {result[1]['id']}")
|
|
mock_request.assert_awaited_once_with(f"{bot.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT&pageNum=1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_single_page_with_paging(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test fetching ads from single page with paging info showing 1/1."""
|
|
response_data = {"ads": [{"id": 1, "title": "Ad 1"}], "paging": {"pageNum": 1, "last": 1}}
|
|
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.return_value = {"content": json.dumps(response_data)}
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
if len(result) != 1:
|
|
pytest.fail(f"Expected 1 ad, got {len(result)}")
|
|
if result[0].get("id") != 1:
|
|
pytest.fail(f"Expected ad id 1, got {result[0].get('id')}")
|
|
mock_request.assert_awaited_once_with(f"{bot.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT&pageNum=1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_multi_page(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test fetching ads from multiple pages (3 pages, 2 ads each)."""
|
|
page1_data = {"ads": [{"id": 1}, {"id": 2}], "paging": {"pageNum": 1, "last": 3, "next": 2}}
|
|
page2_data = {"ads": [{"id": 3}, {"id": 4}], "paging": {"pageNum": 2, "last": 3, "next": 3}}
|
|
page3_data = {"ads": [{"id": 5}, {"id": 6}], "paging": {"pageNum": 3, "last": 3}}
|
|
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.side_effect = [
|
|
{"content": json.dumps(page1_data)},
|
|
{"content": json.dumps(page2_data)},
|
|
{"content": json.dumps(page3_data)},
|
|
]
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
if len(result) != 6:
|
|
pytest.fail(f"Expected 6 ads but got {len(result)}")
|
|
if [ad["id"] for ad in result] != [1, 2, 3, 4, 5, 6]:
|
|
pytest.fail(f"Expected ids [1, 2, 3, 4, 5, 6] but got {[ad['id'] for ad in result]}")
|
|
if mock_request.call_count != 3:
|
|
pytest.fail(f"Expected 3 web_request calls but got {mock_request.call_count}")
|
|
mock_request.assert_any_await(f"{bot.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT&pageNum=1")
|
|
mock_request.assert_any_await(f"{bot.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT&pageNum=2")
|
|
mock_request.assert_any_await(f"{bot.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT&pageNum=3")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_empty_list(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test handling of empty ads list."""
|
|
response_data = {"ads": [], "paging": {"pageNum": 1, "last": 1}}
|
|
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.return_value = {"content": json.dumps(response_data)}
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
if not isinstance(result, list):
|
|
pytest.fail(f"expected result to be list, got {type(result).__name__}")
|
|
if len(result) != 0:
|
|
pytest.fail(f"expected empty list from _fetch_published_ads, got {len(result)} items")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_invalid_json(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test handling of invalid JSON response."""
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.return_value = {"content": "invalid json"}
|
|
|
|
result = await bot._fetch_published_ads()
|
|
if result != []:
|
|
pytest.fail(f"Expected empty list on invalid JSON, got {result}")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_missing_paging_dict(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test handling of missing paging dict."""
|
|
response_data = {"ads": [{"id": 1}, {"id": 2}]}
|
|
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.return_value = {"content": json.dumps(response_data)}
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
if len(result) != 2:
|
|
pytest.fail(f"expected 2 ads, got {len(result)}")
|
|
mock_request.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_non_integer_paging_values(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test handling of non-integer paging values."""
|
|
response_data = {"ads": [{"id": 1}], "paging": {"pageNum": "invalid", "last": "also-invalid"}}
|
|
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.return_value = {"content": json.dumps(response_data)}
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
# Should return ads from first page and stop due to invalid paging
|
|
if len(result) != 1:
|
|
pytest.fail(f"Expected 1 ad, got {len(result)}")
|
|
if result[0].get("id") != 1:
|
|
pytest.fail(f"Expected ad id 1, got {result[0].get('id')}")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_non_list_ads(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test handling of non-list ads field."""
|
|
response_data = {"ads": "not a list", "paging": {"pageNum": 1, "last": 1}}
|
|
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.return_value = {"content": json.dumps(response_data)}
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
# Should return empty list when ads is not a list
|
|
if not isinstance(result, list):
|
|
pytest.fail(f"expected empty list when 'ads' is not a list, got: {result}")
|
|
if len(result) != 0:
|
|
pytest.fail(f"expected empty list when 'ads' is not a list, got: {result}")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_published_ads_timeout(self, bot:KleinanzeigenBot) -> None:
|
|
"""Test handling of timeout during pagination."""
|
|
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
|
|
mock_request.side_effect = TimeoutError("timeout")
|
|
|
|
result = await bot._fetch_published_ads()
|
|
|
|
if result != []:
|
|
pytest.fail(f"Expected empty list on timeout, got {result}")
|