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

feat: Add extend command to extend ads before expiry (#732)

Browse Source 
## ℹ️ Description

Add a manual "extend" command to extend listings shortly before they
expire. This keeps existing watchers/savers and does not count toward
the current 100 ads/month quota.

- Link to the related issue(s): Issue #664
- **Motivation**: Users need a way to extend ads before they expire
without republishing (which consumes quota).

## 📋 Changes Summary

### Implementation
- Add `extend` command case in `run()`
- Implement `extend_ads()` to filter and process eligible ads
- Implement `extend_ad()` for browser automation
- Add German translations for all user-facing messages

### Testing
- Tests cover: filtering logic, date parsing, browser automation, error
handling, edge cases

### Features
- Detects ads within the **8-day extension window** (kleinanzeigen.de
policy)
- Uses API `endDate` from `/m-meine-anzeigen-verwalten.json` for
eligibility
- Only extends active ads (`state == "active"`)
- Handles confirmation dialog (close dialog / skip paid bump-up)
- Updates `updated_on` in YAML after successful extension
- Supports `--ads` parameter to extend specific ad IDs

### Usage
```bash
kleinanzeigen-bot extend                  # Extend all eligible ads
kleinanzeigen-bot extend --ads=1,2,3      # Extend specific ads
```

### ⚙️ Type of Change
- [x] ✨ New feature (adds new functionality without breaking existing
usage)

## ✅ 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 updated documentation where necessary (help text in English
+ German).

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**
* Added an "extend" command to find ads nearing expiry (default 8-day
window) or target specific IDs, open a session, attempt extensions, and
record per-ad outcomes.

* **Documentation**
* Updated CLI/help (bilingual) and README to document the extend
command, options (--ads), default behavior, and expiry-window
limitations.

* **Tests**
* Added comprehensive unit tests for eligibility rules, date parsing
(including German format), edge cases, UI interaction flows, timing, and
error handling.

<sub>✏️ Tip: You can customize this high-level summary in your review
settings.</sub>

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Jens
2026-01-19 10:24:23 +01:00
committed by GitHub
parent a2473081e6
commit 6ef6aea3a8
4 changed files with 734 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
README.md
View File

@@ -31,6 +31,7 @@ For details on the new smoke test strategy and contributor guidance, see [TESTIN
- **Smart Republishing**: Automatically republish listings at configurable intervals to keep them at the top of search results
- **Bulk Management**: Update or delete multiple listings at once
- **Download Listings**: Download existing listings from your profile to local configuration files
- **Extend Listings**: Extend ads close to expiry to keep watchers/savers and preserve the monthly ad quota
- **Browser Automation**: Uses Chromium-based browsers (Chrome, Edge, Chromium) for reliable automation
- **Flexible Configuration**: Configure defaults once, override per listing as needed
@@ -198,6 +199,7 @@ Commands:
delete - deletes ads
update - updates published ads
download - downloads one or multiple ads
extend - extends active ads that expire soon (keeps watchers/savers and does not count towards the monthly ad quota)
update-check - checks for available updates
update-content-hash – recalculates each ad's content_hash based on the current ad_defaults;
use this after changing config.yaml/ad_defaults to avoid every ad being marked "changed" and republished
@@ -221,6 +223,11 @@ Options:
* all: downloads all ads from your profile
* new: downloads ads from your profile that are not locally saved yet
* <id(s)>: provide one or several ads by ID to download, like e.g. "--ads=1,2,3"
--ads=all|<id(s)> (extend) - specifies which ads to extend (DEFAULT: all)
Possible values:
* all: extend all eligible ads in your profile
* <id(s)>: provide one or several ads by ID to extend, like e.g. "--ads=1,2,3"
* Note: kleinanzeigen.de only allows extending ads within 8 days of expiry; ads outside this window are skipped.
--ads=changed|<id(s)> (update) - specifies which ads to update (DEFAULT: changed)
Possible values:
* changed: only update ads that have been modified since last publication

136
src/kleinanzeigen_bot/__init__.py
View File

@@ -4,6 +4,7 @@
import atexit, enum, json, os, re, signal, sys, textwrap # isort: skip
import getopt # pylint: disable=deprecated-module
import urllib.parse as urllib_parse
from datetime import datetime
from gettext import gettext as _
from pathlib import Path
from typing import Any, Final
@@ -309,6 +310,26 @@ class KleinanzeigenBot(WebScrapingMixin):
LOG.info("############################################")
LOG.info("DONE: No ads to delete found.")
LOG.info("############################################")
case "extend":
self.configure_file_logging()
self.load_config()
# Check for updates on startup
checker = UpdateChecker(self.config)
checker.check_for_updates()
# Default to all ads if no selector provided
if not re.compile(r"\d+[,\d+]*").search(self.ads_selector):
LOG.info(_("Extending all ads within 8-day window..."))
self.ads_selector = "all"
if ads := self.load_ads():
await self.create_browser_session()
await self.login()
await self.extend_ads(ads)
else:
LOG.info("############################################")
LOG.info("DONE: No ads found to extend.")
LOG.info("############################################")
case "download":
self.configure_file_logging()
# ad IDs depends on selector
@@ -346,6 +367,7 @@ class KleinanzeigenBot(WebScrapingMixin):
verify - Überprüft die Konfigurationsdateien
delete - Löscht Anzeigen
update - Aktualisiert bestehende Anzeigen
extend - Verlängert Anzeigen innerhalb des 8-Tage-Zeitfensters
download - Lädt eine oder mehrere Anzeigen herunter
update-check - Prüft auf verfügbare Updates
update-content-hash - Berechnet den content_hash aller Anzeigen anhand der aktuellen ad_defaults neu;
@@ -376,6 +398,9 @@ class KleinanzeigenBot(WebScrapingMixin):
Mögliche Werte:
* changed: Aktualisiert nur Anzeigen, die seit der letzten Veröffentlichung geändert wurden
* <id(s)>: Gibt eine oder mehrere Anzeigen-IDs zum Aktualisieren an, z. B. "--ads=1,2,3"
--ads=<id(s)> (extend) - Gibt an, welche Anzeigen verlängert werden sollen
Standardmäßig werden alle Anzeigen verlängert, die innerhalb von 8 Tagen ablaufen.
Mit dieser Option können Sie bestimmte Anzeigen-IDs angeben, z. B. "--ads=1,2,3"
--force - Alias für '--ads=all'
--keep-old - Verhindert das Löschen alter Anzeigen bei erneuter Veröffentlichung
--config=<PATH> - Pfad zur YAML- oder JSON-Konfigurationsdatei (STANDARD: ./config.yaml)
@@ -392,6 +417,7 @@ class KleinanzeigenBot(WebScrapingMixin):
verify - verifies the configuration files
delete - deletes ads
update - updates published ads
extend - extends ads within the 8-day window before expiry
download - downloads one or multiple ads
update-check - checks for available updates
update-content-hash – recalculates each ad's content_hash based on the current ad_defaults;
@@ -420,6 +446,9 @@ class KleinanzeigenBot(WebScrapingMixin):
Possible values:
* changed: only update ads that have been modified since last publication
* <id(s)>: provide one or several ads by ID to update, like e.g. "--ads=1,2,3"
--ads=<id(s)> (extend) - specifies which ads to extend
By default, extends all ads expiring within 8 days.
Use this option to specify ad IDs, e.g. "--ads=1,2,3"
--force - alias for '--ads=all'
--keep-old - don't delete old ads on republication
--config=<PATH> - path to the config YAML or JSON file (DEFAULT: ./config.yaml)
@@ -879,6 +908,113 @@ class KleinanzeigenBot(WebScrapingMixin):
ad_cfg.id = None
return True
async def extend_ads(self, ad_cfgs:list[tuple[str, Ad, dict[str, Any]]]) -> None:
"""Extends ads that are close to expiry."""
# Fetch currently published ads from API
published_ads = json.loads(
(await self.web_request(f"{self.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT"))["content"])["ads"]
# Filter ads that need extension
ads_to_extend = []
for (ad_file, ad_cfg, ad_cfg_orig) in ad_cfgs:
# Skip unpublished ads (no ID)
if not ad_cfg.id:
LOG.info(_(" -> SKIPPED: ad '%s' is not published yet"), ad_cfg.title)
continue
# Find ad in published list
published_ad = next((ad for ad in published_ads if ad["id"] == ad_cfg.id), None)
if not published_ad:
LOG.warning(_(" -> SKIPPED: ad '%s' (ID: %s) not found in published ads"), ad_cfg.title, ad_cfg.id)
continue
# Skip non-active ads
if published_ad.get("state") != "active":
LOG.info(_(" -> SKIPPED: ad '%s' is not active (state: %s)"), ad_cfg.title, published_ad.get("state"))
continue
# Check if ad is within 8-day extension window using API's endDate
end_date_str = published_ad.get("endDate")
if not end_date_str:
LOG.warning(_(" -> SKIPPED: ad '%s' has no endDate in API response"), ad_cfg.title)
continue
# Intentionally parsing naive datetime from kleinanzeigen API's German date format, timezone not relevant for date-only comparison
end_date = datetime.strptime(end_date_str, "%d.%m.%Y") # noqa: DTZ007
days_until_expiry = (end_date.date() - misc.now().date()).days
# Magic value 8 is kleinanzeigen.de's platform policy: extensions only possible within 8 days of expiry
if days_until_expiry <= 8: # noqa: PLR2004
LOG.info(_(" -> ad '%s' expires in %d days, will extend"), ad_cfg.title, days_until_expiry)
ads_to_extend.append((ad_file, ad_cfg, ad_cfg_orig, published_ad))
else:
LOG.info(_(" -> SKIPPED: ad '%s' expires in %d days (can only extend within 8 days)"),
ad_cfg.title, days_until_expiry)
if not ads_to_extend:
LOG.info(_("No ads need extension at this time."))
LOG.info("############################################")
LOG.info(_("DONE: No ads extended."))
LOG.info("############################################")
return
# Process extensions
success_count = 0
for idx, (ad_file, ad_cfg, ad_cfg_orig, _published_ad) in enumerate(ads_to_extend, start = 1):
LOG.info(_("Processing %s/%s: '%s' from [%s]..."), idx, len(ads_to_extend), ad_cfg.title, ad_file)
if await self.extend_ad(ad_file, ad_cfg, ad_cfg_orig):
success_count += 1
await self.web_sleep()
LOG.info("############################################")
LOG.info(_("DONE: Extended %s"), pluralize("ad", success_count))
LOG.info("############################################")
async def extend_ad(self, ad_file:str, ad_cfg:Ad, ad_cfg_orig:dict[str, Any]) -> bool:
"""Extends a single ad listing."""
LOG.info(_("Extending ad '%s' (ID: %s)..."), ad_cfg.title, ad_cfg.id)
try:
# Navigate to ad management page
await self.web_open(f"{self.root_url}/m-meine-anzeigen.html")
# Find and click "Verlängern" (extend) button for this ad
extend_button_xpath = f'//li[@data-adid="{ad_cfg.id}"]//button[contains(., "Verlängern")]'
try:
await self.web_click(By.XPATH, extend_button_xpath)
except TimeoutError:
LOG.error(_(" -> FAILED: Could not find extend button for ad ID %s"), ad_cfg.id)
return False
# Handle confirmation dialog
# After clicking "Verlängern", a dialog appears with:
# - Title: "Vielen Dank!"
# - Message: "Deine Anzeige ... wurde erfolgreich verlängert."
# - Paid bump-up option (skipped by closing dialog)
# Simply close the dialog with the X button (aria-label="Schließen")
try:
dialog_close_timeout = self._timeout("quick_dom")
await self.web_click(By.CSS_SELECTOR, 'button[aria-label="Schließen"]', timeout = dialog_close_timeout)
LOG.debug(" -> Closed confirmation dialog")
except TimeoutError:
LOG.warning(_(" -> No confirmation dialog found, extension may have completed directly"))
# Update metadata in YAML file
# Update updated_on to track when ad was extended
ad_cfg_orig["updated_on"] = misc.now().isoformat(timespec = "seconds")
dicts.save_dict(ad_file, ad_cfg_orig)
LOG.info(_(" -> SUCCESS: ad extended with ID %s"), ad_cfg.id)
return True
except TimeoutError as ex:
LOG.error(_(" -> FAILED: Timeout while extending ad '%s': %s"), ad_cfg.title, ex)
return False
except OSError as ex:
LOG.error(_(" -> FAILED: Could not persist extension for ad '%s': %s"), ad_cfg.title, ex)
return False
async def __check_publishing_result(self) -> bool:
# Check for success messages
return await self.web_check(By.ID, "checking-done", Is.DISPLAYED) or await self.web_check(By.ID, "not-completed", Is.DISPLAYED)

23
src/kleinanzeigen_bot/resources/translations.de.yaml
View File

@@ -91,6 +91,27 @@ kleinanzeigen_bot/__init__.py:
"Expected CSRF Token not found in HTML content!": "Erwartetes CSRF-Token wurde im HTML-Inhalt nicht gefunden!"
" -> deleting %s '%s'...": " -> lösche %s '%s'..."
extend_ads:
"No ads need extension at this time.": "Keine Anzeigen müssen derzeit verlängert werden."
"DONE: No ads extended.": "FERTIG: Keine Anzeigen verlängert."
"DONE: Extended %s": "FERTIG: %s verlängert"
"ad": "Anzeige"
" -> SKIPPED: ad '%s' is not published yet": " -> ÜBERSPRUNGEN: Anzeige '%s' ist noch nicht veröffentlicht"
" -> SKIPPED: ad '%s' (ID: %s) not found in published ads": " -> ÜBERSPRUNGEN: Anzeige '%s' (ID: %s) nicht gefunden"
" -> SKIPPED: ad '%s' is not active (state: %s)": " -> ÜBERSPRUNGEN: Anzeige '%s' ist nicht aktiv (Status: %s)"
" -> SKIPPED: ad '%s' has no endDate in API response": " -> ÜBERSPRUNGEN: Anzeige '%s' hat kein Ablaufdatum in API-Antwort"
" -> ad '%s' expires in %d days, will extend": " -> Anzeige '%s' läuft in %d Tagen ab, wird verlängert"
" -> SKIPPED: ad '%s' expires in %d days (can only extend within 8 days)": " -> ÜBERSPRUNGEN: Anzeige '%s' läuft in %d Tagen ab (Verlängern nur innerhalb von 8 Tagen möglich)"
"Processing %s/%s: '%s' from [%s]...": "Verarbeite %s/%s: '%s' aus [%s]..."
extend_ad:
"Extending ad '%s' (ID: %s)...": "Verlängere Anzeige '%s' (ID: %s)..."
" -> FAILED: Could not find extend button for ad ID %s": " -> FEHLER: 'Verlängern'-Button für Anzeigen-ID %s nicht gefunden"
" -> No confirmation dialog found, extension may have completed directly": " -> Kein Bestätigungsdialog gefunden"
" -> SUCCESS: ad extended with ID %s": " -> ERFOLG: Anzeige mit ID %s verlängert"
" -> FAILED: Timeout while extending ad '%s': %s": " -> FEHLER: Zeitüberschreitung beim Verlängern der Anzeige '%s': %s"
" -> FAILED: Could not persist extension for ad '%s': %s": " -> FEHLER: Verlängerung der Anzeige '%s' konnte nicht gespeichert werden: %s"
publish_ads:
"Processing %s/%s: '%s' from [%s]...": "Verarbeite %s/%s: '%s' von [%s]..."
"Skipping because ad is reserved": "Überspringen, da Anzeige reserviert ist"
@@ -183,6 +204,8 @@ kleinanzeigen_bot/__init__.py:
"DONE: No new/outdated ads found.": "FERTIG: Keine neuen/veralteten Anzeigen gefunden."
"DONE: No ads to delete found.": "FERTIG: Keine zu löschenden Anzeigen gefunden."
"DONE: No changed ads found.": "FERTIG: Keine geänderten Anzeigen gefunden."
"Extending all ads within 8-day window...": "Verlängere alle Anzeigen innerhalb des 8-Tage-Zeitfensters..."
"DONE: No ads found to extend.": "FERTIG: Keine Anzeigen zum Verlängern gefunden."
"You provided no ads selector. Defaulting to \"new\".": "Es wurden keine Anzeigen-Selektor angegeben. Es wird \"new\" verwendet."
"You provided no ads selector. Defaulting to \"changed\".": "Es wurden keine Anzeigen-Selektor angegeben. Es wird \"changed\" verwendet."
"Unknown command: %s": "Unbekannter Befehl: %s"

568
tests/unit/test_extend_command.py Normal file
View File

@@ -0,0 +1,568 @@
# 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 json # isort: skip
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from kleinanzeigen_bot import KleinanzeigenBot, misc
from kleinanzeigen_bot.model.ad_model import Ad
from kleinanzeigen_bot.utils import dicts
@pytest.fixture
def base_ad_config_with_id() -> dict[str, Any]:
"""Provide a base ad configuration with an ID for extend tests."""
return {
"id": 12345,
"title": "Test Ad Title",
"description": "Test Description",
"type": "OFFER",
"price_type": "FIXED",
"price": 100,
"shipping_type": "SHIPPING",
"shipping_options": [],
"category": "160",
"special_attributes": {},
"sell_directly": False,
"images": [],
"active": True,
"republication_interval": 7,
"created_on": "2024-12-07T10:00:00",
"updated_on": "2024-12-10T15:20:00",
"contact": {
"name": "Test User",
"zipcode": "12345",
"location": "Test City",
"street": "",
"phone": ""
}
}
class TestExtendCommand:
"""Tests for the extend command functionality."""
@pytest.mark.asyncio
async def test_run_extend_command_no_ads(self, test_bot:KleinanzeigenBot) -> None:
"""Test running extend command with no ads."""
with patch.object(test_bot, "load_config"), \
patch.object(test_bot, "load_ads", return_value = []), \
patch("kleinanzeigen_bot.UpdateChecker"):
await test_bot.run(["script.py", "extend"])
assert test_bot.command == "extend"
assert test_bot.ads_selector == "all"
@pytest.mark.asyncio
async def test_run_extend_command_with_specific_ids(self, test_bot:KleinanzeigenBot) -> None:
"""Test running extend command with specific ad IDs."""
with patch.object(test_bot, "load_config"), \
patch.object(test_bot, "load_ads", return_value = []), \
patch.object(test_bot, "create_browser_session", new_callable = AsyncMock), \
patch.object(test_bot, "login", new_callable = AsyncMock), \
patch("kleinanzeigen_bot.UpdateChecker"):
await test_bot.run(["script.py", "extend", "--ads=12345,67890"])
assert test_bot.command == "extend"
assert test_bot.ads_selector == "12345,67890"
class TestExtendAdsMethod:
"""Tests for the extend_ads() method."""
@pytest.mark.asyncio
async def test_extend_ads_skips_unpublished_ad(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads skips ads without an ID (unpublished)."""
# Create ad without ID
ad_config = base_ad_config_with_id.copy()
ad_config["id"] = None
ad_cfg = Ad.model_validate(ad_config)
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock):
mock_request.return_value = {"content": '{"ads": []}'}
await test_bot.extend_ads([("test.yaml", ad_cfg, ad_config)])
# Verify no extension was attempted
mock_request.assert_called_once() # Only the API call to get published ads
@pytest.mark.asyncio
async def test_extend_ads_skips_ad_not_in_published_list(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads skips ads not found in the published ads API response."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock):
# Return empty published ads list
mock_request.return_value = {"content": '{"ads": []}'}
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify no extension was attempted
mock_request.assert_called_once()
@pytest.mark.asyncio
async def test_extend_ads_skips_inactive_ad(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads skips ads with state != 'active'."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "paused", # Not active
"endDate": "05.02.2026"
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was not called
mock_extend_ad.assert_not_called()
@pytest.mark.asyncio
async def test_extend_ads_skips_ad_without_enddate(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads skips ads without endDate in API response."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active"
# No endDate field
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was not called
mock_extend_ad.assert_not_called()
@pytest.mark.asyncio
async def test_extend_ads_skips_ad_outside_window(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads skips ads expiring more than 8 days in the future."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Set end date to 30 days from now (outside 8-day window)
future_date = misc.now() + timedelta(days = 30)
end_date_str = future_date.strftime("%d.%m.%Y")
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active",
"endDate": end_date_str
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was not called
mock_extend_ad.assert_not_called()
@pytest.mark.asyncio
async def test_extend_ads_extends_ad_within_window(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads extends ads within the 8-day window."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Set end date to 5 days from now (within 8-day window)
future_date = misc.now() + timedelta(days = 5)
end_date_str = future_date.strftime("%d.%m.%Y")
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active",
"endDate": end_date_str
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
mock_extend_ad.return_value = True
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was called
mock_extend_ad.assert_called_once()
@pytest.mark.asyncio
async def test_extend_ads_no_eligible_ads(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test extend_ads when no ads are eligible for extension."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Set end date to 30 days from now (outside window)
future_date = misc.now() + timedelta(days = 30)
end_date_str = future_date.strftime("%d.%m.%Y")
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active",
"endDate": end_date_str
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was not called
mock_extend_ad.assert_not_called()
@pytest.mark.asyncio
async def test_extend_ads_handles_multiple_ads(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads processes multiple ads correctly."""
ad_cfg1 = Ad.model_validate(base_ad_config_with_id)
# Create second ad
ad_config2 = base_ad_config_with_id.copy()
ad_config2["id"] = 67890
ad_config2["title"] = "Second Test Ad"
ad_cfg2 = Ad.model_validate(ad_config2)
# Set end dates - one within window, one outside
within_window = misc.now() + timedelta(days = 5)
outside_window = misc.now() + timedelta(days = 30)
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active",
"endDate": within_window.strftime("%d.%m.%Y")
},
{
"id": 67890,
"title": "Second Test Ad",
"state": "active",
"endDate": outside_window.strftime("%d.%m.%Y")
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
mock_extend_ad.return_value = True
await test_bot.extend_ads([
("test1.yaml", ad_cfg1, base_ad_config_with_id),
("test2.yaml", ad_cfg2, ad_config2)
])
# Verify extend_ad was called only once (for the ad within window)
assert mock_extend_ad.call_count == 1
class TestExtendAdMethod:
"""Tests for the extend_ad() method."""
@pytest.mark.asyncio
async def test_extend_ad_success(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any],
tmp_path:Path
) -> None:
"""Test successful ad extension."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Create temporary YAML file
ad_file = tmp_path / "test_ad.yaml"
dicts.save_dict(str(ad_file), base_ad_config_with_id)
with patch.object(test_bot, "web_open", new_callable = AsyncMock), \
patch.object(test_bot, "web_click", new_callable = AsyncMock), \
patch("kleinanzeigen_bot.misc.now") as mock_now:
# Test mock datetime - timezone not relevant for timestamp formatting test
mock_now.return_value = datetime(2025, 1, 28, 14, 30, 0) # noqa: DTZ001
result = await test_bot.extend_ad(str(ad_file), ad_cfg, base_ad_config_with_id)
assert result is True
# Verify updated_on was updated in the YAML file
updated_config = dicts.load_dict(str(ad_file))
assert updated_config["updated_on"] == "2025-01-28T14:30:00"
@pytest.mark.asyncio
async def test_extend_ad_button_not_found(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any],
tmp_path:Path
) -> None:
"""Test extend_ad when the Verlängern button is not found."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Create temporary YAML file
ad_file = tmp_path / "test_ad.yaml"
dicts.save_dict(str(ad_file), base_ad_config_with_id)
with patch.object(test_bot, "web_open", new_callable = AsyncMock), \
patch.object(test_bot, "web_click", new_callable = AsyncMock) as mock_click:
# Simulate button not found by raising TimeoutError
mock_click.side_effect = TimeoutError("Button not found")
result = await test_bot.extend_ad(str(ad_file), ad_cfg, base_ad_config_with_id)
assert result is False
@pytest.mark.asyncio
async def test_extend_ad_dialog_timeout(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any],
tmp_path:Path
) -> None:
"""Test extend_ad when the confirmation dialog times out (no dialog appears)."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Create temporary YAML file
ad_file = tmp_path / "test_ad.yaml"
dicts.save_dict(str(ad_file), base_ad_config_with_id)
with patch.object(test_bot, "web_open", new_callable = AsyncMock), \
patch.object(test_bot, "web_click", new_callable = AsyncMock) as mock_click, \
patch("kleinanzeigen_bot.misc.now") as mock_now:
# Test mock datetime - timezone not relevant for timestamp formatting test
mock_now.return_value = datetime(2025, 1, 28, 14, 30, 0) # noqa: DTZ001
# First click (Verlängern button) succeeds, second click (dialog close) times out
mock_click.side_effect = [None, TimeoutError("Dialog not found")]
result = await test_bot.extend_ad(str(ad_file), ad_cfg, base_ad_config_with_id)
# Should still succeed (dialog might not appear)
assert result is True
@pytest.mark.asyncio
async def test_extend_ad_exception_handling(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any],
tmp_path:Path
) -> None:
"""Test extend_ad propagates unexpected exceptions."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Create temporary YAML file
ad_file = tmp_path / "test_ad.yaml"
dicts.save_dict(str(ad_file), base_ad_config_with_id)
with patch.object(test_bot, "web_open", new_callable = AsyncMock) as mock_open:
# Simulate unexpected exception
mock_open.side_effect = Exception("Unexpected error")
with pytest.raises(Exception, match = "Unexpected error"):
await test_bot.extend_ad(str(ad_file), ad_cfg, base_ad_config_with_id)
@pytest.mark.asyncio
async def test_extend_ad_updates_yaml_file(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any],
tmp_path:Path
) -> None:
"""Test that extend_ad correctly updates the YAML file with new timestamp."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Create temporary YAML file
ad_file = tmp_path / "test_ad.yaml"
original_updated_on = base_ad_config_with_id["updated_on"]
dicts.save_dict(str(ad_file), base_ad_config_with_id)
with patch.object(test_bot, "web_open", new_callable = AsyncMock), \
patch.object(test_bot, "web_click", new_callable = AsyncMock), \
patch("kleinanzeigen_bot.misc.now") as mock_now:
# Test mock datetime - timezone not relevant for timestamp formatting test
mock_now.return_value = datetime(2025, 1, 28, 14, 30, 0) # noqa: DTZ001
await test_bot.extend_ad(str(ad_file), ad_cfg, base_ad_config_with_id)
# Load the updated file and verify the timestamp changed
updated_config = dicts.load_dict(str(ad_file))
assert updated_config["updated_on"] != original_updated_on
assert updated_config["updated_on"] == "2025-01-28T14:30:00"
class TestExtendEdgeCases:
"""Tests for edge cases and boundary conditions."""
@pytest.mark.asyncio
async def test_extend_ads_exactly_8_days(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that ads expiring exactly in 8 days are eligible for extension."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Set end date to exactly 8 days from now (boundary case)
future_date = misc.now() + timedelta(days = 8)
end_date_str = future_date.strftime("%d.%m.%Y")
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active",
"endDate": end_date_str
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
mock_extend_ad.return_value = True
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was called (8 days is within the window)
mock_extend_ad.assert_called_once()
@pytest.mark.asyncio
async def test_extend_ads_exactly_9_days(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that ads expiring in exactly 9 days are not eligible for extension."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Set end date to exactly 9 days from now (just outside window)
future_date = misc.now() + timedelta(days = 9)
end_date_str = future_date.strftime("%d.%m.%Y")
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active",
"endDate": end_date_str
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad:
mock_request.return_value = {"content": json.dumps(published_ads_json)}
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was not called (9 days is outside the window)
mock_extend_ad.assert_not_called()
@pytest.mark.asyncio
async def test_extend_ads_date_parsing_german_format(
self,
test_bot:KleinanzeigenBot,
base_ad_config_with_id:dict[str, Any]
) -> None:
"""Test that extend_ads correctly parses German date format (DD.MM.YYYY)."""
ad_cfg = Ad.model_validate(base_ad_config_with_id)
# Use a specific German date format
published_ads_json = {
"ads": [
{
"id": 12345,
"title": "Test Ad Title",
"state": "active",
"endDate": "05.02.2026" # German format: DD.MM.YYYY
}
]
}
with patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad, \
patch("kleinanzeigen_bot.misc.now") as mock_now:
# Mock now() to return a date where 05.02.2026 would be within 8 days
# Test mock datetime - timezone not relevant for date comparison test
mock_now.return_value = datetime(2026, 1, 28) # noqa: DTZ001
mock_request.return_value = {"content": json.dumps(published_ads_json)}
mock_extend_ad.return_value = True
await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)])
# Verify extend_ad was called (date was parsed correctly)
mock_extend_ad.assert_called_once()