diff --git a/README.md b/README.md index 2a49720..7ce862c 100644 --- a/README.md +++ b/README.md @@ -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 * : provide one or several ads by ID to download, like e.g. "--ads=1,2,3" + --ads=all| (extend) - specifies which ads to extend (DEFAULT: all) + Possible values: + * all: extend all eligible ads in your profile + * : 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| (update) - specifies which ads to update (DEFAULT: changed) Possible values: * changed: only update ads that have been modified since last publication diff --git a/src/kleinanzeigen_bot/__init__.py b/src/kleinanzeigen_bot/__init__.py index 1819722..52758cf 100644 --- a/src/kleinanzeigen_bot/__init__.py +++ b/src/kleinanzeigen_bot/__init__.py @@ -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 * : Gibt eine oder mehrere Anzeigen-IDs zum Aktualisieren an, z. B. "--ads=1,2,3" + --ads= (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= - 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 * : provide one or several ads by ID to update, like e.g. "--ads=1,2,3" + --ads= (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 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) diff --git a/src/kleinanzeigen_bot/resources/translations.de.yaml b/src/kleinanzeigen_bot/resources/translations.de.yaml index 8497808..6358085 100644 --- a/src/kleinanzeigen_bot/resources/translations.de.yaml +++ b/src/kleinanzeigen_bot/resources/translations.de.yaml @@ -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" diff --git a/tests/unit/test_extend_command.py b/tests/unit/test_extend_command.py new file mode 100644 index 0000000..6e4e1be --- /dev/null +++ b/tests/unit/test_extend_command.py @@ -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()