feat: Add automatic price reduction on reposts (#691)

This commit is contained in:
Jens
2025-12-17 20:31:58 +01:00
committed by GitHub
parent 25079c32c0
commit 920ddf5533
13 changed files with 1753 additions and 22 deletions

View File

@@ -384,8 +384,16 @@ description_suffix: # optional suffix to be added to the description overriding
# or category ID (e.g. 161/278) # or category ID (e.g. 161/278)
category: # e.g. "Elektronik > Notebooks" category: # e.g. "Elektronik > Notebooks"
price: # without decimals, e.g. 75 price: # price in euros; decimals allowed but will be rounded to nearest whole euro on processing (prefer whole euros for predictability)
price_type: # one of: FIXED, NEGOTIABLE, GIVE_AWAY (default: NEGOTIABLE) price_type: # one of: FIXED, NEGOTIABLE, GIVE_AWAY (default: NEGOTIABLE)
auto_price_reduction:
enabled: # true or false to enable automatic price reduction on reposts (default: false)
strategy: # "PERCENTAGE" or "FIXED" (required when enabled is true)
amount: # reduction amount; interpreted as percent for PERCENTAGE or currency units for FIXED (prefer whole euros for predictability)
min_price: # required when enabled is true; minimum price floor (use 0 for no lower bound, prefer whole euros for predictability)
delay_reposts: # number of reposts to wait before first reduction (default: 0)
delay_days: # number of days to wait after publication before reductions (default: 0)
# NOTE: All prices are rounded to whole euros after each reduction step.
special_attributes: special_attributes:
# haus_mieten.zimmer_d: value # Zimmer # haus_mieten.zimmer_d: value # Zimmer
@@ -428,8 +436,61 @@ id: # the ID assigned by kleinanzeigen.de
created_on: # ISO timestamp when the ad was first published created_on: # ISO timestamp when the ad was first published
updated_on: # ISO timestamp when the ad was last published updated_on: # ISO timestamp when the ad was last published
content_hash: # hash of the ad content, used to detect changes content_hash: # hash of the ad content, used to detect changes
repost_count: # how often the ad has been (re)published; used for automatic price reductions
``` ```
#### Automatic price reduction on reposts
When `auto_price_reduction.enabled` is set to `true`, the bot lowers the configured `price` every time the ad is reposted. The starting point for the calculation is always the base price from your ad file (the value of `price`), ensuring the first publication uses the unchanged amount. For each repost the bot subtracts either a percentage of the previously published price (strategy: PERCENTAGE) or a fixed amount (strategy: FIXED) and clamps the result to `min_price`.
**Important:** Price reductions only apply when using the `publish` command (which deletes the old ad and creates a new one). Using the `update` command to modify ad content does NOT trigger price reductions or increment `repost_count`.
`repost_count` is tracked for every ad (and persisted inside the corresponding `ad_*.yaml`) so reductions continue across runs.
`min_price` is required whenever `enabled` is `true` and must be less than or equal to `price`; this makes an explicit floor (including `0`) mandatory. If `min_price` equals the current price, the bot will log a warning and perform no reduction.
**Note:** `repost_count` and price reduction counters are only incremented and persisted after a successful publish. Failed publish attempts do not advance the counters.
**PERCENTAGE strategy example:**
```yaml
price: 150
price_type: FIXED
auto_price_reduction:
enabled: true
strategy: PERCENTAGE
amount: 10
min_price: 90
delay_reposts: 0
delay_days: 0
```
This posts the ad at 150 € the first time, then 135 € (10%), 122 € (10%), 110 € (10%), 99 € (10%), and stops decreasing at 90 €.
**Note:** The bot applies commercial rounding (ROUND_HALF_UP) to full euros after each reduction step. For example, 121.5 rounds to 122, and 109.8 rounds to 110. This step-wise rounding affects the final price progression, especially for percentage-based reductions.
**FIXED strategy example:**
```yaml
price: 150
price_type: FIXED
auto_price_reduction:
enabled: true
strategy: FIXED
amount: 15
min_price: 90
delay_reposts: 0
delay_days: 0
```
This posts the ad at 150 € the first time, then 135 € (15 €), 120 € (15 €), 105 € (15 €), and stops decreasing at 90 €.
**Note on `delay_days` behavior:** The `delay_days` parameter counts complete 24-hour periods (whole days) since the ad was published. For example, if `delay_days: 7` and the ad was published 6 days and 23 hours ago, the reduction will not yet apply. This ensures predictable behavior and avoids partial-day ambiguity.
Set `auto_price_reduction.enabled: false` (or omit the entire `auto_price_reduction` section) to keep the existing behaviour—prices stay fixed and `repost_count` only acts as tracked metadata for future changes.
You can configure `auto_price_reduction` once under `ad_defaults` in `config.yaml`. The `min_price` can be set there or overridden per ad file as needed.
### <a name="description-prefix-suffix"></a>3) Description Prefix and Suffix ### <a name="description-prefix-suffix"></a>3) Description Prefix and Suffix
You can add prefix and suffix text to your ad descriptions in two ways: You can add prefix and suffix text to your ad descriptions in two ways:

View File

@@ -353,7 +353,8 @@ markers = [
"slow: marks a test as long running", "slow: marks a test as long running",
"smoke: marks a test as a high-level smoke test (critical path, no mocks)", "smoke: marks a test as a high-level smoke test (critical path, no mocks)",
"itest: marks a test as an integration test (i.e. a test with external dependencies)", "itest: marks a test as an integration test (i.e. a test with external dependencies)",
"asyncio: mark test as async" "asyncio: mark test as async",
"unit: marks a test as a unit test"
] ]
asyncio_mode = "auto" asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function" asyncio_default_fixture_loop_scope = "function"

View File

@@ -1,5 +1,76 @@
{ {
"$defs": { "$defs": {
"AutoPriceReductionConfig": {
"properties": {
"enabled": {
"default": false,
"description": "automatically lower the price of reposted ads",
"title": "Enabled",
"type": "boolean"
},
"strategy": {
"anyOf": [
{
"enum": [
"FIXED",
"PERCENTAGE"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "PERCENTAGE reduces by a percentage of the previous price, FIXED reduces by a fixed amount",
"title": "Strategy"
},
"amount": {
"anyOf": [
{
"exclusiveMinimum": 0,
"type": "number"
},
{
"type": "null"
}
],
"default": null,
"description": "magnitude of the reduction; interpreted as percent for PERCENTAGE or currency units for FIXED",
"title": "Amount"
},
"min_price": {
"anyOf": [
{
"minimum": 0,
"type": "number"
},
{
"type": "null"
}
],
"default": null,
"description": "required when enabled is true; minimum price floor (use 0 for no lower bound)",
"title": "Min Price"
},
"delay_reposts": {
"default": 0,
"description": "number of reposts to wait before applying the first automatic price reduction",
"minimum": 0,
"title": "Delay Reposts",
"type": "integer"
},
"delay_days": {
"default": 0,
"description": "number of days to wait after publication before applying automatic price reductions",
"minimum": 0,
"title": "Delay Days",
"type": "integer"
}
},
"title": "AutoPriceReductionConfig",
"type": "object"
},
"ContactPartial": { "ContactPartial": {
"properties": { "properties": {
"name": { "name": {
@@ -181,6 +252,32 @@
"default": null, "default": null,
"title": "Price Type" "title": "Price Type"
}, },
"auto_price_reduction": {
"anyOf": [
{
"$ref": "#/$defs/AutoPriceReductionConfig"
},
{
"type": "null"
}
],
"default": null,
"description": "automatic price reduction configuration"
},
"repost_count": {
"default": 0,
"description": "number of successful publications for this ad (persisted between runs)",
"minimum": 0,
"title": "Repost Count",
"type": "integer"
},
"price_reduction_count": {
"default": 0,
"description": "internal counter: number of automatic price reductions already applied",
"minimum": 0,
"title": "Price Reduction Count",
"type": "integer"
},
"shipping_type": { "shipping_type": {
"anyOf": [ "anyOf": [
{ {

View File

@@ -64,6 +64,10 @@
"title": "Price Type", "title": "Price Type",
"type": "string" "type": "string"
}, },
"auto_price_reduction": {
"$ref": "#/$defs/AutoPriceReductionConfig",
"description": "automatic price reduction configuration"
},
"shipping_type": { "shipping_type": {
"default": "SHIPPING", "default": "SHIPPING",
"enum": [ "enum": [
@@ -107,6 +111,77 @@
"title": "AdDefaults", "title": "AdDefaults",
"type": "object" "type": "object"
}, },
"AutoPriceReductionConfig": {
"properties": {
"enabled": {
"default": false,
"description": "automatically lower the price of reposted ads",
"title": "Enabled",
"type": "boolean"
},
"strategy": {
"anyOf": [
{
"enum": [
"FIXED",
"PERCENTAGE"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "PERCENTAGE reduces by a percentage of the previous price, FIXED reduces by a fixed amount",
"title": "Strategy"
},
"amount": {
"anyOf": [
{
"exclusiveMinimum": 0,
"type": "number"
},
{
"type": "null"
}
],
"default": null,
"description": "magnitude of the reduction; interpreted as percent for PERCENTAGE or currency units for FIXED",
"title": "Amount"
},
"min_price": {
"anyOf": [
{
"minimum": 0,
"type": "number"
},
{
"type": "null"
}
],
"default": null,
"description": "required when enabled is true; minimum price floor (use 0 for no lower bound)",
"title": "Min Price"
},
"delay_reposts": {
"default": 0,
"description": "number of reposts to wait before applying the first automatic price reduction",
"minimum": 0,
"title": "Delay Reposts",
"type": "integer"
},
"delay_days": {
"default": 0,
"description": "number of days to wait after publication before applying automatic price reductions",
"minimum": 0,
"title": "Delay Days",
"type": "integer"
}
},
"title": "AutoPriceReductionConfig",
"type": "object"
},
"BrowserConfig": { "BrowserConfig": {
"properties": { "properties": {
"arguments": { "arguments": {

View File

@@ -5,6 +5,7 @@ import atexit, enum, json, os, re, signal, sys, textwrap # isort: skip
import getopt # pylint: disable=deprecated-module import getopt # pylint: disable=deprecated-module
import urllib.parse as urllib_parse import urllib.parse as urllib_parse
from gettext import gettext as _ from gettext import gettext as _
from pathlib import Path
from typing import Any, Final from typing import Any, Final
import certifi, colorama, nodriver # isort: skip import certifi, colorama, nodriver # isort: skip
@@ -13,7 +14,7 @@ from wcmatch import glob
from . import extract, resources from . import extract, resources
from ._version import __version__ from ._version import __version__
from .model.ad_model import MAX_DESCRIPTION_LENGTH, Ad, AdPartial from .model.ad_model import MAX_DESCRIPTION_LENGTH, Ad, AdPartial, calculate_auto_price
from .model.config_model import Config from .model.config_model import Config
from .update_checker import UpdateChecker from .update_checker import UpdateChecker
from .utils import dicts, error_handlers, loggers, misc from .utils import dicts, error_handlers, loggers, misc
@@ -36,6 +37,145 @@ class AdUpdateStrategy(enum.Enum):
MODIFY = enum.auto() MODIFY = enum.auto()
def _repost_cycle_ready(ad_cfg:Ad, ad_file_relative:str) -> bool:
"""
Check if the repost cycle delay has been satisfied.
:param ad_cfg: The ad configuration
:param ad_file_relative: Relative path to the ad file for logging
:return: True if ready to apply price reduction, False otherwise
"""
total_reposts = ad_cfg.repost_count or 0
delay_reposts = ad_cfg.auto_price_reduction.delay_reposts
applied_cycles = ad_cfg.price_reduction_count or 0
eligible_cycles = max(total_reposts - delay_reposts, 0)
if total_reposts <= delay_reposts:
remaining = (delay_reposts + 1) - total_reposts
LOG.info(
_("Auto price reduction delayed for [%s]: waiting %s more reposts (completed %s, applied %s reductions)"),
ad_file_relative,
max(remaining, 1), # Clamp to 1 to avoid showing "0 more reposts" when at threshold
total_reposts,
applied_cycles
)
return False
if eligible_cycles <= applied_cycles:
LOG.debug(
_("Auto price reduction already applied for [%s]: %s reductions match %s eligible reposts"),
ad_file_relative,
applied_cycles,
eligible_cycles
)
return False
return True
def _day_delay_elapsed(ad_cfg:Ad, ad_file_relative:str) -> bool:
"""
Check if the day delay has elapsed since the ad was last published.
:param ad_cfg: The ad configuration
:param ad_file_relative: Relative path to the ad file for logging
:return: True if the delay has elapsed, False otherwise
"""
delay_days = ad_cfg.auto_price_reduction.delay_days
if delay_days == 0:
return True
reference = ad_cfg.updated_on or ad_cfg.created_on
if not reference:
LOG.info(
_("Auto price reduction delayed for [%s]: waiting %s days but publish timestamp missing"),
ad_file_relative,
delay_days
)
return False
# Note: .days truncates to whole days (e.g., 1.9 days -> 1 day)
# This is intentional: delays count complete 24-hour periods since publish
# Both misc.now() and stored timestamps use UTC (via misc.now()), ensuring consistent calculations
elapsed_days = (misc.now() - reference).days
if elapsed_days < delay_days:
LOG.info(
_("Auto price reduction delayed for [%s]: waiting %s days (elapsed %s)"),
ad_file_relative,
delay_days,
elapsed_days
)
return False
return True
def apply_auto_price_reduction(ad_cfg:Ad, _ad_cfg_orig:dict[str, Any], ad_file_relative:str) -> None:
"""
Apply automatic price reduction to an ad based on repost count and configuration.
This function modifies ad_cfg in-place, updating the price and price_reduction_count
fields when a reduction is applicable.
:param ad_cfg: The ad configuration to potentially modify
:param _ad_cfg_orig: The original ad configuration (unused, kept for compatibility)
:param ad_file_relative: Relative path to the ad file for logging
"""
if not ad_cfg.auto_price_reduction.enabled:
return
base_price = ad_cfg.price
if base_price is None:
LOG.warning(_("Auto price reduction is enabled for [%s] but no price is configured."), ad_file_relative)
return
if ad_cfg.auto_price_reduction.min_price is not None and ad_cfg.auto_price_reduction.min_price == base_price:
LOG.warning(
_("Auto price reduction is enabled for [%s] but min_price equals price (%s) - no reductions will occur."),
ad_file_relative,
base_price
)
return
if not _repost_cycle_ready(ad_cfg, ad_file_relative):
return
if not _day_delay_elapsed(ad_cfg, ad_file_relative):
return
applied_cycles = ad_cfg.price_reduction_count or 0
next_cycle = applied_cycles + 1
effective_price = calculate_auto_price(
base_price = base_price,
auto_price_reduction = ad_cfg.auto_price_reduction,
target_reduction_cycle = next_cycle
)
if effective_price is None:
return
if effective_price == base_price:
# Still increment counter so small fractional reductions can accumulate over multiple cycles
ad_cfg.price_reduction_count = next_cycle
LOG.info(
_("Auto price reduction kept price %s after attempting %s reduction cycles"),
effective_price,
next_cycle
)
return
LOG.info(
_("Auto price reduction applied: %s -> %s after %s reduction cycles"),
base_price,
effective_price,
next_cycle
)
ad_cfg.price = effective_price
ad_cfg.price_reduction_count = next_cycle
# Note: price_reduction_count is persisted to ad_cfg_orig only after successful publish
class KleinanzeigenBot(WebScrapingMixin): class KleinanzeigenBot(WebScrapingMixin):
def __init__(self) -> None: def __init__(self) -> None:
@@ -364,7 +504,11 @@ class KleinanzeigenBot(WebScrapingMixin):
dicts.save_dict( dicts.save_dict(
self.config_file_path, self.config_file_path,
default_config.model_dump(exclude_none = True, exclude = {"ad_defaults": {"description"}}), default_config.model_dump(exclude_none = True, exclude = {"ad_defaults": {"description"}}),
header = "# yaml-language-server: $schema=https://raw.githubusercontent.com/Second-Hand-Friends/kleinanzeigen-bot/refs/heads/main/schemas/config.schema.json" header = (
"# yaml-language-server: $schema="
"https://raw.githubusercontent.com/Second-Hand-Friends/kleinanzeigen-bot"
"/refs/heads/main/schemas/config.schema.json"
)
) )
def load_config(self) -> None: def load_config(self) -> None:
@@ -571,6 +715,18 @@ class KleinanzeigenBot(WebScrapingMixin):
def load_ad(self, ad_cfg_orig:dict[str, Any]) -> Ad: 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 AdPartial.model_validate(ad_cfg_orig).to_ad(self.config.ad_defaults)
def __apply_auto_price_reduction(self, ad_cfg:Ad, _ad_cfg_orig:dict[str, Any], ad_file_relative:str) -> None:
"""Delegate to the module-level function."""
apply_auto_price_reduction(ad_cfg, _ad_cfg_orig, ad_file_relative)
def __repost_cycle_ready(self, ad_cfg:Ad, ad_file_relative:str) -> bool:
"""Delegate to the module-level function."""
return _repost_cycle_ready(ad_cfg, ad_file_relative)
def __day_delay_elapsed(self, ad_cfg:Ad, ad_file_relative:str) -> bool:
"""Delegate to the module-level function."""
return _day_delay_elapsed(ad_cfg, ad_file_relative)
async def check_and_wait_for_captcha(self, *, is_login_page:bool = True) -> None: async def check_and_wait_for_captcha(self, *, is_login_page:bool = True) -> None:
try: try:
captcha_timeout = self._timeout("captcha_detection") captcha_timeout = self._timeout("captcha_detection")
@@ -775,6 +931,15 @@ class KleinanzeigenBot(WebScrapingMixin):
if self.config.publishing.delete_old_ads == "BEFORE_PUBLISH" and not self.keep_old_ads: if self.config.publishing.delete_old_ads == "BEFORE_PUBLISH" and not self.keep_old_ads:
await self.delete_ad(ad_cfg, published_ads, delete_old_ads_by_title = self.config.publishing.delete_old_ads_by_title) await self.delete_ad(ad_cfg, published_ads, delete_old_ads_by_title = self.config.publishing.delete_old_ads_by_title)
# Apply auto price reduction only for REPLACE operations (actual reposts)
# This ensures price reductions only happen on republish, not on UPDATE
try:
ad_file_relative = str(Path(ad_file).relative_to(Path(self.config_file_path).parent))
except ValueError:
# On Windows, relative_to fails when paths are on different drives
ad_file_relative = ad_file
self.__apply_auto_price_reduction(ad_cfg, ad_cfg_orig, ad_file_relative)
LOG.info("Publishing ad '%s'...", ad_cfg.title) LOG.info("Publishing ad '%s'...", ad_cfg.title)
await self.web_open(f"{self.root_url}/p-anzeige-aufgeben-schritt2.html") await self.web_open(f"{self.root_url}/p-anzeige-aufgeben-schritt2.html")
else: else:
@@ -979,6 +1144,22 @@ class KleinanzeigenBot(WebScrapingMixin):
if not ad_cfg.created_on and not ad_cfg.id: if not ad_cfg.created_on and not ad_cfg.id:
ad_cfg_orig["created_on"] = ad_cfg_orig["updated_on"] ad_cfg_orig["created_on"] = ad_cfg_orig["updated_on"]
# Increment repost_count and persist price_reduction_count only for REPLACE operations (actual reposts)
# This ensures counters only advance on republish, not on UPDATE
if mode == AdUpdateStrategy.REPLACE:
# Increment repost_count after successful publish
# Note: This happens AFTER publish, so price reduction logic (which runs before publish)
# sees the count from the PREVIOUS run. This is intentional: the first publish uses
# repost_count=0 (no reduction), the second publish uses repost_count=1 (first reduction), etc.
current_reposts = int(ad_cfg_orig.get("repost_count", ad_cfg.repost_count or 0))
ad_cfg_orig["repost_count"] = current_reposts + 1
ad_cfg.repost_count = ad_cfg_orig["repost_count"]
# Persist price_reduction_count after successful publish
# This ensures failed publishes don't incorrectly increment the reduction counter
if ad_cfg.price_reduction_count is not None and ad_cfg.price_reduction_count > 0:
ad_cfg_orig["price_reduction_count"] = ad_cfg.price_reduction_count
if mode == AdUpdateStrategy.REPLACE: if mode == AdUpdateStrategy.REPLACE:
LOG.info(" -> SUCCESS: ad published with ID %s", ad_id) LOG.info(" -> SUCCESS: ad published with ID %s", ad_id)
else: else:

View File

@@ -4,18 +4,22 @@
from __future__ import annotations from __future__ import annotations
import hashlib, json # isort: skip import hashlib, json # isort: skip
from collections.abc import Mapping, Sequence
from datetime import datetime # noqa: TC003 Move import into a type-checking block from datetime import datetime # noqa: TC003 Move import into a type-checking block
from typing import Annotated, Any, Dict, Final, List, Literal, Mapping, Sequence from decimal import ROUND_CEILING, ROUND_HALF_UP, Decimal
from gettext import gettext as _
from typing import Annotated, Any, Final, Literal
from pydantic import AfterValidator, Field, field_validator, model_validator from pydantic import AfterValidator, Field, field_validator, model_validator
from typing_extensions import Self from typing_extensions import Self
from kleinanzeigen_bot.model.config_model import AdDefaults # noqa: TC001 Move application import into a type-checking block from kleinanzeigen_bot.model.config_model import AdDefaults, AutoPriceReductionConfig # noqa: TC001 Move application import into a type-checking block
from kleinanzeigen_bot.utils import dicts from kleinanzeigen_bot.utils import dicts
from kleinanzeigen_bot.utils.misc import parse_datetime, parse_decimal from kleinanzeigen_bot.utils.misc import parse_datetime, parse_decimal
from kleinanzeigen_bot.utils.pydantics import ContextualModel from kleinanzeigen_bot.utils.pydantics import ContextualModel
MAX_DESCRIPTION_LENGTH:Final[int] = 4000 MAX_DESCRIPTION_LENGTH:Final[int] = 4000
EURO_PRECISION:Final[Decimal] = Decimal("1")
def _OPTIONAL() -> Any: def _OPTIONAL() -> Any:
@@ -61,6 +65,45 @@ def _validate_shipping_option_item(v:str) -> str:
ShippingOption = Annotated[str, AfterValidator(_validate_shipping_option_item)] ShippingOption = Annotated[str, AfterValidator(_validate_shipping_option_item)]
def _validate_auto_price_reduction_constraints(
price:int | None,
auto_price_reduction:AutoPriceReductionConfig | dict[str, Any] | None
) -> None:
"""
Validate auto_price_reduction configuration constraints.
Raises ValueError if:
- auto_price_reduction is enabled but price is None
- min_price exceeds price
"""
if not auto_price_reduction:
return
# Handle both dict (from before validation) and AutoPriceReductionConfig (after validation)
if isinstance(auto_price_reduction, dict):
enabled = auto_price_reduction.get("enabled", False)
min_price = auto_price_reduction.get("min_price")
else:
enabled = auto_price_reduction.enabled
min_price = auto_price_reduction.min_price
if not enabled:
return
if price is None:
raise ValueError(_("price must be specified when auto_price_reduction is enabled"))
if min_price is not None:
try:
min_price_dec = Decimal(str(min_price))
price_dec = Decimal(str(price))
except Exception:
# Let Pydantic's type validation surface the underlying issue
return
if min_price_dec > price_dec:
raise ValueError(_("min_price must not exceed price"))
class AdPartial(ContextualModel): class AdPartial(ContextualModel):
active:bool | None = _OPTIONAL() active:bool | None = _OPTIONAL()
type:Literal["OFFER", "WANTED"] | None = _OPTIONAL() type:Literal["OFFER", "WANTED"] | None = _OPTIONAL()
@@ -69,14 +112,28 @@ class AdPartial(ContextualModel):
description_prefix:str | None = _OPTIONAL() description_prefix:str | None = _OPTIONAL()
description_suffix:str | None = _OPTIONAL() description_suffix:str | None = _OPTIONAL()
category:str category:str
special_attributes:Dict[str, str] | None = _OPTIONAL() special_attributes:dict[str, str] | None = _OPTIONAL()
price:int | None = _OPTIONAL() price:int | None = _OPTIONAL()
price_type:Literal["FIXED", "NEGOTIABLE", "GIVE_AWAY", "NOT_APPLICABLE"] | None = _OPTIONAL() price_type:Literal["FIXED", "NEGOTIABLE", "GIVE_AWAY", "NOT_APPLICABLE"] | None = _OPTIONAL()
auto_price_reduction:AutoPriceReductionConfig | None = Field(
default = None,
description = "automatic price reduction configuration"
)
repost_count:int = Field(
default = 0,
ge = 0,
description = "number of successful publications for this ad (persisted between runs)"
)
price_reduction_count:int = Field(
default = 0,
ge = 0,
description = "internal counter: number of automatic price reductions already applied"
)
shipping_type:Literal["PICKUP", "SHIPPING", "NOT_APPLICABLE"] | None = _OPTIONAL() shipping_type:Literal["PICKUP", "SHIPPING", "NOT_APPLICABLE"] | None = _OPTIONAL()
shipping_costs:float | None = _OPTIONAL() shipping_costs:float | None = _OPTIONAL()
shipping_options:List[ShippingOption] | None = _OPTIONAL() shipping_options:list[ShippingOption] | None = _OPTIONAL()
sell_directly:bool | None = _OPTIONAL() sell_directly:bool | None = _OPTIONAL()
images:List[str] | None = _OPTIONAL() images:list[str] | None = _OPTIONAL()
contact:ContactPartial | None = _OPTIONAL() contact:ContactPartial | None = _OPTIONAL()
republication_interval:int | None = _OPTIONAL() republication_interval:int | None = _OPTIONAL()
@@ -106,13 +163,19 @@ class AdPartial(ContextualModel):
@model_validator(mode = "before") @model_validator(mode = "before")
@classmethod @classmethod
def _validate_price_and_price_type(cls, values:Dict[str, Any]) -> Dict[str, Any]: def _validate_price_and_price_type(cls, values:dict[str, Any]) -> dict[str, Any]:
price_type = values.get("price_type") price_type = values.get("price_type")
price = values.get("price") price = values.get("price")
auto_price_reduction = values.get("auto_price_reduction")
if price_type == "GIVE_AWAY" and price is not None: if price_type == "GIVE_AWAY" and price is not None:
raise ValueError("price must not be specified when price_type is GIVE_AWAY") raise ValueError("price must not be specified when price_type is GIVE_AWAY")
if price_type == "FIXED" and price is None: if price_type == "FIXED" and price is None:
raise ValueError("price is required when price_type is FIXED") raise ValueError("price is required when price_type is FIXED")
# Validate auto_price_reduction configuration
_validate_auto_price_reduction_constraints(price, auto_price_reduction)
return values return values
def update_content_hash(self) -> Self: def update_content_hash(self) -> Self:
@@ -120,7 +183,14 @@ class AdPartial(ContextualModel):
# 1) Dump to a plain dict, excluding the metadata fields: # 1) Dump to a plain dict, excluding the metadata fields:
raw = self.model_dump( raw = self.model_dump(
exclude = {"id", "created_on", "updated_on", "content_hash"}, exclude = {
"id",
"created_on",
"updated_on",
"content_hash",
"repost_count",
"price_reduction_count",
},
exclude_none = True, exclude_none = True,
exclude_unset = True, exclude_unset = True,
) )
@@ -162,11 +232,70 @@ class AdPartial(ContextualModel):
target = ad_cfg, target = ad_cfg,
defaults = ad_defaults.model_dump(), defaults = ad_defaults.model_dump(),
ignore = lambda k, _: k == "description", # ignore legacy global description config ignore = lambda k, _: k == "description", # ignore legacy global description config
override = lambda _, v: not isinstance(v, list) and v in {None, ""} # noqa: PLC1901 can be simplified override = lambda _, v: (
not isinstance(v, list) and (v is None or (isinstance(v, str) and v == "")) # noqa: PLC1901
) )
)
# Ensure internal counters are integers (not user-configurable)
if not isinstance(ad_cfg.get("price_reduction_count"), int):
ad_cfg["price_reduction_count"] = 0
if not isinstance(ad_cfg.get("repost_count"), int):
ad_cfg["repost_count"] = 0
return Ad.model_validate(ad_cfg) return Ad.model_validate(ad_cfg)
def calculate_auto_price(
*,
base_price:int | float | None,
auto_price_reduction:AutoPriceReductionConfig | None,
target_reduction_cycle:int
) -> int | None:
"""
Calculate the effective price for the current run using commercial rounding.
Args:
base_price: original configured price used as the starting point.
auto_price_reduction: reduction configuration (enabled, strategy, amount, min_price, delays).
target_reduction_cycle: which reduction cycle to calculate the price for (0 = no reduction, 1 = first reduction, etc.).
Percentage reductions apply to the current price each cycle (compounded). Each reduction step is rounded
to full euros (commercial rounding with ROUND_HALF_UP) before the next reduction is applied.
Returns an int representing whole euros, or None when base_price is None.
"""
if base_price is None:
return None
price = Decimal(str(base_price))
if not auto_price_reduction or not auto_price_reduction.enabled or target_reduction_cycle <= 0:
return int(price.quantize(EURO_PRECISION, rounding = ROUND_HALF_UP))
if auto_price_reduction.strategy is None or auto_price_reduction.amount is None:
return int(price.quantize(EURO_PRECISION, rounding = ROUND_HALF_UP))
if auto_price_reduction.min_price is None:
raise ValueError(_("min_price must be specified when auto_price_reduction is enabled"))
# Prices are published as whole euros; ensure the configured floor cannot be undercut by int() conversion.
price_floor = Decimal(str(auto_price_reduction.min_price)).quantize(EURO_PRECISION, rounding = ROUND_CEILING)
repost_cycles = target_reduction_cycle
for _cycle in range(repost_cycles):
reduction_value = (
price * Decimal(str(auto_price_reduction.amount)) / Decimal("100")
if auto_price_reduction.strategy == "PERCENTAGE"
else Decimal(str(auto_price_reduction.amount))
)
price -= reduction_value
# Commercial rounding: round to full euros after each reduction step
price = price.quantize(EURO_PRECISION, rounding = ROUND_HALF_UP)
if price <= price_floor:
price = price_floor
break
return int(price)
# pyright: reportGeneralTypeIssues=false, reportIncompatibleVariableOverride=false # pyright: reportGeneralTypeIssues=false, reportIncompatibleVariableOverride=false
class Contact(ContactPartial): class Contact(ContactPartial):
name:str name:str
@@ -183,3 +312,12 @@ class Ad(AdPartial):
sell_directly:bool sell_directly:bool
contact:Contact contact:Contact
republication_interval:int republication_interval:int
auto_price_reduction:AutoPriceReductionConfig = Field(default_factory = AutoPriceReductionConfig)
price_reduction_count:int = 0
@model_validator(mode = "after")
def _validate_auto_price_config(self) -> "Ad":
# Validate the final Ad object after merging with defaults
# This ensures the merged configuration is valid even if raw YAML had None values
_validate_auto_price_reduction_constraints(self.price, self.auto_price_reduction)
return self

View File

@@ -4,7 +4,8 @@
from __future__ import annotations from __future__ import annotations
import copy import copy
from typing import Annotated, Any, List, Literal from gettext import gettext as _
from typing import Annotated, Any, Final, Literal
from pydantic import AfterValidator, Field, model_validator from pydantic import AfterValidator, Field, model_validator
from typing_extensions import deprecated from typing_extensions import deprecated
@@ -14,6 +15,52 @@ from kleinanzeigen_bot.utils import dicts
from kleinanzeigen_bot.utils.misc import get_attr from kleinanzeigen_bot.utils.misc import get_attr
from kleinanzeigen_bot.utils.pydantics import ContextualModel from kleinanzeigen_bot.utils.pydantics import ContextualModel
_MAX_PERCENTAGE:Final[int] = 100
class AutoPriceReductionConfig(ContextualModel):
enabled:bool = Field(
default = False,
description = "automatically lower the price of reposted ads"
)
strategy:Literal["FIXED", "PERCENTAGE"] | None = Field(
default = None,
description = "PERCENTAGE reduces by a percentage of the previous price, FIXED reduces by a fixed amount"
)
amount:float | None = Field(
default = None,
gt = 0,
description = "magnitude of the reduction; interpreted as percent for PERCENTAGE or currency units for FIXED"
)
min_price:float | None = Field(
default = None,
ge = 0,
description = "required when enabled is true; minimum price floor (use 0 for no lower bound)"
)
delay_reposts:int = Field(
default = 0,
ge = 0,
description = "number of reposts to wait before applying the first automatic price reduction"
)
delay_days:int = Field(
default = 0,
ge = 0,
description = "number of days to wait after publication before applying automatic price reductions"
)
@model_validator(mode = "after")
def _validate_config(self) -> "AutoPriceReductionConfig":
if self.enabled:
if self.strategy is None:
raise ValueError(_("strategy must be specified when auto_price_reduction is enabled"))
if self.amount is None:
raise ValueError(_("amount must be specified when auto_price_reduction is enabled"))
if self.min_price is None:
raise ValueError(_("min_price must be specified when auto_price_reduction is enabled"))
if self.strategy == "PERCENTAGE" and self.amount > _MAX_PERCENTAGE:
raise ValueError(_("Percentage reduction amount must not exceed %s") % _MAX_PERCENTAGE)
return self
class ContactDefaults(ContextualModel): class ContactDefaults(ContextualModel):
name:str | None = None name:str | None = None
@@ -35,9 +82,13 @@ class AdDefaults(ContextualModel):
description_prefix:str | None = Field(default = None, description = "prefix for the ad description") description_prefix:str | None = Field(default = None, description = "prefix for the ad description")
description_suffix:str | None = Field(default = None, description = " suffix for the ad description") description_suffix:str | None = Field(default = None, description = " suffix for the ad description")
price_type:Literal["FIXED", "NEGOTIABLE", "GIVE_AWAY", "NOT_APPLICABLE"] = "NEGOTIABLE" price_type:Literal["FIXED", "NEGOTIABLE", "GIVE_AWAY", "NOT_APPLICABLE"] = "NEGOTIABLE"
auto_price_reduction:AutoPriceReductionConfig = Field(
default_factory = AutoPriceReductionConfig,
description = "automatic price reduction configuration"
)
shipping_type:Literal["PICKUP", "SHIPPING", "NOT_APPLICABLE"] = "SHIPPING" shipping_type:Literal["PICKUP", "SHIPPING", "NOT_APPLICABLE"] = "SHIPPING"
sell_directly:bool = Field(default = False, description = "requires shipping_type SHIPPING to take effect") sell_directly:bool = Field(default = False, description = "requires shipping_type SHIPPING to take effect")
images:List[str] | None = Field(default = None) images:list[str] | None = Field(default = None)
contact:ContactDefaults = Field(default_factory = ContactDefaults) contact:ContactDefaults = Field(default_factory = ContactDefaults)
republication_interval:int = 7 republication_interval:int = 7
@@ -62,7 +113,7 @@ class DownloadConfig(ContextualModel):
default = False, default = False,
description = "if true, all shipping options matching the package size will be included" description = "if true, all shipping options matching the package size will be included"
) )
excluded_shipping_options:List[str] = Field( excluded_shipping_options:list[str] = Field(
default_factory = list, default_factory = list,
description = "list of shipping options to exclude, e.g. ['DHL_2', 'DHL_5']" description = "list of shipping options to exclude, e.g. ['DHL_2', 'DHL_5']"
) )
@@ -79,7 +130,7 @@ class DownloadConfig(ContextualModel):
class BrowserConfig(ContextualModel): class BrowserConfig(ContextualModel):
arguments:List[str] = Field( arguments:list[str] = Field(
default_factory = lambda: ["--user-data-dir=.temp/browser-profile"], default_factory = lambda: ["--user-data-dir=.temp/browser-profile"],
description = "See https://peter.sh/experiments/chromium-command-line-switches/" description = "See https://peter.sh/experiments/chromium-command-line-switches/"
) )
@@ -87,7 +138,7 @@ class BrowserConfig(ContextualModel):
default = None, default = None,
description = "path to custom browser executable, if not specified will be looked up on PATH" description = "path to custom browser executable, if not specified will be looked up on PATH"
) )
extensions:List[str] = Field( extensions:list[str] = Field(
default_factory = list, default_factory = list,
description = "a list of .crx extension files to be loaded" description = "a list of .crx extension files to be loaded"
) )
@@ -175,7 +226,7 @@ GlobPattern = Annotated[str, AfterValidator(_validate_glob_pattern)]
class Config(ContextualModel): class Config(ContextualModel):
ad_files:List[GlobPattern] = Field( ad_files:list[GlobPattern] = Field(
default_factory = lambda: ["./**/ad_*.{json,yml,yaml}"], default_factory = lambda: ["./**/ad_*.{json,yml,yaml}"],
min_items = 1, min_items = 1,
description = """ description = """

View File

@@ -96,6 +96,17 @@ kleinanzeigen_bot/__init__.py:
"Skipping because ad is reserved": "Überspringen, da Anzeige reserviert ist" "Skipping because ad is reserved": "Überspringen, da Anzeige reserviert ist"
"DONE: (Re-)published %s": "FERTIG: %s (erneut) veröffentlicht" "DONE: (Re-)published %s": "FERTIG: %s (erneut) veröffentlicht"
"ad": "Anzeige" "ad": "Anzeige"
apply_auto_price_reduction:
"Auto price reduction is enabled for [%s] but no price is configured.": "Automatische Preisreduzierung ist für [%s] aktiviert, aber es wurde kein Preis konfiguriert."
"Auto price reduction is enabled for [%s] but min_price equals price (%s) - no reductions will occur.": "Automatische Preisreduzierung ist für [%s] aktiviert, aber min_price entspricht dem Preis (%s) - es werden keine Reduktionen auftreten."
"Auto price reduction applied: %s -> %s after %s reduction cycles": "Automatische Preisreduzierung angewendet: %s -> %s nach %s Reduktionszyklen"
"Auto price reduction kept price %s after attempting %s reduction cycles": "Automatische Preisreduzierung hat Preis %s beibehalten nach dem Versuch von %s Reduktionszyklen"
_repost_cycle_ready:
"Auto price reduction delayed for [%s]: waiting %s more reposts (completed %s, applied %s reductions)": "Automatische Preisreduzierung für [%s] verzögert: Warte %s weitere erneute Veröffentlichungen (abgeschlossen %s, angewendet %s Reduktionen)"
"Auto price reduction already applied for [%s]: %s reductions match %s eligible reposts": "Automatische Preisreduzierung für [%s] bereits angewendet: %s Reduktionen entsprechen %s berechtigten erneuten Veröffentlichungen"
_day_delay_elapsed:
"Auto price reduction delayed for [%s]: waiting %s days (elapsed %s)": "Automatische Preisreduzierung für [%s] verzögert: Warte %s Tage (vergangen %s)"
"Auto price reduction delayed for [%s]: waiting %s days but publish timestamp missing": "Automatische Preisreduzierung für [%s] verzögert: Warte %s Tage, aber Zeitstempel der Veröffentlichung fehlt"
publish_ad: publish_ad:
"Publishing ad '%s'...": "Veröffentliche Anzeige '%s'..." "Publishing ad '%s'...": "Veröffentliche Anzeige '%s'..."
@@ -531,6 +542,24 @@ kleinanzeigen_bot/update_checker.py:
? "You are on a different commit than the release for channel '%s' (tag: %s). This may mean you are ahead, behind, or on a different branch. Local commit: %s (%s UTC), Release commit: %s (%s UTC)" ? "You are on a different commit than the release for channel '%s' (tag: %s). This may mean you are ahead, behind, or on a different branch. Local commit: %s (%s UTC), Release commit: %s (%s UTC)"
: "Sie befinden sich auf einem anderen Commit als das Release für Kanal '%s' (Tag: %s). Dies kann bedeuten, dass Sie voraus, hinterher oder auf einem anderen Branch sind. Lokaler Commit: %s (%s UTC), Release-Commit: %s (%s UTC)" : "Sie befinden sich auf einem anderen Commit als das Release für Kanal '%s' (Tag: %s). Dies kann bedeuten, dass Sie voraus, hinterher oder auf einem anderen Branch sind. Lokaler Commit: %s (%s UTC), Release-Commit: %s (%s UTC)"
#################################################
kleinanzeigen_bot/model/config_model.py:
#################################################
_validate_config:
"strategy must be specified when auto_price_reduction is enabled": "strategy muss angegeben werden, wenn auto_price_reduction aktiviert ist"
"amount must be specified when auto_price_reduction is enabled": "amount muss angegeben werden, wenn auto_price_reduction aktiviert ist"
"min_price must be specified when auto_price_reduction is enabled": "min_price muss angegeben werden, wenn auto_price_reduction aktiviert ist"
"Percentage reduction amount must not exceed %s": "Prozentuale Reduktionsmenge darf %s nicht überschreiten"
#################################################
kleinanzeigen_bot/model/ad_model.py:
#################################################
_validate_auto_price_reduction_constraints:
"price must be specified when auto_price_reduction is enabled": "price muss angegeben werden, wenn auto_price_reduction aktiviert ist"
"min_price must not exceed price": "min_price darf price nicht überschreiten"
calculate_auto_price:
"min_price must be specified when auto_price_reduction is enabled": "min_price muss angegeben werden, wenn auto_price_reduction aktiviert ist"
################################################# #################################################
kleinanzeigen_bot/model/update_check_state.py: kleinanzeigen_bot/model/update_check_state.py:
################################################# #################################################

View File

@@ -192,6 +192,14 @@ def __get_message_template(error_code:str) -> str | None:
case "decimal_max_digits": return _("Decimal input should have no more than {max_digits} digit{expected_plural} in total") case "decimal_max_digits": return _("Decimal input should have no more than {max_digits} digit{expected_plural} in total")
case "decimal_max_places": return _("Decimal input should have no more than {decimal_places} decimal place{expected_plural}") case "decimal_max_places": return _("Decimal input should have no more than {decimal_places} decimal place{expected_plural}")
case "decimal_whole_digits": return _("Decimal input should have no more than {whole_digits} digit{expected_plural} before the decimal point") case "decimal_whole_digits": return _("Decimal input should have no more than {whole_digits} digit{expected_plural} before the decimal point")
case "complex_type": return _("Input should be a valid python complex object, a number, or a valid complex string following the rules at https://docs.python.org/3/library/functions.html#complex") case "complex_type":
case "complex_str_parsing": return _("Input should be a valid complex string following the rules at https://docs.python.org/3/library/functions.html#complex") return _(
"Input should be a valid python complex object, a number, or a valid complex string "
"following the rules at https://docs.python.org/3/library/functions.html#complex"
)
case "complex_str_parsing":
return _(
"Input should be a valid complex string following the rules at "
"https://docs.python.org/3/library/functions.html#complex"
)
case _: return None case _: return None

View File

@@ -1,11 +1,18 @@
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors # SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later # SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ # SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
from __future__ import annotations
import math import math
from kleinanzeigen_bot.model.ad_model import AdPartial import pytest
from kleinanzeigen_bot.model.ad_model import MAX_DESCRIPTION_LENGTH, Ad, AdPartial, ShippingOption, calculate_auto_price
from kleinanzeigen_bot.model.config_model import AdDefaults, AutoPriceReductionConfig
from kleinanzeigen_bot.utils.pydantics import ContextualModel, ContextualValidationError
@pytest.mark.unit
def test_update_content_hash() -> None: def test_update_content_hash() -> None:
minimal_ad_cfg = { minimal_ad_cfg = {
"id": "123456789", "id": "123456789",
@@ -41,6 +48,37 @@ def test_update_content_hash() -> None:
}).update_content_hash().content_hash != minimal_ad_cfg_hash }).update_content_hash().content_hash != minimal_ad_cfg_hash
@pytest.mark.unit
def test_price_reduction_count_does_not_influence_content_hash() -> None:
base_ad_cfg = {
"id": "123456789",
"title": "Test Ad Title",
"category": "160",
"description": "Test Description",
"price_type": "NEGOTIABLE",
}
hash_without_reposts = AdPartial.model_validate(base_ad_cfg | {"price_reduction_count": 0}).update_content_hash().content_hash
hash_with_reposts = AdPartial.model_validate(base_ad_cfg | {"price_reduction_count": 5}).update_content_hash().content_hash
assert hash_without_reposts == hash_with_reposts
@pytest.mark.unit
def test_repost_count_does_not_influence_content_hash() -> None:
base_ad_cfg = {
"id": "123456789",
"title": "Test Ad Title",
"category": "160",
"description": "Test Description",
"price_type": "NEGOTIABLE",
}
hash_without_reposts = AdPartial.model_validate(base_ad_cfg | {"repost_count": 0}).update_content_hash().content_hash
hash_with_reposts = AdPartial.model_validate(base_ad_cfg | {"repost_count": 5}).update_content_hash().content_hash
assert hash_without_reposts == hash_with_reposts
@pytest.mark.unit
def test_shipping_costs() -> None: def test_shipping_costs() -> None:
minimal_ad_cfg = { minimal_ad_cfg = {
"id": "123456789", "id": "123456789",
@@ -60,3 +98,337 @@ def test_shipping_costs() -> None:
assert AdPartial.model_validate(minimal_ad_cfg | {"shipping_costs": " "}).shipping_costs is None assert AdPartial.model_validate(minimal_ad_cfg | {"shipping_costs": " "}).shipping_costs is None
assert AdPartial.model_validate(minimal_ad_cfg | {"shipping_costs": None}).shipping_costs is None assert AdPartial.model_validate(minimal_ad_cfg | {"shipping_costs": None}).shipping_costs is None
assert AdPartial.model_validate(minimal_ad_cfg).shipping_costs is None assert AdPartial.model_validate(minimal_ad_cfg).shipping_costs is None
class ShippingOptionWrapper(ContextualModel):
option:ShippingOption
@pytest.mark.unit
def test_shipping_option_must_not_be_blank() -> None:
with pytest.raises(ContextualValidationError, match = "must be non-empty and non-blank"):
ShippingOptionWrapper.model_validate({"option": " "})
@pytest.mark.unit
def test_description_length_limit() -> None:
cfg = {
"title": "Description Length",
"category": "160",
"description": "x" * (MAX_DESCRIPTION_LENGTH + 1)
}
with pytest.raises(ContextualValidationError, match = f"description length exceeds {MAX_DESCRIPTION_LENGTH} characters"):
AdPartial.model_validate(cfg)
@pytest.fixture
def base_ad_cfg() -> dict[str, object]:
return {
"title": "Test Ad Title",
"category": "160",
"description": "Test Description",
"price_type": "NEGOTIABLE",
"contact": {"name": "Test User", "zipcode": "12345"},
"shipping_type": "PICKUP",
"sell_directly": False,
"type": "OFFER",
"active": True
}
@pytest.fixture
def complete_ad_cfg(base_ad_cfg:dict[str, object]) -> dict[str, object]:
return base_ad_cfg | {
"republication_interval": 7,
"price": 100,
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 5,
"min_price": 50,
"delay_reposts": 0,
"delay_days": 0
}
}
class SparseDumpAdPartial(AdPartial):
def model_dump(self, *args:object, **kwargs:object) -> dict[str, object]:
data = super().model_dump(*args, **kwargs) # type: ignore[arg-type]
data.pop("price_reduction_count", None)
data.pop("repost_count", None)
return data
@pytest.mark.unit
def test_auto_reduce_requires_price(base_ad_cfg:dict[str, object]) -> None:
cfg = base_ad_cfg.copy() | {
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 5,
"min_price": 50
}
}
with pytest.raises(ContextualValidationError, match = "price must be specified"):
AdPartial.model_validate(cfg).to_ad(AdDefaults())
@pytest.mark.unit
def test_auto_reduce_requires_strategy(base_ad_cfg:dict[str, object]) -> None:
cfg = base_ad_cfg.copy() | {
"price": 100,
"auto_price_reduction": {
"enabled": True,
"min_price": 50
}
}
with pytest.raises(ContextualValidationError, match = "strategy must be specified"):
AdPartial.model_validate(cfg).to_ad(AdDefaults())
@pytest.mark.unit
def test_prepare_ad_model_fills_missing_counters(base_ad_cfg:dict[str, object]) -> None:
cfg = base_ad_cfg.copy() | {
"price": 120,
"shipping_type": "SHIPPING",
"sell_directly": False
}
ad = AdPartial.model_validate(cfg).to_ad(AdDefaults())
assert ad.auto_price_reduction.delay_reposts == 0
assert ad.auto_price_reduction.delay_days == 0
assert ad.price_reduction_count == 0
assert ad.repost_count == 0
@pytest.mark.unit
def test_min_price_must_not_exceed_price(base_ad_cfg:dict[str, object]) -> None:
cfg = base_ad_cfg.copy() | {
"price": 100,
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 5,
"min_price": 120
}
}
with pytest.raises(ContextualValidationError, match = "min_price must not exceed price"):
AdPartial.model_validate(cfg)
@pytest.mark.unit
def test_min_price_validation_defers_to_pydantic_for_invalid_types(base_ad_cfg:dict[str, object]) -> None:
# Test that invalid price/min_price types are handled gracefully
# The safe Decimal comparison should catch conversion errors and defer to Pydantic
cfg = base_ad_cfg.copy() | {
"price": "not_a_number",
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 5,
"min_price": 100
}
}
# Should raise Pydantic validation error for invalid price type, not our custom validation error
with pytest.raises(ContextualValidationError):
AdPartial.model_validate(cfg)
# Test with invalid min_price type
cfg2 = base_ad_cfg.copy() | {
"price": 100,
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 5,
"min_price": "invalid"
}
}
# Should raise Pydantic validation error for invalid min_price type
with pytest.raises(ContextualValidationError):
AdPartial.model_validate(cfg2)
@pytest.mark.unit
def test_auto_reduce_requires_min_price(base_ad_cfg:dict[str, object]) -> None:
cfg = base_ad_cfg.copy() | {
"price": 100,
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 5
}
}
with pytest.raises(ContextualValidationError, match = "min_price must be specified"):
AdPartial.model_validate(cfg).to_ad(AdDefaults())
@pytest.mark.unit
def test_to_ad_stabilizes_counters_when_defaults_omit(base_ad_cfg:dict[str, object]) -> None:
cfg = base_ad_cfg.copy() | {
"republication_interval": 7,
"price": 120
}
ad = AdPartial.model_validate(cfg).to_ad(AdDefaults())
assert ad.auto_price_reduction.delay_reposts == 0
assert ad.auto_price_reduction.delay_days == 0
assert ad.price_reduction_count == 0
assert ad.repost_count == 0
@pytest.mark.unit
def test_to_ad_sets_zero_when_counts_missing_from_dump(base_ad_cfg:dict[str, object]) -> None:
cfg = base_ad_cfg.copy() | {
"republication_interval": 7,
"price": 130
}
ad = SparseDumpAdPartial.model_validate(cfg).to_ad(AdDefaults())
assert ad.price_reduction_count == 0
assert ad.repost_count == 0
@pytest.mark.unit
def test_ad_model_auto_reduce_requires_price(complete_ad_cfg:dict[str, object]) -> None:
cfg = complete_ad_cfg.copy() | {"price": None}
with pytest.raises(ContextualValidationError, match = "price must be specified"):
Ad.model_validate(cfg)
@pytest.mark.unit
def test_ad_model_auto_reduce_requires_strategy(complete_ad_cfg:dict[str, object]) -> None:
cfg_copy = complete_ad_cfg.copy()
cfg_copy["auto_price_reduction"] = {
"enabled": True,
"min_price": 50
}
with pytest.raises(ContextualValidationError, match = "strategy must be specified"):
Ad.model_validate(cfg_copy)
@pytest.mark.unit
def test_price_reduction_delay_inherited_from_defaults(complete_ad_cfg:dict[str, object]) -> None:
# When auto_price_reduction is not specified in ad config, it inherits from defaults
cfg = complete_ad_cfg.copy()
cfg.pop("auto_price_reduction", None) # Remove to inherit from defaults
defaults = AdDefaults(
auto_price_reduction = AutoPriceReductionConfig(
enabled = True,
strategy = "FIXED",
amount = 5,
min_price = 50,
delay_reposts = 4,
delay_days = 0
)
)
ad = AdPartial.model_validate(cfg).to_ad(defaults)
assert ad.auto_price_reduction.delay_reposts == 4
@pytest.mark.unit
def test_price_reduction_delay_override_zero(complete_ad_cfg:dict[str, object]) -> None:
cfg = complete_ad_cfg.copy()
# Type-safe way to modify nested dict
cfg["auto_price_reduction"] = {
"enabled": True,
"strategy": "FIXED",
"amount": 5,
"min_price": 50,
"delay_reposts": 0,
"delay_days": 0
}
defaults = AdDefaults(
auto_price_reduction = AutoPriceReductionConfig(
enabled = True,
strategy = "FIXED",
amount = 5,
min_price = 50,
delay_reposts = 4,
delay_days = 0
)
)
ad = AdPartial.model_validate(cfg).to_ad(defaults)
assert ad.auto_price_reduction.delay_reposts == 0
@pytest.mark.unit
def test_ad_model_auto_reduce_requires_min_price(complete_ad_cfg:dict[str, object]) -> None:
cfg_copy = complete_ad_cfg.copy()
cfg_copy["auto_price_reduction"] = {
"enabled": True,
"strategy": "FIXED",
"amount": 5
}
with pytest.raises(ContextualValidationError, match = "min_price must be specified"):
Ad.model_validate(cfg_copy)
@pytest.mark.unit
def test_ad_model_min_price_must_not_exceed_price(complete_ad_cfg:dict[str, object]) -> None:
cfg_copy = complete_ad_cfg.copy()
cfg_copy["price"] = 100
cfg_copy["auto_price_reduction"] = {
"enabled": True,
"strategy": "FIXED",
"amount": 5,
"min_price": 150,
"delay_reposts": 0,
"delay_days": 0
}
with pytest.raises(ContextualValidationError, match = "min_price must not exceed price"):
Ad.model_validate(cfg_copy)
@pytest.mark.unit
def test_calculate_auto_price_with_missing_strategy() -> None:
"""Test calculate_auto_price when strategy is None but enabled is True (defensive check)"""
# Use model_construct to bypass validation and reach defensive lines 234-235
config = AutoPriceReductionConfig.model_construct(
enabled = True, strategy = None, amount = None, min_price = 50
)
result = calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 1
)
assert result == 100 # Should return base price when strategy is None
@pytest.mark.unit
def test_calculate_auto_price_with_missing_amount() -> None:
"""Test calculate_auto_price when amount is None but enabled is True (defensive check)"""
# Use model_construct to bypass validation and reach defensive lines 234-235
config = AutoPriceReductionConfig.model_construct(
enabled = True, strategy = "FIXED", amount = None, min_price = 50
)
result = calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 1
)
assert result == 100 # Should return base price when amount is None
@pytest.mark.unit
def test_calculate_auto_price_raises_when_min_price_none_and_enabled() -> None:
"""Test that calculate_auto_price raises ValueError when min_price is None during calculation (defensive check)"""
# Use model_construct to bypass validation and reach defensive line 237-238
config = AutoPriceReductionConfig.model_construct(
enabled = True, strategy = "FIXED", amount = 10, min_price = None
)
with pytest.raises(ValueError, match = "min_price must be specified when auto_price_reduction is enabled"):
calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 1
)
@pytest.mark.unit
def test_auto_price_reduction_config_requires_amount_when_enabled() -> None:
"""Test AutoPriceReductionConfig validator requires amount when enabled"""
with pytest.raises(ValueError, match = "amount must be specified when auto_price_reduction is enabled"):
AutoPriceReductionConfig(enabled = True, strategy = "FIXED", amount = None, min_price = 50)

View File

@@ -5,7 +5,7 @@ import copy, io, json, logging, os, tempfile # isort: skip
from collections.abc import Generator from collections.abc import Generator
from contextlib import redirect_stdout from contextlib import redirect_stdout
from datetime import timedelta from datetime import timedelta
from pathlib import Path from pathlib import Path, PureWindowsPath
from typing import Any, cast from typing import Any, cast
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
@@ -1072,6 +1072,124 @@ class TestKleinanzeigenBotShippingOptions:
# Verify the file was created in the temporary directory # Verify the file was created in the temporary directory
assert ad_file.exists() assert ad_file.exists()
@pytest.mark.asyncio
async def test_cross_drive_path_fallback_windows(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None:
"""Test that cross-drive path handling falls back to absolute path on Windows."""
# Create ad config
ad_cfg = Ad.model_validate(base_ad_config | {
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00",
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 10,
"min_price": 50,
"delay_reposts": 0,
"delay_days": 0
},
"price": 100,
"repost_count": 1,
"price_reduction_count": 0
})
ad_cfg.update_content_hash()
ad_cfg_orig = ad_cfg.model_dump()
# Simulate Windows cross-drive scenario
# Config on D:, ad file on C:
test_bot.config_file_path = "D:\\project\\config.yaml"
ad_file = "C:\\temp\\test_ad.yaml"
# Create a sentinel exception to abort publish_ad early
class _SentinelException(Exception):
pass
# Track what path argument __apply_auto_price_reduction receives
recorded_path:list[str] = []
def mock_apply_auto_price_reduction(ad_cfg:Ad, ad_cfg_orig:dict[str, Any], ad_file_relative:str) -> None:
recorded_path.append(ad_file_relative)
raise _SentinelException("Abort early for test")
# Mock Path to use PureWindowsPath for testing cross-drive behavior
with patch("kleinanzeigen_bot.Path", PureWindowsPath), \
patch.object(test_bot, "_KleinanzeigenBot__apply_auto_price_reduction", side_effect = mock_apply_auto_price_reduction), \
patch.object(test_bot, "web_open", new_callable = AsyncMock), \
patch.object(test_bot, "delete_ad", new_callable = AsyncMock):
# Call publish_ad and expect sentinel exception
try:
await test_bot.publish_ad(ad_file, ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.REPLACE)
pytest.fail("Expected _SentinelException to be raised")
except _SentinelException:
# This is expected - the test aborts early
pass
# Verify the path argument is the absolute path (fallback behavior)
assert len(recorded_path) == 1
assert recorded_path[0] == ad_file, f"Expected absolute path fallback, got: {recorded_path[0]}"
@pytest.mark.asyncio
async def test_auto_price_reduction_only_on_replace_not_update(
self,
test_bot:KleinanzeigenBot,
base_ad_config:dict[str, Any],
tmp_path:Path
) -> None:
"""Test that auto price reduction is ONLY applied on REPLACE mode, not UPDATE."""
# Create ad with auto price reduction enabled
ad_cfg = Ad.model_validate(base_ad_config | {
"id": 12345,
"price": 200,
"auto_price_reduction": {
"enabled": True,
"strategy": "FIXED",
"amount": 50,
"min_price": 50,
"delay_reposts": 0,
"delay_days": 0
},
"repost_count": 1,
"price_reduction_count": 0,
"updated_on": "2024-01-01T00:00:00",
"created_on": "2024-01-01T00:00:00"
})
ad_cfg.update_content_hash()
ad_cfg_orig = ad_cfg.model_dump()
# Mock the private __apply_auto_price_reduction method
with patch.object(test_bot, "_KleinanzeigenBot__apply_auto_price_reduction") as mock_apply:
# Mock other dependencies
mock_response = {"statusCode": 200, "statusMessage": "OK", "content": "{}"}
with patch.object(test_bot, "web_find", new_callable = AsyncMock), \
patch.object(test_bot, "web_input", new_callable = AsyncMock), \
patch.object(test_bot, "web_click", new_callable = AsyncMock), \
patch.object(test_bot, "web_open", new_callable = AsyncMock), \
patch.object(test_bot, "web_select", new_callable = AsyncMock), \
patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = False), \
patch.object(test_bot, "web_await", new_callable = AsyncMock), \
patch.object(test_bot, "web_sleep", new_callable = AsyncMock), \
patch.object(test_bot, "web_execute", new_callable = AsyncMock, return_value = mock_response), \
patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = mock_response), \
patch.object(test_bot, "web_scroll_page_down", new_callable = AsyncMock), \
patch.object(test_bot, "web_find_all", new_callable = AsyncMock, return_value = []), \
patch.object(test_bot, "check_and_wait_for_captcha", new_callable = AsyncMock), \
patch("builtins.input", return_value = ""), \
patch("kleinanzeigen_bot.utils.misc.ainput", new_callable = AsyncMock, return_value = ""):
test_bot.page = MagicMock()
test_bot.page.url = "https://www.kleinanzeigen.de/p-anzeige-aufgeben-bestaetigung.html?adId=12345"
test_bot.config.publishing.delete_old_ads = "BEFORE_PUBLISH"
# Test REPLACE mode - should call __apply_auto_price_reduction
await test_bot.publish_ad(str(tmp_path / "ad.yaml"), ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.REPLACE)
assert mock_apply.call_count == 1, "Auto price reduction should be called on REPLACE"
# Reset mock
mock_apply.reset_mock()
# Test MODIFY mode - should NOT call __apply_auto_price_reduction
await test_bot.publish_ad(str(tmp_path / "ad.yaml"), ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.MODIFY)
assert mock_apply.call_count == 0, "Auto price reduction should NOT be called on MODIFY"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_special_attributes_with_non_string_values(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: async def test_special_attributes_with_non_string_values(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None:
"""Test that special attributes with non-string values are converted to strings.""" """Test that special attributes with non-string values are converted to strings."""
@@ -1462,3 +1580,45 @@ def test_file_logger_writes_message(tmp_path:Path, caplog:pytest.LogCaptureFixtu
with open(log_path, "r", encoding = "utf-8") as f: with open(log_path, "r", encoding = "utf-8") as f:
contents = f.read() contents = f.read()
assert "Logger test log message" in contents assert "Logger test log message" in contents
class TestPriceReductionPersistence:
"""Tests for price_reduction_count persistence logic."""
@pytest.mark.unit
def test_persistence_logic_saves_when_count_positive(self) -> None:
"""Test the conditional logic that decides whether to persist price_reduction_count."""
# Simulate the logic from publish_ad lines 1076-1079
ad_cfg_orig:dict[str, Any] = {}
# Test case 1: price_reduction_count = 3 (should persist)
price_reduction_count = 3
if price_reduction_count is not None and price_reduction_count > 0:
ad_cfg_orig["price_reduction_count"] = price_reduction_count
assert "price_reduction_count" in ad_cfg_orig
assert ad_cfg_orig["price_reduction_count"] == 3
@pytest.mark.unit
def test_persistence_logic_skips_when_count_zero(self) -> None:
"""Test that price_reduction_count == 0 does not get persisted."""
ad_cfg_orig:dict[str, Any] = {}
# Test case 2: price_reduction_count = 0 (should NOT persist)
price_reduction_count = 0
if price_reduction_count is not None and price_reduction_count > 0:
ad_cfg_orig["price_reduction_count"] = price_reduction_count
assert "price_reduction_count" not in ad_cfg_orig
@pytest.mark.unit
def test_persistence_logic_skips_when_count_none(self) -> None:
"""Test that price_reduction_count == None does not get persisted."""
ad_cfg_orig:dict[str, Any] = {}
# Test case 3: price_reduction_count = None (should NOT persist)
price_reduction_count = None
if price_reduction_count is not None and price_reduction_count > 0:
ad_cfg_orig["price_reduction_count"] = price_reduction_count
assert "price_reduction_count" not in ad_cfg_orig

View File

@@ -0,0 +1,550 @@
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import logging
from datetime import datetime, timedelta, timezone
from gettext import gettext as _
from types import SimpleNamespace
from typing import Any, Protocol, runtime_checkable
import pytest
import kleinanzeigen_bot
from kleinanzeigen_bot.model.ad_model import calculate_auto_price
from kleinanzeigen_bot.model.config_model import AutoPriceReductionConfig
from kleinanzeigen_bot.utils.pydantics import ContextualValidationError
@runtime_checkable
class _ApplyAutoPriceReduction(Protocol):
def __call__(self, ad_cfg:SimpleNamespace, ad_cfg_orig:dict[str, Any], ad_file_relative:str) -> None:
...
@pytest.fixture
def apply_auto_price_reduction() -> _ApplyAutoPriceReduction:
# Return the module-level function directly (no more name-mangling!)
return kleinanzeigen_bot.apply_auto_price_reduction # type: ignore[return-value]
@pytest.mark.unit
def test_initial_posting_uses_base_price() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 50)
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 0
) == 100
@pytest.mark.unit
def test_auto_price_returns_none_without_base_price() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 10)
assert calculate_auto_price(
base_price = None,
auto_price_reduction = config,
target_reduction_cycle = 3
) is None
@pytest.mark.unit
def test_negative_price_reduction_count_is_treated_like_zero() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 25, min_price = 50)
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = -3
) == 100
@pytest.mark.unit
def test_missing_price_reduction_returns_base_price() -> None:
assert calculate_auto_price(
base_price = 150,
auto_price_reduction = None,
target_reduction_cycle = 4
) == 150
@pytest.mark.unit
def test_percentage_reduction_on_float_rounds_half_up() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 12.5, min_price = 50)
assert calculate_auto_price(
base_price = 99.99,
auto_price_reduction = config,
target_reduction_cycle = 1
) == 87
@pytest.mark.unit
def test_fixed_reduction_on_float_rounds_half_up() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "FIXED", amount = 12.4, min_price = 50)
assert calculate_auto_price(
base_price = 80.51,
auto_price_reduction = config,
target_reduction_cycle = 1
) == 68
@pytest.mark.unit
def test_percentage_price_reduction_over_time() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 50)
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 1
) == 90
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 2
) == 81
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 3
) == 73
@pytest.mark.unit
def test_fixed_price_reduction_over_time() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "FIXED", amount = 15, min_price = 50)
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 1
) == 85
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 2
) == 70
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 3
) == 55
@pytest.mark.unit
def test_min_price_boundary_is_respected() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "FIXED", amount = 20, min_price = 50)
assert calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 5
) == 50
@pytest.mark.unit
def test_min_price_zero_is_allowed() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "FIXED", amount = 5, min_price = 0)
assert calculate_auto_price(
base_price = 20,
auto_price_reduction = config,
target_reduction_cycle = 5
) == 0
@pytest.mark.unit
def test_missing_min_price_raises_error() -> None:
# min_price validation happens at config initialization when enabled=True
with pytest.raises(ContextualValidationError, match = "min_price must be specified"):
AutoPriceReductionConfig.model_validate({"enabled": True, "strategy": "PERCENTAGE", "amount": 50, "min_price": None})
@pytest.mark.unit
def test_percentage_above_100_raises_error() -> None:
with pytest.raises(ContextualValidationError, match = "Percentage reduction amount must not exceed 100"):
AutoPriceReductionConfig.model_validate({"enabled": True, "strategy": "PERCENTAGE", "amount": 150, "min_price": 50})
@pytest.mark.unit
def test_feature_disabled_path_leaves_price_unchanged() -> None:
config = AutoPriceReductionConfig(enabled = False, strategy = "PERCENTAGE", amount = 25, min_price = 50)
price = calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 4
)
assert price == 100
@pytest.mark.unit
def test_apply_auto_price_reduction_logs_drop(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = 200,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 25, min_price = 50, delay_reposts = 0, delay_days = 0
),
price_reduction_count = 0,
repost_count = 1
)
ad_orig:dict[str, Any] = {}
with caplog.at_level(logging.INFO):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_test.yaml")
expected = _("Auto price reduction applied: %s -> %s after %s reduction cycles") % (200, 150, 1)
assert any(expected in message for message in caplog.messages)
assert ad_cfg.price == 150
assert ad_cfg.price_reduction_count == 1
# Note: price_reduction_count is NOT persisted to ad_orig until after successful publish
assert "price_reduction_count" not in ad_orig
@pytest.mark.unit
def test_apply_auto_price_reduction_logs_unchanged_price_at_floor(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
# Test scenario: price has been reduced to just above min_price,
# and the next reduction would drop it below, so it gets clamped
ad_cfg = SimpleNamespace(
price = 95,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "FIXED", amount = 10, min_price = 90, delay_reposts = 0, delay_days = 0
),
price_reduction_count = 0,
repost_count = 1
)
ad_orig:dict[str, Any] = {}
with caplog.at_level(logging.INFO):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_test.yaml")
# Price: 95 - 10 = 85, clamped to 90 (floor)
# So the effective price is 90, not 95, meaning reduction was applied
expected = _("Auto price reduction applied: %s -> %s after %s reduction cycles") % (95, 90, 1)
assert any(expected in message for message in caplog.messages)
assert ad_cfg.price == 90
assert ad_cfg.price_reduction_count == 1
assert "price_reduction_count" not in ad_orig
@pytest.mark.unit
def test_apply_auto_price_reduction_warns_when_price_missing(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = None,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 25, min_price = 10, delay_reposts = 0, delay_days = 0
),
price_reduction_count = 2,
repost_count = 2
)
ad_orig:dict[str, Any] = {}
with caplog.at_level(logging.WARNING):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_warning.yaml")
expected = _("Auto price reduction is enabled for [%s] but no price is configured.") % ("ad_warning.yaml",)
assert any(expected in message for message in caplog.messages)
assert ad_cfg.price is None
@pytest.mark.unit
def test_apply_auto_price_reduction_warns_when_min_price_equals_price(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = 100,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 25, min_price = 100, delay_reposts = 0, delay_days = 0
),
price_reduction_count = 0,
repost_count = 1
)
ad_orig:dict[str, Any] = {}
with caplog.at_level(logging.WARNING):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_equal_prices.yaml")
expected = _("Auto price reduction is enabled for [%s] but min_price equals price (%s) - no reductions will occur.") % ("ad_equal_prices.yaml", 100)
assert any(expected in message for message in caplog.messages)
assert ad_cfg.price == 100
assert ad_cfg.price_reduction_count == 0
@pytest.mark.unit
def test_apply_auto_price_reduction_respects_repost_delay(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = 200,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 25, min_price = 50, delay_reposts = 3, delay_days = 0
),
price_reduction_count = 0,
repost_count = 2
)
ad_orig:dict[str, Any] = {}
with caplog.at_level(logging.INFO):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_delay.yaml")
assert ad_cfg.price == 200
delayed_message = _("Auto price reduction delayed for [%s]: waiting %s more reposts (completed %s, applied %s reductions)") % ("ad_delay.yaml", 2, 2, 0)
assert any(delayed_message in message for message in caplog.messages)
@pytest.mark.unit
def test_apply_auto_price_reduction_after_repost_delay_reduces_once(
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = 100,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 50, delay_reposts = 2, delay_days = 0
),
price_reduction_count = 0,
repost_count = 3
)
ad_cfg_orig:dict[str, Any] = {}
apply_auto_price_reduction(ad_cfg, ad_cfg_orig, "ad_after_delay.yaml")
assert ad_cfg.price == 90
assert ad_cfg.price_reduction_count == 1
# Note: price_reduction_count is NOT persisted to ad_orig until after successful publish
assert "price_reduction_count" not in ad_cfg_orig
@pytest.mark.unit
def test_apply_auto_price_reduction_waits_when_reduction_already_applied(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = 100,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 50, delay_reposts = 0, delay_days = 0
),
price_reduction_count = 3,
repost_count = 3
)
ad_orig:dict[str, Any] = {}
with caplog.at_level(logging.DEBUG):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_already.yaml")
expected = _("Auto price reduction already applied for [%s]: %s reductions match %s eligible reposts") % ("ad_already.yaml", 3, 3)
assert any(expected in message for message in caplog.messages)
assert ad_cfg.price == 100
assert ad_cfg.price_reduction_count == 3
assert "price_reduction_count" not in ad_orig
@pytest.mark.unit
def test_apply_auto_price_reduction_respects_day_delay(
monkeypatch:pytest.MonkeyPatch,
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
reference = datetime(2025, 1, 1, tzinfo = timezone.utc)
ad_cfg = SimpleNamespace(
price = 150,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 25, min_price = 50, delay_reposts = 0, delay_days = 3
),
price_reduction_count = 0,
repost_count = 1,
updated_on = reference,
created_on = reference
)
monkeypatch.setattr("kleinanzeigen_bot.misc.now", lambda: reference + timedelta(days = 1))
ad_orig:dict[str, Any] = {}
with caplog.at_level("INFO"):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_delay_days.yaml")
assert ad_cfg.price == 150
delayed_message = _("Auto price reduction delayed for [%s]: waiting %s days (elapsed %s)") % ("ad_delay_days.yaml", 3, 1)
assert any(delayed_message in message for message in caplog.messages)
@pytest.mark.unit
def test_apply_auto_price_reduction_runs_after_delays(
monkeypatch:pytest.MonkeyPatch,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
reference = datetime(2025, 1, 1, tzinfo = timezone.utc)
ad_cfg = SimpleNamespace(
price = 120,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "PERCENTAGE", amount = 25, min_price = 60, delay_reposts = 2, delay_days = 3
),
price_reduction_count = 0,
repost_count = 3,
updated_on = reference - timedelta(days = 5),
created_on = reference - timedelta(days = 10)
)
monkeypatch.setattr("kleinanzeigen_bot.misc.now", lambda: reference)
ad_orig:dict[str, Any] = {}
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_ready.yaml")
assert ad_cfg.price == 90
@pytest.mark.unit
def test_apply_auto_price_reduction_delayed_when_timestamp_missing(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = 200,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "FIXED", amount = 20, min_price = 50, delay_reposts = 0, delay_days = 2
),
price_reduction_count = 0,
repost_count = 1,
updated_on = None,
created_on = None
)
ad_orig:dict[str, Any] = {}
with caplog.at_level("INFO"):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_missing_time.yaml")
expected = _("Auto price reduction delayed for [%s]: waiting %s days but publish timestamp missing") % ("ad_missing_time.yaml", 2)
assert any(expected in message for message in caplog.messages)
@pytest.mark.unit
def test_fractional_reduction_increments_counter_even_when_price_unchanged(
caplog:pytest.LogCaptureFixture,
apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
# Test that small fractional reductions increment the counter even when rounded price doesn't change
# This allows cumulative reductions to eventually show visible effect
ad_cfg = SimpleNamespace(
price = 100,
auto_price_reduction = AutoPriceReductionConfig(
enabled = True, strategy = "FIXED", amount = 0.3, min_price = 50, delay_reposts = 0, delay_days = 0
),
price_reduction_count = 0,
repost_count = 1
)
ad_orig:dict[str, Any] = {}
with caplog.at_level(logging.INFO):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_fractional.yaml")
# Price: 100 - 0.3 = 99.7, rounds to 100 (no visible change)
# But counter should still increment for future cumulative reductions
expected = _("Auto price reduction kept price %s after attempting %s reduction cycles") % (100, 1)
assert any(expected in message for message in caplog.messages)
assert ad_cfg.price == 100
assert ad_cfg.price_reduction_count == 1 # Counter incremented despite no visible price change
assert "price_reduction_count" not in ad_orig
@pytest.mark.unit
def test_reduction_value_zero_raises_error() -> None:
with pytest.raises(ContextualValidationError, match = "Input should be greater than 0"):
AutoPriceReductionConfig.model_validate({"enabled": True, "strategy": "PERCENTAGE", "amount": 0, "min_price": 50})
@pytest.mark.unit
def test_reduction_value_negative_raises_error() -> None:
with pytest.raises(ContextualValidationError, match = "Input should be greater than 0"):
AutoPriceReductionConfig.model_validate({"enabled": True, "strategy": "FIXED", "amount": -5, "min_price": 50})
@pytest.mark.unit
def test_percentage_reduction_100_percent() -> None:
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 100, min_price = 0)
assert calculate_auto_price(
base_price = 150,
auto_price_reduction = config,
target_reduction_cycle = 1
) == 0
@pytest.mark.unit
def test_extreme_reduction_cycles() -> None:
# Test that extreme cycle counts don't cause performance issues or errors
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 0)
result = calculate_auto_price(
base_price = 1000,
auto_price_reduction = config,
target_reduction_cycle = 100
)
# With commercial rounding (round after each step), price stabilizes at 5
# because 5 * 0.9 = 4.5 rounds back to 5 with ROUND_HALF_UP
assert result == 5
@pytest.mark.unit
def test_commercial_rounding_each_step() -> None:
"""Test that commercial rounding is applied after each reduction step, not just at the end."""
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 0)
# With 135 EUR and 2x 10% reduction:
# Step 1: 135 * 0.9 = 121.5 → rounds to 122 EUR
# Step 2: 122 * 0.9 = 109.8 → rounds to 110 EUR
# (Without intermediate rounding, it would be: 135 * 0.9^2 = 109.35 → 109 EUR)
result = calculate_auto_price(
base_price = 135,
auto_price_reduction = config,
target_reduction_cycle = 2
)
assert result == 110 # Commercial rounding result
@pytest.mark.unit
def test_extreme_reduction_cycles_with_floor() -> None:
# Test that extreme cycles stop at min_price and don't cause issues
config = AutoPriceReductionConfig(enabled = True, strategy = "PERCENTAGE", amount = 10, min_price = 50)
result = calculate_auto_price(
base_price = 1000,
auto_price_reduction = config,
target_reduction_cycle = 1000
)
# Should stop at min_price, not go to 0, regardless of cycle count
assert result == 50
@pytest.mark.unit
def test_fractional_min_price_is_rounded_up_with_ceiling() -> None:
# Test that fractional min_price is rounded UP using ROUND_CEILING
# This prevents the price from going below min_price due to int() conversion
# Example: min_price=90.5 should become floor of 91, not 90
config = AutoPriceReductionConfig(enabled = True, strategy = "FIXED", amount = 10, min_price = 90.5)
# Start at 100, reduce by 10 = 90
# But min_price=90.5 rounds UP to 91 with ROUND_CEILING
# So the result should be 91, not 90
result = calculate_auto_price(
base_price = 100,
auto_price_reduction = config,
target_reduction_cycle = 1
)
assert result == 91 # Rounded up from 90.5 floor
# Verify with another fractional value
config2 = AutoPriceReductionConfig(enabled = True, strategy = "FIXED", amount = 5, min_price = 49.1)
result2 = calculate_auto_price(
base_price = 60,
auto_price_reduction = config2,
target_reduction_cycle = 3 # 60 - 5 - 5 - 5 = 45, clamped to ceil(49.1) = 50
)
assert result2 == 50 # Rounded up from 49.1 floor

View File

@@ -151,6 +151,14 @@ class TestFormatValidationError:
[{"loc": ("decimal_max_digits",), "msg": "dummy", "type": "decimal_max_digits", "ctx": {"max_digits": 10, "expected_plural": "s"}}], [{"loc": ("decimal_max_digits",), "msg": "dummy", "type": "decimal_max_digits", "ctx": {"max_digits": 10, "expected_plural": "s"}}],
"Decimal input should have no more than 10 digits in total", "Decimal input should have no more than 10 digits in total",
), ),
(
[{"loc": ("decimal_max_places",), "msg": "dummy", "type": "decimal_max_places", "ctx": {"decimal_places": 2, "expected_plural": "s"}}],
"Decimal input should have no more than 2 decimal places",
),
(
[{"loc": ("decimal_whole_digits",), "msg": "dummy", "type": "decimal_whole_digits", "ctx": {"whole_digits": 3, "expected_plural": ""}}],
"Decimal input should have no more than 3 digits before the decimal point",
),
# Complex number related errors # Complex number related errors
( (
[{"loc": ("complex",), "msg": "dummy", "type": "complex_type"}], [{"loc": ("complex",), "msg": "dummy", "type": "complex_type"}],