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

refactor: extract ad loading and selector resolution into ad_loading.py (#1098)

Browse Source 
## ℹ️ Description

Extract browser-free ad discovery, validation, selector filtering,
category resolution, content-hash helpers, and image globbing from
`__init__.py` into a new `ad_loading.py` module with explicit
dependencies.

Non-breaking internal refactor — no CLI, config, or user-facing behavior
changes.

## 📋 Changes Summary

- **New module** `src/kleinanzeigen_bot/ad_loading.py` (373 lines) with
9 public functions
- **Removed** `_is_valid_ads_selector`, `__check_ad_republication`,
`__check_ad_changed`, `load_ad`, `update_content_hashes` from
`KleinanzeigenBot`
- `load_ads()` kept as temporary delegator (7+ call sites in `run()` /
`download_ads()`)
- **37 focused tests** in `tests/unit/test_ad_loading.py`
- **Removed** 14 duplicate tests from `tests/unit/test_init.py`
- **Translations** moved from `__init__.py` to `ad_loading.py` function
names

### ⚙️ 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

* **Refactor**
* Moved ad discovery, validation, filtering, image/category resolution,
and content-hash management into a dedicated ad-loading component; CLI
selector validation and content-hash updates now use that shared logic.

* **Tests**
* Added comprehensive unit tests for ad discovery, selector
parsing/validation, change detection, republication rules,
image/category resolution, and content-hash update behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Jens
2026-06-10 16:08:01 +02:00
committed by GitHub
parent c7f16e3361
commit 20be5e9528
5 changed files with 1199 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files

229
src/kleinanzeigen_bot/__init__.py
View File

@@ -12,12 +12,11 @@ from typing import TYPE_CHECKING, Any, Final, Sequence, cast
import certifi, colorama # isort: skip
from nodriver.core.connection import ProtocolException
from ruamel.yaml import YAML
from wcmatch import glob
from . import ad_form_helpers as _ad_form_helpers
from . import ad_loading, extract
from . import ad_state as _ad_state
from . import download_selection as _download_selection
from . import extract
from . import local_path_renaming as _local_path_renaming
from . import price_reduction as _price_reduction
from . import runtime_config as _runtime_config
@@ -290,7 +289,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
checker.check_for_updates()
self.ads_selector = "all"
if ads := self.load_ads(exclude_ads_with_id = False):
self.update_content_hashes(ads)
ad_loading.update_content_hashes(ads)
else:
LOG.info("############################################")
LOG.info("DONE: No active ads found.")
@@ -301,7 +300,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
checker = UpdateChecker(self.config, self._update_check_state_path)
checker.check_for_updates()
if not self._is_valid_ads_selector({"all", "new", "due", "changed"}):
if not ad_loading.is_valid_ads_selector(self.ads_selector, {"all", "new", "due", "changed"}):
if self._ads_selector_explicit:
LOG.error(
'Invalid --ads selector: "%s". Valid values: comma-separated keywords (all, new, due, changed) or numeric IDs.',
@@ -321,7 +320,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
case "update":
_bootstrap_runtime()
if not self._is_valid_ads_selector({"all", "changed"}):
if not ad_loading.is_valid_ads_selector(self.ads_selector, {"all", "changed"}):
if self._ads_selector_explicit:
LOG.error('Invalid --ads selector: "%s". Valid values: comma-separated keywords (all, changed) or numeric IDs.', self.ads_selector)
sys.exit(2)
@@ -355,7 +354,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
checker.check_for_updates()
# Default to all ads if no selector provided, but reject invalid values
if not self._is_valid_ads_selector({"all"}):
if not ad_loading.is_valid_ads_selector(self.ads_selector, {"all"}):
if self._ads_selector_explicit:
LOG.error('Invalid --ads selector: "%s". Valid values: all or comma-separated numeric IDs.', self.ads_selector)
sys.exit(2)
@@ -372,7 +371,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
LOG.info("############################################")
case "download":
# ad IDs depends on selector
if not self._is_valid_ads_selector({"all", "new"}):
if not ad_loading.is_valid_ads_selector(self.ads_selector, {"all", "new"}):
if self._ads_selector_explicit:
LOG.error('Invalid --ads selector: "%s". Valid values: comma-separated keywords (all, new) or numeric IDs.', self.ads_selector)
sys.exit(2)
@@ -399,193 +398,24 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
except Exception as exc: # noqa: BLE001
LOG.warning("Timing collector flush failed: %s", exc)
def _is_valid_ads_selector(self, valid_keywords:set[str]) -> bool:
"""Check if the current ads_selector is valid for the given set of keyword selectors.
Accepts a single keyword, a comma-separated list of keywords, or a comma-separated
list of numeric IDs. Mixed keyword+numeric selectors are not supported.
"""
return (
self.ads_selector in valid_keywords
or all(s.strip() in valid_keywords for s in self.ads_selector.split(","))
or _download_selection.is_numeric_ids_selector(self.ads_selector)
)
def __check_ad_republication(self, ad_cfg:Ad, ad_file_relative:str) -> bool:
"""
Check if an ad needs to be republished based on republication interval.
Note: This method does not check for content changes. Use __check_ad_changed for that.
Returns:
True if the ad should be republished based on the interval.
"""
if ad_cfg.updated_on:
last_updated_on = ad_cfg.updated_on
elif ad_cfg.created_on:
last_updated_on = ad_cfg.created_on
else:
return True
if not last_updated_on:
return True
# Check republication interval
ad_age = _misc.now() - last_updated_on
if ad_age.days <= ad_cfg.republication_interval:
LOG.info(
" -> SKIPPED: ad [%s] was last published %d days ago. republication is only required every %s days",
ad_file_relative,
ad_age.days,
ad_cfg.republication_interval,
)
return False
return True
def __check_ad_changed(self, ad_cfg:Ad, ad_cfg_orig:dict[str, Any], ad_file_relative:str) -> bool:
"""
Check if an ad has been changed since last publication.
Returns:
True if the ad has been changed.
"""
if not ad_cfg.id:
# New ads are not considered "changed"
return False
# Calculate hash on original config to match what was stored
current_hash = AdPartial.model_validate(ad_cfg_orig).update_content_hash().content_hash
stored_hash = ad_cfg_orig.get("content_hash")
LOG.debug("Hash comparison for [%s]:", ad_file_relative)
LOG.debug(" Stored hash: %s", stored_hash)
LOG.debug(" Current hash: %s", current_hash)
if stored_hash and current_hash != stored_hash:
LOG.info("Changes detected in ad [%s], will republish", ad_file_relative)
# Update hash in original configuration
ad_cfg_orig["content_hash"] = current_hash
return True
return False
def load_ads(self, *, ignore_inactive:bool = True, exclude_ads_with_id:bool = True) -> list[tuple[str, Ad, dict[str, Any]]]:
"""Load and validate all ad config files.
Temporary thin delegator to :func:`ad_loading.load_ads` — kept to
avoid churn at 7+ call sites inside :meth:`run` and
:meth:`download_ads`. Will be removed when the bot class itself is
decomposed in steps 9–11.
"""
Load and validate all ad config files, optionally filtering out inactive or already-published ads.
Args:
ignore_inactive (bool):
Skip ads with `active=False`.
exclude_ads_with_id (bool):
Skip ads whose raw data already contains an `id`, i.e. was published before.
Returns:
list[tuple[str, Ad, dict[str, Any]]]:
Tuples of (file_path, validated Ad model, original raw data).
"""
LOG.info("Searching for ad config files...")
ad_files:dict[str, str] = {}
data_root_dir = os.path.dirname(self.config_file_path)
for file_pattern in self.config.ad_files:
for ad_file in glob.glob(file_pattern, root_dir = data_root_dir, flags = glob.GLOBSTAR | glob.BRACE | glob.EXTGLOB):
if not str(ad_file).endswith("ad_fields.yaml"):
ad_files[abspath(ad_file, relative_to = data_root_dir)] = ad_file
LOG.info(" -> found %s", pluralize("ad config file", ad_files))
if not ad_files:
return []
ids = []
use_specific_ads = False
selectors = self.ads_selector.split(",")
if _download_selection.is_numeric_ids_selector(self.ads_selector):
ids = [int(n) for n in self.ads_selector.split(",")]
use_specific_ads = True
LOG.info("Start fetch task for the ad(s) with id(s):")
LOG.info(" | ".join([str(id_) for id_ in ids]))
ads = []
for ad_file, ad_file_relative in sorted(ad_files.items()):
ad_cfg_orig:dict[str, Any] = _dicts.load_dict(ad_file, "ad")
ad_cfg:Ad = self.load_ad(ad_cfg_orig)
if ignore_inactive and not ad_cfg.active:
LOG.info(" -> SKIPPED: inactive ad [%s]", ad_file_relative)
continue
if use_specific_ads:
if ad_cfg.id not in ids:
LOG.info(" -> SKIPPED: ad [%s] is not in list of given ids.", ad_file_relative)
continue
else:
# Check if ad should be included based on selectors
should_include = False
# Check for 'changed' selector
if "changed" in selectors and self.__check_ad_changed(ad_cfg, ad_cfg_orig, ad_file_relative):
should_include = True
elif "changed" in selectors and self.command == "update" and _price_reduction.is_auto_price_reduction_due(ad_cfg, ad_file_relative):
should_include = True
# Check for 'new' selector
if "new" in selectors and (not ad_cfg.id or not exclude_ads_with_id):
should_include = True
elif "new" in selectors and ad_cfg.id and exclude_ads_with_id:
LOG.info(" -> SKIPPED: ad [%s] is not new. already has an id assigned.", ad_file_relative)
# Check for 'due' selector
if "due" in selectors:
# For 'due' selector, check if the ad is due for republication based on interval
if self.__check_ad_republication(ad_cfg, ad_file_relative):
should_include = True
# Check for 'all' selector (always include)
if "all" in selectors:
should_include = True
if not should_include:
continue
ensure(get_ad_description(ad_cfg, self.config.ad_defaults, with_affixes = False), _("-> property [description] not specified @ [%s]") % ad_file)
get_ad_description(ad_cfg, self.config.ad_defaults, with_affixes = True) # validates complete description
if ad_cfg.category:
resolved_category_id = self.categories.get(ad_cfg.category)
if not resolved_category_id and ">" in ad_cfg.category:
# this maps actually to the sonstiges/weiteres sub-category
parent_category = ad_cfg.category.rpartition(">")[0].strip()
resolved_category_id = self.categories.get(parent_category)
if resolved_category_id:
LOG.warning("Category [%s] unknown. Using category [%s] with ID [%s] instead.", ad_cfg.category, parent_category, resolved_category_id)
if resolved_category_id:
ad_cfg.category = resolved_category_id
if ad_cfg.images:
images = []
ad_dir = os.path.dirname(ad_file)
for image_pattern in ad_cfg.images:
pattern_images = set()
for image_file in glob.glob(image_pattern, root_dir = ad_dir, flags = glob.GLOBSTAR | glob.BRACE | glob.EXTGLOB):
image_file_ext = os.path.splitext(image_file)[1]
ensure(image_file_ext.lower() in {".gif", ".jpg", ".jpeg", ".png"}, f"Unsupported image file type [{image_file}]")
if os.path.isabs(image_file):
pattern_images.add(image_file)
else:
pattern_images.add(abspath(image_file, relative_to = ad_file))
images.extend(sorted(pattern_images))
ensure(images or not ad_cfg.images, f"No images found for given file patterns {ad_cfg.images} at {ad_dir}")
ad_cfg.images = list(dict.fromkeys(images))
LOG.info(" -> LOADED: ad [%s]", ad_file_relative)
ads.append((ad_file, ad_cfg, ad_cfg_orig))
LOG.info("Loaded %s", pluralize("ad", ads))
return ads
def load_ad(self, ad_cfg_orig:dict[str, Any]) -> Ad:
return AdPartial.model_validate(ad_cfg_orig).to_ad(self.config.ad_defaults)
return ad_loading.load_ads(
config_file_path = self.config_file_path,
ad_file_patterns = self.config.ad_files,
ad_defaults = self.config.ad_defaults,
categories = self.categories,
ads_selector = self.ads_selector,
command = self.command,
ignore_inactive = ignore_inactive,
exclude_ads_with_id = exclude_ads_with_id,
)
async def check_and_wait_for_captcha(self, *, is_login_page:bool = True, page_context:str | None = None) -> None:
captcha_elem = await self.web_probe(
@@ -3018,21 +2848,6 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
else:
LOG.error("The page with the id %d does not exist!", ad_id)
def update_content_hashes(self, ads:list[tuple[str, Ad, dict[str, Any]]]) -> None:
changed = 0
for idx, (ad_file, ad_cfg, ad_cfg_orig) in enumerate(ads, start = 1):
LOG.info("Processing %s/%s: '%s' from [%s]...", idx, len(ads), ad_cfg.title, ad_file)
ad_cfg.update_content_hash()
if ad_cfg.content_hash != ad_cfg_orig["content_hash"]:
changed += 1
ad_cfg_orig["content_hash"] = ad_cfg.content_hash
_dicts.save_dict(ad_file, ad_cfg_orig)
LOG.info("############################################")
LOG.info("DONE: Updated [content_hash] in %s", pluralize("ad", changed))
LOG.info("############################################")
#############################
# main entry point

373
src/kleinanzeigen_bot/ad_loading.py Normal file
View File

@@ -0,0 +1,373 @@
# SPDX-FileCopyrightText: © Jens Bergmann and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""Browser-free ad discovery, validation, selector filtering, and content-hash helpers.
All functions take explicit dependencies — no implicit ``self`` plumbing, no
browser imports. The module owns:
- file discovery from glob patterns
- selector validation and filtering (``new``, ``changed``, ``due``, ``all``,
numeric IDs)
- ad model validation and default application
- category alias resolution
- image globbing and validation
- content-hash comparison and persistence
Orchestration entry point: :func:`load_ads`.
"""
from __future__ import annotations
import os
from datetime import datetime # noqa: TC003 — used in runtime type narrowing via _misc.now()
from gettext import gettext as _
from typing import Any, Final
from wcmatch import glob
from . import download_selection as _download_selection
from . import price_reduction as _price_reduction
from .ad_description import get_ad_description
from .model.ad_model import Ad, AdPartial
from .utils import dicts as _dicts
from .utils import loggers as _loggers
from .utils import misc as _misc
from .utils.files import abspath
from .utils.i18n import pluralize
from .utils.misc import ensure
LOG:Final[_loggers.Logger] = _loggers.get_logger(__name__)
LOG.setLevel(_loggers.INFO)
# --------------------------------------------------------------------------- #
# File discovery
# --------------------------------------------------------------------------- #
def discover_ad_files(
config_file_path:str,
ad_file_patterns:list[str],
) -> dict[str, str]:
"""Glob for ad config files matching *ad_file_patterns*.
Returns a ``{abspath: relative_path}`` dict, excluding any file whose
basename is ``ad_fields.yaml``.
"""
ad_files:dict[str, str] = {}
data_root_dir = os.path.dirname(config_file_path)
for file_pattern in ad_file_patterns:
for ad_file in glob.glob(
file_pattern,
root_dir = data_root_dir,
flags = glob.GLOBSTAR | glob.BRACE | glob.EXTGLOB,
):
if not str(ad_file).endswith("ad_fields.yaml"):
ad_files[abspath(ad_file, relative_to = data_root_dir)] = ad_file
return ad_files
# --------------------------------------------------------------------------- #
# Ad model loading
# --------------------------------------------------------------------------- #
def load_ad(ad_cfg_orig:dict[str, Any], ad_defaults:Any) -> Ad:
"""Validate a raw YAML dict into an :class:`Ad` with *ad_defaults* applied."""
return AdPartial.model_validate(ad_cfg_orig).to_ad(ad_defaults)
# --------------------------------------------------------------------------- #
# Selector helpers
# --------------------------------------------------------------------------- #
def is_valid_ads_selector(ads_selector:str, valid_keywords:set[str]) -> bool:
"""Check whether *ads_selector* is valid for *valid_keywords*.
Accepts a single keyword, a comma-separated list of keywords, or a
comma-separated list of numeric IDs. Mixed keyword+numeric selectors
are rejected.
"""
return (
ads_selector in valid_keywords
or all(s.strip() in valid_keywords for s in ads_selector.split(","))
or _download_selection.is_numeric_ids_selector(ads_selector)
)
# --------------------------------------------------------------------------- #
# Republication / change detection
# --------------------------------------------------------------------------- #
def check_ad_republication(
ad_cfg:Ad,
ad_file_relative:str,
*,
now:datetime | None = None,
) -> bool:
"""Return ``True`` when *ad_cfg* is due for republication.
The interval is measured against :func:`~kleinanzeigen_bot.utils.misc.now`
by default; pass *now* to make the function deterministic in tests.
"""
if ad_cfg.updated_on:
last_updated_on = ad_cfg.updated_on
elif ad_cfg.created_on:
last_updated_on = ad_cfg.created_on
else:
return True
if not last_updated_on:
return True
if now is None:
now = _misc.now()
ad_age = now - last_updated_on
if ad_age.days <= ad_cfg.republication_interval:
LOG.info(
" -> SKIPPED: ad [%s] was last published %d days ago. republication is only required every %s days",
ad_file_relative,
ad_age.days,
ad_cfg.republication_interval,
)
return False
return True
def check_ad_changed(
ad_cfg:Ad,
ad_cfg_orig:dict[str, Any],
ad_file_relative:str,
) -> bool:
"""Return ``True`` when the ad's content hash differs from its stored hash.
.. important::
As a deliberate side effect, this function **mutates**
``ad_cfg_orig["content_hash"]`` with the freshly computed hash when a
change is detected. Callers must be aware of this mutation.
"""
if not ad_cfg.id:
# New ads are not considered "changed"
return False
# Calculate hash on original config to match what was stored
current_hash = AdPartial.model_validate(ad_cfg_orig).update_content_hash().content_hash
stored_hash = ad_cfg_orig.get("content_hash")
LOG.debug("Hash comparison for [%s]:", ad_file_relative)
LOG.debug(" Stored hash: %s", stored_hash)
LOG.debug(" Current hash: %s", current_hash)
if stored_hash and current_hash != stored_hash:
LOG.info("Changes detected in ad [%s], will republish", ad_file_relative)
# Update hash in original configuration
ad_cfg_orig["content_hash"] = current_hash
return True
return False
# --------------------------------------------------------------------------- #
# Category & image resolution
# --------------------------------------------------------------------------- #
def resolve_ad_category(ad_cfg:Ad, categories:dict[str, str]) -> None:
"""Resolve *ad_cfg.category* alias to a numeric category ID in-place.
When the exact alias is unknown but contains ``>``, falls back to the
parent category if that parent is a known alias.
"""
if not ad_cfg.category:
return
resolved_category_id = categories.get(ad_cfg.category)
if not resolved_category_id and ">" in ad_cfg.category:
# this maps actually to the sonstiges/weiteres sub-category
parent_category = ad_cfg.category.rpartition(">")[0].strip()
resolved_category_id = categories.get(parent_category)
if resolved_category_id:
LOG.warning(
"Category [%s] unknown. Using category [%s] with ID [%s] instead.",
ad_cfg.category,
parent_category,
resolved_category_id,
)
if resolved_category_id:
ad_cfg.category = resolved_category_id
def resolve_ad_images(ad_file:str, image_patterns:list[str]) -> list[str]:
"""Glob and validate image files matching *image_patterns*.
Images are resolved relative to *ad_file*'s directory. Supported
extensions: ``.gif``, ``.jpg``, ``.jpeg``, ``.png``.
Returns a deduplicated, ordered list of absolute paths.
"""
if not image_patterns:
return []
images:list[str] = []
ad_dir = os.path.dirname(ad_file)
for image_pattern in image_patterns:
pattern_images = set()
for image_file in glob.glob(
image_pattern,
root_dir = ad_dir,
flags = glob.GLOBSTAR | glob.BRACE | glob.EXTGLOB,
):
image_file_ext = os.path.splitext(image_file)[1]
ensure(
image_file_ext.lower() in {".gif", ".jpg", ".jpeg", ".png"},
_("Unsupported image file type [%s]") % image_file,
)
if os.path.isabs(image_file):
pattern_images.add(image_file)
else:
pattern_images.add(abspath(image_file, relative_to = ad_file))
images.extend(sorted(pattern_images))
ensure(
images or not image_patterns,
_("No images found for given file patterns %s at %s") % (image_patterns, ad_dir),
)
return list(dict.fromkeys(images))
# --------------------------------------------------------------------------- #
# Main ad loading orchestration
# --------------------------------------------------------------------------- #
def load_ads( # noqa: PLR0915
*,
config_file_path:str,
ad_file_patterns:list[str],
ad_defaults:Any,
categories:dict[str, str],
ads_selector:str,
command:str,
ignore_inactive:bool = True,
exclude_ads_with_id:bool = True,
) -> list[tuple[str, Ad, dict[str, Any]]]:
"""Load and validate all ad config files, optionally filtering inactive or already-published ads.
This is the main orchestration function — it wires together file
discovery, model validation, selector filtering, category resolution,
and image globbing. All inputs are explicit.
Returns:
list[tuple[str, Ad, dict[str, Any]]]:
Tuples of ``(file_path, validated Ad model, original raw data)``.
"""
LOG.info("Searching for ad config files...")
ad_files = discover_ad_files(config_file_path, ad_file_patterns)
LOG.info(" -> found %s", pluralize("ad config file", ad_files))
if not ad_files:
return []
ids = []
use_specific_ads = False
# Strip whitespace from selector tokens to match is_valid_ads_selector.
selectors = [token.strip() for token in ads_selector.split(",") if token.strip()]
if _download_selection.is_numeric_ids_selector(ads_selector):
ids = [int(n) for n in ads_selector.split(",")]
use_specific_ads = True
LOG.info("Start fetch task for the ad(s) with id(s):")
LOG.info(" | ".join([str(id_) for id_ in ids]))
ads:list[tuple[str, Ad, dict[str, Any]]] = []
for ad_file, ad_file_relative in sorted(ad_files.items()):
ad_cfg_orig:dict[str, Any] = _dicts.load_dict(ad_file, "ad")
ad_cfg:Ad = load_ad(ad_cfg_orig, ad_defaults)
# Inactive check runs before numeric ID filtering — an inactive ad
# with a matching numeric ID will still be skipped.
if ignore_inactive and not ad_cfg.active:
LOG.info(" -> SKIPPED: inactive ad [%s]", ad_file_relative)
continue
if use_specific_ads:
if ad_cfg.id not in ids:
LOG.info(" -> SKIPPED: ad [%s] is not in list of given ids.", ad_file_relative)
continue
else:
# Check if ad should be included based on selectors
should_include = False
# Check for 'changed' selector
if "changed" in selectors and check_ad_changed(ad_cfg, ad_cfg_orig, ad_file_relative):
should_include = True
elif "changed" in selectors and command == "update" and _price_reduction.is_auto_price_reduction_due(ad_cfg, ad_file_relative):
# Only the "update" command considers pending price reductions
# as a reason to include a "changed" ad.
should_include = True
# Check for 'new' selector
if "new" in selectors and (not ad_cfg.id or not exclude_ads_with_id):
should_include = True
elif "new" in selectors and ad_cfg.id and exclude_ads_with_id:
LOG.info(" -> SKIPPED: ad [%s] is not new. already has an id assigned.", ad_file_relative)
# Check for 'due' selector
if "due" in selectors:
if check_ad_republication(ad_cfg, ad_file_relative):
should_include = True
# Check for 'all' selector (always include)
if "all" in selectors:
should_include = True
if not should_include:
continue
ensure(
get_ad_description(ad_cfg, ad_defaults, with_affixes = False),
_("-> property [description] not specified @ [%s]") % ad_file,
)
get_ad_description(ad_cfg, ad_defaults, with_affixes = True) # validates complete description
resolve_ad_category(ad_cfg, categories)
if ad_cfg.images:
ad_cfg.images = resolve_ad_images(ad_file, ad_cfg.images)
LOG.info(" -> LOADED: ad [%s]", ad_file_relative)
ads.append((ad_file, ad_cfg, ad_cfg_orig))
LOG.info("Loaded %s", pluralize("ad", ads))
return ads
# --------------------------------------------------------------------------- #
# Content hash updates
# --------------------------------------------------------------------------- #
def update_content_hashes(ads:list[tuple[str, Ad, dict[str, Any]]]) -> int:
"""Recompute and persist content hashes for every loaded ad.
Returns the count of ads whose hash actually changed.
"""
changed = 0
for idx, (ad_file, ad_cfg, ad_cfg_orig) in enumerate(ads, start = 1):
LOG.info("Processing %s/%s: '%s' from [%s]...", idx, len(ads), ad_cfg.title, ad_file)
ad_cfg.update_content_hash()
if ad_cfg.content_hash != ad_cfg_orig["content_hash"]:
changed += 1
ad_cfg_orig["content_hash"] = ad_cfg.content_hash
_dicts.save_dict(ad_file, ad_cfg_orig)
LOG.info("############################################")
LOG.info("DONE: Updated [content_hash] in %s", pluralize("ad", changed))
LOG.info("############################################")
return changed

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

@@ -73,26 +73,6 @@ kleinanzeigen_bot/__init__.py:
"Invalid 'next' page value in paging info: %s, stopping pagination": "Ungültiger 'next'-Seitenwert in Paginierungsinfo: %s, beende Paginierung"
"Invalid 'pageNum' in paging info: %s, stopping pagination": "Ungültiger 'pageNum'-Wert in Paginierungsinfo: %s, beende Paginierung"
__check_ad_changed:
"Hash comparison for [%s]:": "Hash-Vergleich für [%s]:"
" Stored hash: %s": " Gespeicherter Hash: %s"
" Current hash: %s": " Aktueller Hash: %s"
"Changes detected in ad [%s], will republish": "Änderungen in Anzeige [%s] erkannt, wird erneut veröffentlicht"
load_ads:
"Searching for ad config files...": "Suche nach Anzeigendateien..."
" -> found %s": "-> %s gefunden"
"ad config file": "Anzeigendatei"
"Start fetch task for the ad(s) with id(s):": "Starte Abrufaufgabe für die Anzeige(n) mit ID(s):"
" -> SKIPPED: inactive ad [%s]": " -> ÜBERSPRUNGEN: inaktive Anzeige [%s]"
" -> SKIPPED: ad [%s] is not in list of given ids.": " -> ÜBERSPRUNGEN: Anzeige [%s] ist nicht in der Liste der angegebenen IDs."
" -> SKIPPED: ad [%s] is not new. already has an id assigned.": " -> ÜBERSPRUNGEN: Anzeige [%s] ist nicht neu. Eine ID wurde bereits zugewiesen."
"-> property [description] not specified @ [%s]": "-> Eigenschaft [description] nicht angegeben @ [%s]"
"Category [%s] unknown. Using category [%s] with ID [%s] instead.": "Kategorie [%s] unbekannt. Verwende stattdessen Kategorie [%s] mit ID [%s]."
" -> LOADED: ad [%s]": " -> GELADEN: Anzeige [%s]"
"Loaded %s": "%s geladen"
"ad": "Anzeige"
check_and_wait_for_captcha:
"# Captcha present! Please solve the captcha.": "# Captcha vorhanden! Bitte lösen Sie das Captcha."
"Captcha recognized - auto-restart enabled, abort run...": "Captcha erkannt - Auto-Neustart aktiviert, Durchlauf wird beendet..."
@@ -283,9 +263,6 @@ kleinanzeigen_bot/__init__.py:
check_thumbnails_uploaded:
" -> %d of %d images processed": " -> %d von %d Bildern verarbeitet"
__check_ad_republication:
" -> SKIPPED: ad [%s] was last published %d days ago. republication is only required every %s days": " -> ÜBERSPRUNGEN: Anzeige [%s] wurde zuletzt vor %d Tagen veröffentlicht. Erneute Veröffentlichung ist erst nach %s Tagen erforderlich"
__select_button_combobox:
"Option '%(value)s' not found in button combobox '%(id)s'": "Option '%(value)s' nicht in Button-Combobox '%(id)s' gefunden"
@@ -357,7 +334,37 @@ kleinanzeigen_bot/__init__.py:
"Unknown shipping option(s), please refer to the documentation/README: %s": "Unbekannte Versandoption(en), bitte lies die Dokumentation/das README: %s"
"You can only specify shipping options for one package size!": "Du kannst nur Versandoptionen für eine Paketgröße angeben!"
#################################################
kleinanzeigen_bot/ad_loading.py:
#################################################
check_ad_changed:
"Changes detected in ad [%s], will republish": "Änderungen in Anzeige [%s] erkannt, wird erneut veröffentlicht"
check_ad_republication:
" -> SKIPPED: ad [%s] was last published %d days ago. republication is only required every %s days": " -> ÜBERSPRUNGEN: Anzeige [%s] wurde zuletzt vor %d Tagen veröffentlicht. Erneute Veröffentlichung ist erst nach %s Tagen erforderlich"
load_ads:
"Searching for ad config files...": "Suche nach Anzeigendateien..."
" -> found %s": "-> %s gefunden"
"ad config file": "Anzeigendatei"
"Start fetch task for the ad(s) with id(s):": "Starte Abrufaufgabe für die Anzeige(n) mit ID(s):"
" -> SKIPPED: inactive ad [%s]": " -> ÜBERSPRUNGEN: inaktive Anzeige [%s]"
" -> SKIPPED: ad [%s] is not in list of given ids.": " -> ÜBERSPRUNGEN: Anzeige [%s] ist nicht in der Liste der angegebenen IDs."
" -> SKIPPED: ad [%s] is not new. already has an id assigned.": " -> ÜBERSPRUNGEN: Anzeige [%s] ist nicht neu. Eine ID wurde bereits zugewiesen."
"-> property [description] not specified @ [%s]": "-> Eigenschaft [description] nicht angegeben @ [%s]"
" -> LOADED: ad [%s]": " -> GELADEN: Anzeige [%s]"
"Loaded %s": "%s geladen"
"ad": "Anzeige"
resolve_ad_category:
"Category [%s] unknown. Using category [%s] with ID [%s] instead.": "Kategorie [%s] unbekannt. Verwende stattdessen Kategorie [%s] mit ID [%s]."
resolve_ad_images:
"Unsupported image file type [%s]": "Nicht unterstütztes Bildformat [%s]"
"No images found for given file patterns %s at %s": "Keine Bilder für die Dateimuster %s in %s gefunden"
update_content_hashes:
"############################################": "############################################"
"DONE: Updated [content_hash] in %s": "FERTIG: [content_hash] in %s aktualisiert."
"Processing %s/%s: '%s' from [%s]...": "Verarbeite %s/%s: '%s' von [%s]..."
"ad": "Anzeige"

772
tests/unit/test_ad_loading.py Normal file
View File

@@ -0,0 +1,772 @@
# 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 logging
import tempfile
from datetime import timedelta
from pathlib import Path
from typing import Any
from unittest.mock import patch
import pytest
from pydantic import ValidationError
from kleinanzeigen_bot.ad_loading import (
check_ad_changed,
check_ad_republication,
discover_ad_files,
is_valid_ads_selector,
load_ads,
resolve_ad_category,
resolve_ad_images,
update_content_hashes,
)
from kleinanzeigen_bot.model.ad_model import Ad
from kleinanzeigen_bot.model.config_model import (
Config,
)
from kleinanzeigen_bot.utils import dicts, misc
# --------------------------------------------------------------------------- #
# Fixtures (local copies for standalone module tests)
# --------------------------------------------------------------------------- #
@pytest.fixture
def base_ad_config() -> dict[str, Any]:
"""Provide a base ad configuration that can be used across tests."""
return {
"id": None,
"title": "Test 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": None,
"contact": {"name": "Test User", "zipcode": "12345", "location": "Test City", "street": "", "phone": ""},
}
@pytest.fixture
def test_bot_config() -> Config:
"""Provides a basic sample configuration for testing."""
return Config.model_validate({
"ad_defaults": {
"contact": {
"name": "dummy_name",
"zipcode": "12345"
},
},
"login": {
"username": "dummy_user",
"password": "dummy_password"
},
"publishing": {
"delete_old_ads": "BEFORE_PUBLISH",
"delete_old_ads_by_title": False
}
})
# --------------------------------------------------------------------------- #
# discover_ad_files
# --------------------------------------------------------------------------- #
def test_discover_ad_files_no_matches(tmp_path:Path) -> None:
"""Globbing with no matching files returns an empty dict."""
config_file = tmp_path / "config.yaml"
config_file.write_text("")
result = discover_ad_files(str(config_file), ["nonexistent/*.yaml"])
assert result == {}
def test_discover_ad_files_finds_matching(tmp_path:Path, base_ad_config:dict[str, Any]) -> None:
"""Globbing finds matching ad files and skips ad_fields.yaml."""
ads_dir = tmp_path / "ads"
ads_dir.mkdir()
ad1 = ads_dir / "my_ad.yaml"
dicts.save_dict(ad1, base_ad_config | {"title": "Ad 1"})
ad2 = ads_dir / "other_ad.yaml"
dicts.save_dict(ad2, base_ad_config | {"title": "Ad 2"})
# ad_fields.yaml should be filtered out
(ads_dir / "ad_fields.yaml").write_text("")
config_file = tmp_path / "config.yaml"
config_file.write_text("")
result = discover_ad_files(str(config_file), ["ads/*.yaml"])
assert len(result) == 2
assert all(str(k).endswith(".yaml") for k in result)
# --------------------------------------------------------------------------- #
# is_valid_ads_selector
# --------------------------------------------------------------------------- #
class TestIsValidAdsSelector:
"""Focused tests for is_valid_ads_selector."""
def test_single_keyword(self) -> None:
assert is_valid_ads_selector("all", {"all", "new"})
def test_keyword_list(self) -> None:
assert is_valid_ads_selector("all,new", {"all", "new"})
def test_numeric_ids(self) -> None:
assert is_valid_ads_selector("123,456", {"all"})
def test_mixed_keyword_and_numeric_rejected(self) -> None:
assert not is_valid_ads_selector("all,123", {"all", "new"})
def test_invalid_keyword(self) -> None:
assert not is_valid_ads_selector("invalid", {"all", "new"})
def test_whitespace_stripping_in_list(self) -> None:
"""Validation strips whitespace; ' all , new ' is valid."""
assert is_valid_ads_selector(" all , new ", {"all", "new"})
# --------------------------------------------------------------------------- #
# check_ad_republication
# --------------------------------------------------------------------------- #
class TestCheckAdRepublication:
"""Focused tests for check_ad_republication."""
def test_no_timestamps_returns_true(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config)
assert check_ad_republication(ad_cfg, "test.yaml") is True
def test_recent_update_returns_false(self, base_ad_config:dict[str, Any]) -> None:
now = misc.now()
yesterday = now - timedelta(days = 1)
ad_cfg = Ad.model_validate(base_ad_config | {
"updated_on": yesterday.isoformat(),
"republication_interval": 7,
})
assert check_ad_republication(ad_cfg, "test.yaml", now = now) is False
def test_old_update_returns_true(self, base_ad_config:dict[str, Any]) -> None:
now = misc.now()
old = now - timedelta(days = 10)
ad_cfg = Ad.model_validate(base_ad_config | {
"updated_on": old.isoformat(),
"republication_interval": 7,
})
assert check_ad_republication(ad_cfg, "test.yaml", now = now) is True
# --------------------------------------------------------------------------- #
# check_ad_changed
# --------------------------------------------------------------------------- #
class TestCheckAdChanged:
"""Focused tests for check_ad_changed."""
def test_no_id_returns_false(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config | {"id": None})
ad_cfg_orig = base_ad_config | {"id": None}
assert check_ad_changed(ad_cfg, ad_cfg_orig, "test.yaml") is False
def test_no_stored_hash_returns_false(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config | {"id": "12345"})
ad_cfg_orig = base_ad_config | {"id": "12345"}
# No content_hash key → stored_hash is None → falsy → returns False
assert check_ad_changed(ad_cfg, ad_cfg_orig, "test.yaml") is False
def test_changed_hash_returns_true_and_mutates(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config | {"id": "12345"})
ad_cfg_orig = base_ad_config | {"id": "12345", "description": "Original"}
ad_cfg_orig["content_hash"] = "old_hash"
result = check_ad_changed(ad_cfg, ad_cfg_orig, "test.yaml")
assert result is True
# Side effect: content_hash was updated
assert ad_cfg_orig["content_hash"] != "old_hash"
# --------------------------------------------------------------------------- #
# resolve_ad_category
# --------------------------------------------------------------------------- #
class TestResolveAdCategory:
"""Focused tests for resolve_ad_category."""
def test_known_alias_resolves(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config | {"category": "garten"})
categories = {"garten": "123"}
resolve_ad_category(ad_cfg, categories)
assert ad_cfg.category == "123"
def test_unknown_without_arrow_unchanged(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config | {"category": "unknown"})
categories = {"garten": "123"}
resolve_ad_category(ad_cfg, categories)
assert ad_cfg.category == "unknown"
def test_parent_fallback(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config | {"category": "garten > blumen"})
categories = {"garten": "123"}
resolve_ad_category(ad_cfg, categories)
assert ad_cfg.category == "123"
def test_parent_fallback_unknown_parent_unchanged(self, base_ad_config:dict[str, Any]) -> None:
ad_cfg = Ad.model_validate(base_ad_config | {"category": "foo > bar"})
categories = {"garten": "123"}
resolve_ad_category(ad_cfg, categories)
assert ad_cfg.category == "foo > bar"
def test_no_category_unresolved_with_empty_categories(self, base_ad_config:dict[str, Any]) -> None:
"""When categories dict is empty, category is kept unchanged."""
ad_cfg = Ad.model_validate(base_ad_config | {"category": "some_category"})
resolve_ad_category(ad_cfg, {})
assert ad_cfg.category == "some_category"
# --------------------------------------------------------------------------- #
# resolve_ad_images
# --------------------------------------------------------------------------- #
class TestResolveAdImages:
"""Focused tests for resolve_ad_images."""
def test_empty_patterns_returns_empty(self, tmp_path:Path) -> None:
ad_file = tmp_path / "test.yaml"
ad_file.write_text("")
assert resolve_ad_images(str(ad_file), []) == []
def test_missing_images_raises(self, tmp_path:Path) -> None:
ad_file = tmp_path / "test.yaml"
ad_file.write_text("")
with pytest.raises(AssertionError, match = "No images found"):
resolve_ad_images(str(ad_file), ["nonexistent/*.jpg"])
def test_finds_matching_images(self, tmp_path:Path) -> None:
ad_file = tmp_path / "test.yaml"
ad_file.write_text("")
(tmp_path / "photo1.jpg").write_text("")
(tmp_path / "photo2.png").write_text("")
result = resolve_ad_images(str(ad_file), ["*.jpg", "*.png"])
assert len(result) == 2
def test_unsupported_extension_raises(self, tmp_path:Path) -> None:
ad_file = tmp_path / "test.yaml"
ad_file.write_text("")
(tmp_path / "photo.bmp").write_text("")
with pytest.raises(AssertionError, match = "Unsupported image file type"):
resolve_ad_images(str(ad_file), ["*.bmp"])
# --------------------------------------------------------------------------- #
# update_content_hashes
# --------------------------------------------------------------------------- #
class TestUpdateContentHashes:
"""Focused tests for update_content_hashes."""
def test_counter_progression(
self, base_ad_config:dict[str, Any], caplog:pytest.LogCaptureFixture
) -> None:
"""Display counter must advance for every ad, even when hash is unchanged."""
ads = [
_build_ad(base_ad_config, None, "Unchanged Ad 1"),
_build_ad(base_ad_config, None, "Changed Ad"),
_build_ad(base_ad_config, None, "Unchanged Ad 2"),
]
# Pre-compute hashes so two match and one differs
for _ad_file, ad_cfg, ad_cfg_orig in ads:
ad_cfg.update_content_hash()
ad_cfg_orig["content_hash"] = ad_cfg.content_hash
# Make the middle ad's original hash differ
ads[1][2]["content_hash"] = "deliberately_wrong_hash"
with (
caplog.at_level(logging.INFO),
patch.object(dicts, "save_dict"),
):
changed = update_content_hashes(ads)
assert changed == 1
processing = [r for r in caplog.records if r.message.startswith("Processing")]
assert len(processing) == 3
assert "1/3" in processing[0].message
assert "2/3" in processing[1].message
assert "3/3" in processing[2].message
summary = [r for r in caplog.records if "DONE:" in r.message and "content_hash" in r.message]
assert any("1 ad" in r.message for r in summary)
# --------------------------------------------------------------------------- #
# load_ads — validation error tests
# --------------------------------------------------------------------------- #
_VALIDATION_ERROR_CASES = [
pytest.param(
{"title": ""},
"title",
id = "missing_title",
),
pytest.param(
{"price_type": "INVALID_TYPE"},
"price_type",
id = "invalid_price_type",
),
pytest.param(
{"shipping_type": "INVALID_TYPE"},
"shipping_type",
id = "invalid_shipping_type",
),
pytest.param(
{"price_type": "GIVE_AWAY", "price": 100},
"price",
id = "invalid_price_config",
),
pytest.param(
{"price_type": "FIXED", "price": None},
"price is required when price_type is FIXED",
id = "missing_price",
),
]
@pytest.mark.parametrize(("ad_overrides", "expected_error"), _VALIDATION_ERROR_CASES)
def test_load_ads_validation_errors(
tmp_path:Path,
base_ad_config:dict[str, Any],
test_bot_config:Config,
ad_overrides:dict[str, Any],
expected_error:str,
) -> None:
"""Invalid ad configurations raise ValidationError with field-specific messages."""
ad_dir = tmp_path / "ads"
ad_dir.mkdir()
ad_file = ad_dir / "test_ad.yaml"
dicts.save_dict(ad_file, base_ad_config | ad_overrides)
config_file = tmp_path / "config.yaml"
config_file.write_text("")
with pytest.raises(ValidationError) as exc_info:
load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = test_bot_config.ad_defaults,
categories = {},
ads_selector = "due",
command = "publish",
)
assert expected_error in str(exc_info.value)
# --------------------------------------------------------------------------- #
# load_ads — selector behavior tests
# --------------------------------------------------------------------------- #
def test_load_ads_with_changed_selector(
test_bot_config:Config, base_ad_config:dict[str, Any]
) -> None:
"""Only changed ads are loaded with the 'changed' selector."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Changed Ad",
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"active": True,
}
)
changed_ad = ad_cfg.model_dump()
changed_hash = ad_cfg.update_content_hash().content_hash
changed_ad["content_hash"] = changed_hash
# Modify to simulate a change
changed_ad["title"] = "Changed Ad - Modified"
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "changed_ad.yaml", changed_ad)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
changed_ad, # First call returns the changed ad
{}, # Second call for ad_fields.yaml
],
):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "changed",
command = "publish",
)
assert len(ads) == 1
assert ads[0][1].title == "Changed Ad - Modified"
def test_load_ads_with_due_selector_includes_all_due_ads(
base_ad_config:dict[str, Any], test_bot_config:Config
) -> None:
"""'due' selector includes all ads due for republication, regardless of changes."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
current_time = misc.now()
old_date = (current_time - timedelta(days = 10)).isoformat()
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Changed Ad",
"updated_on": old_date,
"created_on": old_date,
"republication_interval": 7,
"active": True,
}
)
changed_ad = ad_cfg.model_dump()
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "changed_ad.yaml", changed_ad)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
changed_ad,
{},
],
):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "due",
command = "publish",
)
assert len(ads) == 1
def test_load_ads_with_changed_selector_and_pending_price_reduction(
test_bot_config:Config, base_ad_config:dict[str, Any]
) -> None:
"""'changed' selector also loads ads with pending auto price reductions (update mode)."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Ad With Price Reduction",
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"price": 100,
"price_reduction_count": 0,
"repost_count": 1,
"active": True,
"auto_price_reduction": {
"enabled": True,
"on_update": True,
"strategy": "FIXED",
"amount": 10,
"min_price": 1,
"delay_days": 0,
"delay_reposts": 0,
},
}
)
ad_dict = ad_cfg.model_dump()
ad_dict["content_hash"] = ad_cfg.update_content_hash().content_hash
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "ad_with_reduction.yaml", ad_dict)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
ad_dict,
{},
],
):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "changed",
command = "update",
)
assert len(ads) == 1
assert ads[0][1].title == "Ad With Price Reduction"
def test_load_ads_with_changed_selector_no_price_reduction_when_not_configured(
test_bot_config:Config, base_ad_config:dict[str, Any]
) -> None:
"""'changed' selector does not load an unchanged ad when auto_price_reduction is disabled."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Unchanged Ad",
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"price": 100,
"repost_count": 1,
"active": True,
"auto_price_reduction": {
"enabled": False,
"on_update": True,
"strategy": "FIXED",
"amount": 10,
"min_price": 1,
"delay_days": 0,
"delay_reposts": 0,
},
}
)
ad_dict = ad_cfg.model_dump()
ad_dict["content_hash"] = ad_cfg.update_content_hash().content_hash
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "unchanged_ad.yaml", ad_dict)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
ad_dict,
{},
],
):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "changed",
command = "update",
)
assert len(ads) == 0
def test_load_ads_with_changed_selector_does_not_include_price_reduction_in_publish_mode(
test_bot_config:Config, base_ad_config:dict[str, Any]
) -> None:
"""'changed' selector in publish mode skips unchanged ads even when price reduction is pending."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Ad With Price Reduction",
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"price": 100,
"price_reduction_count": 0,
"repost_count": 1,
"active": True,
"auto_price_reduction": {
"enabled": True,
"on_update": True,
"strategy": "FIXED",
"amount": 10,
"min_price": 1,
"delay_days": 0,
"delay_reposts": 0,
},
}
)
ad_dict = ad_cfg.model_dump()
ad_dict["content_hash"] = ad_cfg.update_content_hash().content_hash
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "ad_with_reduction.yaml", ad_dict)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
ad_dict,
{},
],
):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "changed",
command = "publish",
)
# Should NOT load — price reduction only applies in update mode
assert len(ads) == 0
def test_load_ads_with_new_selector_excludes_already_published(
base_ad_config:dict[str, Any], test_bot_config:Config
) -> None:
"""'new' selector skips ads that already have an id assigned."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
already_published = Ad.model_validate(base_ad_config | {
"id": "12345",
"title": "Already Published",
"active": True,
}).model_dump()
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "published.yaml", already_published)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch("kleinanzeigen_bot.utils.dicts.load_dict", side_effect = [already_published, {}]):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "new",
command = "publish",
)
assert len(ads) == 0
def test_load_ads_with_numeric_ids_includes_only_specified(
base_ad_config:dict[str, Any], test_bot_config:Config
) -> None:
"""Numeric ID selector loads only ads with matching IDs."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
ad1 = Ad.model_validate(base_ad_config | {"id": 101, "title": "Ad Number 101", "active": True}).model_dump()
ad2 = Ad.model_validate(base_ad_config | {"id": 202, "title": "Ad Number 202", "active": True}).model_dump()
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "ad101.yaml", ad1)
dicts.save_dict(ad_dir / "ad202.yaml", ad2)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch("kleinanzeigen_bot.utils.dicts.load_dict", side_effect = [ad1, ad2, {}]):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "101",
command = "publish",
)
assert len(ads) == 1
assert ads[0][1].id == 101
def test_load_ads_skips_inactive_before_numeric_id(
base_ad_config:dict[str, Any], test_bot_config:Config
) -> None:
"""Inactive ads are skipped even when their numeric ID matches the selector."""
ad_defaults = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}).ad_defaults
inactive_ad = Ad.model_validate(base_ad_config | {
"id": 101, "title": "Inactive Ad Title", "active": False,
}).model_dump()
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "inactive.yaml", inactive_ad)
config_file = temp_path / "config.yaml"
config_file.write_text("")
with patch("kleinanzeigen_bot.utils.dicts.load_dict", side_effect = [inactive_ad, {}]):
ads = load_ads(
config_file_path = str(config_file),
ad_file_patterns = ["ads/*.yaml"],
ad_defaults = ad_defaults,
categories = {},
ads_selector = "101",
command = "publish",
)
# Inactive check happens before numeric ID filtering
assert len(ads) == 0
# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def _build_ad(
base_ad_config:dict[str, Any],
ad_id:int | None,
title:str,
) -> tuple[str, Ad, dict[str, Any]]:
"""Build an (ad_file, Ad, raw_dict) tuple for use in update_content_hashes tests."""
ad_file = f"/fake/path/{title.replace(' ', '_')}.yaml"
ad_cfg = Ad.model_validate(base_ad_config | {"id": str(ad_id) if ad_id else None, "title": title})
ad_cfg_orig:dict[str, Any] = ad_cfg.model_dump()
return ad_file, ad_cfg, ad_cfg_orig

469
tests/unit/test_init.py
View File

@@ -1,17 +1,15 @@
# 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 asyncio, copy, fnmatch, json, logging, os, tempfile # isort: skip
import asyncio, copy, fnmatch, json, logging, os # isort: skip
from collections.abc import Callable, Generator
from contextlib import ExitStack, contextmanager
from datetime import timedelta
from pathlib import Path, PureWindowsPath
from typing import Any, Awaitable, Iterator, cast
from unittest.mock import AsyncMock, MagicMock, call, patch
import pytest
from nodriver.core.connection import ProtocolException
from pydantic import ValidationError
from kleinanzeigen_bot import (
LOG,
@@ -24,13 +22,12 @@ from kleinanzeigen_bot import (
from kleinanzeigen_bot._version import __version__
from kleinanzeigen_bot.model.ad_model import Ad, AdUpdateStrategy
from kleinanzeigen_bot.model.config_model import (
AdDefaults,
AutoPriceReductionConfig,
Config,
DiagnosticsConfig,
PublishingConfig,
)
from kleinanzeigen_bot.utils import dicts, misc, xdg_paths
from kleinanzeigen_bot.utils import xdg_paths
from kleinanzeigen_bot.utils.exceptions import CategoryResolutionError, PublishedAdsFetchIncompleteError, PublishSubmissionUncertainError
from kleinanzeigen_bot.utils.web_scraping_mixin import By, Element
@@ -1999,39 +1996,6 @@ class TestDisplayCounterProgression:
def _build_published_ads(*ad_specs:tuple[int, str]) -> list[dict[str, Any]]:
return [{"id": ad_id, "state": state} for ad_id, state in ad_specs]
def test_update_content_hashes_counter_progression(
self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], caplog:pytest.LogCaptureFixture
) -> None:
"""Display counter must advance for every ad, even when hash is unchanged."""
ads = [
self._build_ad(base_ad_config, None, "Unchanged Ad 1"),
self._build_ad(base_ad_config, None, "Changed Ad"),
self._build_ad(base_ad_config, None, "Unchanged Ad 2"),
]
# Pre-compute hashes so two match and one differs
for _ad_file, ad_cfg, ad_cfg_orig in ads:
ad_cfg.update_content_hash()
ad_cfg_orig["content_hash"] = ad_cfg.content_hash
# Make the middle ad's original hash differ
ads[1][2]["content_hash"] = "deliberately_wrong_hash"
with (
caplog.at_level(logging.INFO),
patch.object(dicts, "save_dict"),
):
test_bot.update_content_hashes(ads)
processing = [r for r in caplog.records if r.message.startswith("Processing")]
assert len(processing) == 3
assert "1/3" in processing[0].message
assert "2/3" in processing[1].message
assert "3/3" in processing[2].message
summary = [r for r in caplog.records if "DONE:" in r.message and "content_hash" in r.message]
assert any("1 ad" in r.message for r in summary)
@pytest.mark.asyncio
async def test_publish_ads_counter_progression_with_paused_ads(
self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], caplog:pytest.LogCaptureFixture
@@ -2363,12 +2327,6 @@ class TestKleinanzeigenBotAdOperations:
await test_bot.run(["script.py", "extend"])
assert test_bot.ads_selector == "all"
def test_load_ads_no_files(self, test_bot:KleinanzeigenBot) -> None:
"""Test loading ads with no files."""
test_bot.config.ad_files = ["nonexistent/*.yaml"]
ads = test_bot.load_ads()
assert len(ads) == 0
class TestKleinanzeigenBotAdManagement:
"""Tests for ad management functionality."""
@@ -2410,106 +2368,6 @@ class TestKleinanzeigenBotAdManagement:
assert exc_info.value.code == 2
class TestKleinanzeigenBotAdConfiguration:
"""Tests for ad configuration functionality."""
def test_load_ads_with_missing_title(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None:
"""Test loading ads with missing title."""
temp_path = Path(tmp_path)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
ad_file = ad_dir / "test_ad.yaml"
# Create a minimal config with empty title to trigger validation
ad_cfg = minimal_ad_config | {"title": ""}
dicts.save_dict(ad_file, ad_cfg)
# Set config file path to tmp_path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with pytest.raises(ValidationError) as exc_info:
test_bot.load_ads()
assert "title" in str(exc_info.value)
def test_load_ads_with_invalid_price_type(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None:
"""Test loading ads with invalid price type."""
temp_path = Path(tmp_path)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
ad_file = ad_dir / "test_ad.yaml"
# Create config with invalid price type
ad_cfg = minimal_ad_config | {"price_type": "INVALID_TYPE"}
dicts.save_dict(ad_file, ad_cfg)
# Set config file path to tmp_path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with pytest.raises(ValidationError) as exc_info:
test_bot.load_ads()
assert "price_type" in str(exc_info.value)
def test_load_ads_with_invalid_shipping_type(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None:
"""Test loading ads with invalid shipping type."""
temp_path = Path(tmp_path)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
ad_file = ad_dir / "test_ad.yaml"
# Create config with invalid shipping type
ad_cfg = minimal_ad_config | {"shipping_type": "INVALID_TYPE"}
dicts.save_dict(ad_file, ad_cfg)
# Set config file path to tmp_path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with pytest.raises(ValidationError) as exc_info:
test_bot.load_ads()
assert "shipping_type" in str(exc_info.value)
def test_load_ads_with_invalid_price_config(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None:
"""Test loading ads with invalid price configuration."""
temp_path = Path(tmp_path)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
ad_file = ad_dir / "test_ad.yaml"
# Create config with price for GIVE_AWAY type
ad_cfg = minimal_ad_config | {
"price_type": "GIVE_AWAY",
"price": 100, # Price should not be set for GIVE_AWAY
}
dicts.save_dict(ad_file, ad_cfg)
# Set config file path to tmp_path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with pytest.raises(ValidationError) as exc_info:
test_bot.load_ads()
assert "price" in str(exc_info.value)
def test_load_ads_with_missing_price(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None:
"""Test loading ads with missing price for FIXED price type."""
temp_path = Path(tmp_path)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
ad_file = ad_dir / "test_ad.yaml"
# Create config with FIXED price type but no price
ad_cfg = minimal_ad_config | {
"price_type": "FIXED",
"price": None, # Missing required price for FIXED type
}
dicts.save_dict(ad_file, ad_cfg)
# Set config file path to tmp_path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with pytest.raises(ValidationError) as exc_info:
test_bot.load_ads()
assert "price is required when price_type is FIXED" in str(exc_info.value)
class TestKleinanzeigenBotAdDeletion:
"""Tests for ad deletion functionality."""
@@ -2713,58 +2571,6 @@ class TestKleinanzeigenBotAdDeletion:
mock_web_sleep.assert_not_called()
class TestKleinanzeigenBotAdRepublication:
"""Tests for ad republication functionality."""
def test_check_ad_republication_with_changes(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None:
"""Test that ads with changes are marked for republication."""
# Mock the description config to prevent modification of the description
test_bot.config.ad_defaults = AdDefaults.model_validate({"description": {"prefix": "", "suffix": ""}})
# Create ad config with all necessary fields for republication
ad_cfg = Ad.model_validate(
base_ad_config | {"id": "12345", "updated_on": "2024-01-01T00:00:01", "created_on": "2024-01-01T00:00:01", "description": "Changed description"}
)
# Create a temporary directory and file
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
ad_file = ad_dir / "test_ad.yaml"
dicts.save_dict(ad_file, ad_cfg.model_dump())
# Set config file path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
ads_to_publish = test_bot.load_ads()
assert len(ads_to_publish) == 1
def test_check_ad_republication_no_changes(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None:
"""Test that unchanged ads within interval are not marked for republication."""
current_time = misc.now()
three_days_ago = (current_time - timedelta(days = 3)).isoformat()
# Create ad config with timestamps for republication check
ad_cfg = Ad.model_validate(base_ad_config | {"id": "12345", "updated_on": three_days_ago, "created_on": three_days_ago})
# Calculate hash before making the copy to ensure they match
ad_cfg_orig = ad_cfg.model_dump()
current_hash = ad_cfg.update_content_hash().content_hash
ad_cfg_orig["content_hash"] = current_hash
# Mock the config to prevent actual file operations
test_bot.config.ad_files = ["test.yaml"]
with (
patch("kleinanzeigen_bot.utils.dicts.load_dict_if_exists", return_value = ad_cfg_orig),
patch("kleinanzeigen_bot.utils.dicts.load_dict", return_value = {}),
): # Mock ad_fields.yaml
ads_to_publish = test_bot.load_ads()
assert len(ads_to_publish) == 0 # No ads should be marked for republication
class TestKleinanzeigenBotShippingOptions:
"""Tests for shipping options functionality."""
@@ -4549,277 +4355,6 @@ class TestWantedShippingSelection:
await test_bot.publish_ad(ad_file, ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.REPLACE)
class TestKleinanzeigenBotChangedAds:
"""Tests for the 'changed' ads selector functionality."""
def test_load_ads_with_changed_selector(self, test_bot_config:Config, base_ad_config:dict[str, Any]) -> None:
"""Test that only changed ads are loaded when using the 'changed' selector."""
# Set up the bot with the 'changed' selector
test_bot = KleinanzeigenBot()
test_bot.ads_selector = "changed"
test_bot.config = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}})
# Create a changed ad
ad_cfg = Ad.model_validate(
base_ad_config | {"id": "12345", "title": "Changed Ad", "updated_on": "2024-01-01T00:00:00", "created_on": "2024-01-01T00:00:00", "active": True}
)
# Calculate hash for changed_ad and add it to the config
# Then modify the ad to simulate a change
changed_ad = ad_cfg.model_dump()
changed_hash = ad_cfg.update_content_hash().content_hash
changed_ad["content_hash"] = changed_hash
# Now modify the ad to make it "changed"
changed_ad["title"] = "Changed Ad - Modified"
# Create temporary directory and file
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
# Write the ad file
dicts.save_dict(ad_dir / "changed_ad.yaml", changed_ad)
# Set config file path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
# Mock the loading of the ad configuration
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
changed_ad, # First call returns the changed ad
{}, # Second call for ad_fields.yaml
],
):
ads_to_publish = test_bot.load_ads()
# The changed ad should be loaded
assert len(ads_to_publish) == 1
assert ads_to_publish[0][1].title == "Changed Ad - Modified"
def test_load_ads_with_due_selector_includes_all_due_ads(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None:
"""Test that 'due' selector includes all ads that are due for republication, regardless of changes."""
# Set up the bot with the 'due' selector
test_bot.ads_selector = "due"
# Create a changed ad that is also due for republication
current_time = misc.now()
old_date = (current_time - timedelta(days = 10)).isoformat() # Past republication interval
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Changed Ad",
"updated_on": old_date,
"created_on": old_date,
"republication_interval": 7, # Due for republication after 7 days
"active": True,
}
)
changed_ad = ad_cfg.model_dump()
# Create temporary directory and file
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
# Write the ad file
dicts.save_dict(ad_dir / "changed_ad.yaml", changed_ad)
# Set config file path and use relative path for ad_files
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
# Mock the loading of the ad configuration
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
changed_ad, # First call returns the changed ad
{}, # Second call for ad_fields.yaml
],
):
ads_to_publish = test_bot.load_ads()
# The changed ad should be loaded with 'due' selector because it's due for republication
assert len(ads_to_publish) == 1
def test_load_ads_with_changed_selector_and_pending_price_reduction(self, test_bot_config:Config, base_ad_config:dict[str, Any]) -> None:
"""Test that 'changed' selector also loads ads with pending auto price reductions."""
test_bot = KleinanzeigenBot()
test_bot.ads_selector = "changed"
test_bot.command = "update"
test_bot.config = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}})
# Create an ad with auto_price_reduction configured and ready to trigger
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Ad With Price Reduction",
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"price": 100,
"price_reduction_count": 0,
"repost_count": 1,
"active": True,
"auto_price_reduction": {
"enabled": True,
"on_update": True,
"strategy": "FIXED",
"amount": 10,
"min_price": 1,
"delay_days": 0,
"delay_reposts": 0,
},
}
)
# Store the content hash so __check_ad_changed sees no change
ad_dict = ad_cfg.model_dump()
ad_dict["content_hash"] = ad_cfg.update_content_hash().content_hash
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "ad_with_reduction.yaml", ad_dict)
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
ad_dict, # First call returns the ad
{}, # Second call for ad_fields.yaml
],
):
ads_to_publish = test_bot.load_ads()
# The ad should be loaded because a price reduction is pending
assert len(ads_to_publish) == 1
assert ads_to_publish[0][1].title == "Ad With Price Reduction"
def test_load_ads_with_changed_selector_no_price_reduction_when_not_configured(self, test_bot_config:Config, base_ad_config:dict[str, Any]) -> None:
"""Test that 'changed' selector does not load an unchanged ad when auto_price_reduction is disabled."""
test_bot = KleinanzeigenBot()
test_bot.ads_selector = "changed"
test_bot.command = "update"
test_bot.config = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}})
# Create an ad with auto_price_reduction disabled
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Unchanged Ad",
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"price": 100,
"repost_count": 1,
"active": True,
"auto_price_reduction": {
"enabled": False,
"on_update": True,
"strategy": "FIXED",
"amount": 10,
"min_price": 1,
"delay_days": 0,
"delay_reposts": 0,
},
}
)
# Store the content hash so __check_ad_changed sees no change
ad_dict = ad_cfg.model_dump()
ad_dict["content_hash"] = ad_cfg.update_content_hash().content_hash
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "unchanged_ad.yaml", ad_dict)
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
ad_dict, # First call returns the ad
{}, # Second call for ad_fields.yaml
],
):
ads_to_publish = test_bot.load_ads()
# The ad should NOT be loaded because it's unchanged and auto_price_reduction is disabled
assert len(ads_to_publish) == 0
def test_load_ads_with_changed_selector_does_not_include_price_reduction_in_publish_mode(
self, test_bot_config:Config, base_ad_config:dict[str, Any]) -> None:
"""Test that 'changed' selector in publish mode skips unchanged ads even when price reduction is pending."""
test_bot = KleinanzeigenBot()
test_bot.ads_selector = "changed"
test_bot.command = "publish"
test_bot.config = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}})
# Create an ad with auto_price_reduction configured and ready to trigger
ad_cfg = Ad.model_validate(
base_ad_config
| {
"id": "12345",
"title": "Ad With Price Reduction",
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"price": 100,
"price_reduction_count": 0,
"repost_count": 1,
"active": True,
"auto_price_reduction": {
"enabled": True,
"on_update": True,
"strategy": "FIXED",
"amount": 10,
"min_price": 1,
"delay_days": 0,
"delay_reposts": 0,
},
}
)
# Store the content hash so __check_ad_changed sees no change
ad_dict = ad_cfg.model_dump()
ad_dict["content_hash"] = ad_cfg.update_content_hash().content_hash
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ad_dir = temp_path / "ads"
ad_dir.mkdir()
dicts.save_dict(ad_dir / "ad_with_reduction.yaml", ad_dict)
test_bot.config_file_path = str(temp_path / "config.yaml")
test_bot.config.ad_files = ["ads/*.yaml"]
with patch(
"kleinanzeigen_bot.utils.dicts.load_dict",
side_effect = [
ad_dict, # First call returns the ad
{}, # Second call for ad_fields.yaml
],
):
ads_to_publish = test_bot.load_ads()
# The ad should NOT be loaded because price reduction is only applied in update mode
assert len(ads_to_publish) == 0
def test_file_logger_writes_message(tmp_path:Path, caplog:pytest.LogCaptureFixture) -> None:
"""
Unit: Logger can be initialized and used, robust to pytest log capture.