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

test: improve test quality by replacing brittle assertions (#972)

Browse Source 
## ℹ️ Description
This PR improves overall test quality by removing brittle assertion
patterns, consolidating duplicate coverage, and migrating tests to
clearer ownership while preserving behavior coverage.

- Link to the related issue(s): Issue #N/A
- Describe the motivation and context for this change.
- Reduce flaky/wording-coupled tests and improve long-term
maintainability.
- Keep high-signal behavior assertions while preserving existing
coverage paths.

## 📋 Changes Summary
- Replaced gettext/log-wording-dependent assertions in price reduction
tests with stable behavior-oriented checks.
- Cleaned low-value log assertions in extract/pagination/bot tests.
- Simplified mixed-category assertions in init/diagnostics tests and
removed redundant debug log checks.
- Extracted shared `PYTEST_CURRENT_TEST` fixture handling in chrome
version tests.
- Removed duplicate/subsumed tests in ad-model and update-checker
suites; simplified brittle full-message log checks.
- Parametrized legacy migration tests in config-model for clearer, less
repetitive coverage.
- Removed tests in `test_bot.py` that were subsumed by `test_init.py`,
then migrated remaining unique tests and deleted `test_bot.py`.
- No runtime behavior changes; this is test-suite quality and
maintainability work.

### ⚙️ Type of Change
Select the type(s) of change(s) included in this pull request:
- [x] 🐞 Bug fix (non-breaking change which fixes an issue)
- [ ] ✨ New feature (adds new functionality without breaking existing
usage)
- [ ] 💥 Breaking change (changes that might break existing user setups,
scripts, or configurations)


## ✅ Checklist
Before requesting a review, confirm the following:
- [x] I have reviewed my changes to ensure they meet the project's
standards.
- [x] I have tested my changes and ensured that all tests pass (`pdm run
test`).
- [x] I have formatted the code (`pdm run format`).
- [x] I have verified that linting passes (`pdm run lint`).
- [x] I have updated documentation where necessary.

By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Tests**
* Refactored and simplified the unit test suite for maintainability:
consolidated legacy migration checks into parametrized tests, introduced
shared fixtures, and streamlined test setup/formatting.
* Removed many log-capture assertions and several redundant or
overly-specific tests, and adjusted assertions to focus on functional
outcomes.
  * No production code or user-facing behavior changed.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Jens
2026-04-10 17:43:47 +02:00
committed by GitHub
parent c3584d0b2e
commit c65d1b67b0
10 changed files with 455 additions and 1020 deletions
Download Patch File Download Diff File Expand all files Collapse all files

248
tests/unit/test_ad_model.py
View File

@@ -7,7 +7,7 @@ import math
import pytest
from kleinanzeigen_bot.model.ad_model import MAX_DESCRIPTION_LENGTH, Ad, AdPartial, ShippingOption, calculate_auto_price
from kleinanzeigen_bot.model.ad_model import MAX_DESCRIPTION_LENGTH, AdPartial, ShippingOption, calculate_auto_price
from kleinanzeigen_bot.model.config_model import AdDefaults, AutoPriceReductionConfig
from kleinanzeigen_bot.utils.pydantics import ContextualModel, ContextualValidationError
@@ -24,28 +24,52 @@ def test_update_content_hash() -> None:
assert AdPartial.model_validate(minimal_ad_cfg).update_content_hash().content_hash == minimal_ad_cfg_hash
assert AdPartial.model_validate(minimal_ad_cfg | {
"id": "123456789",
"created_on": "2025-05-08T09:34:03",
"updated_on": "2025-05-14T20:43:16",
"content_hash": "5753ead7cf42b0ace5fe658ecb930b3a8f57ef49bd52b7ea2d64b91b2c75517e"
}).update_content_hash().content_hash == minimal_ad_cfg_hash
assert (
AdPartial.model_validate(
minimal_ad_cfg
| {
"id": "123456789",
"created_on": "2025-05-08T09:34:03",
"updated_on": "2025-05-14T20:43:16",
"content_hash": "5753ead7cf42b0ace5fe658ecb930b3a8f57ef49bd52b7ea2d64b91b2c75517e",
}
)
.update_content_hash()
.content_hash
== minimal_ad_cfg_hash
)
assert AdPartial.model_validate(minimal_ad_cfg | {
"active": None,
"images": None,
"shipping_options": None,
"special_attributes": None,
"contact": None,
}).update_content_hash().content_hash == minimal_ad_cfg_hash
assert (
AdPartial.model_validate(
minimal_ad_cfg
| {
"active": None,
"images": None,
"shipping_options": None,
"special_attributes": None,
"contact": None,
}
)
.update_content_hash()
.content_hash
== minimal_ad_cfg_hash
)
assert AdPartial.model_validate(minimal_ad_cfg | {
"active": True,
"images": [],
"shipping_options": [],
"special_attributes": {},
"contact": {},
}).update_content_hash().content_hash != minimal_ad_cfg_hash
assert (
AdPartial.model_validate(
minimal_ad_cfg
| {
"active": True,
"images": [],
"shipping_options": [],
"special_attributes": {},
"contact": {},
}
)
.update_content_hash()
.content_hash
!= minimal_ad_cfg_hash
)
@pytest.mark.unit
@@ -112,11 +136,7 @@ def test_shipping_option_must_not_be_blank() -> None:
@pytest.mark.unit
def test_description_length_limit() -> None:
cfg = {
"title": "Description Length",
"category": "160",
"description": "x" * (MAX_DESCRIPTION_LENGTH + 1)
}
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)
@@ -133,7 +153,7 @@ def base_ad_cfg() -> dict[str, object]:
"shipping_type": "PICKUP",
"sell_directly": False,
"type": "OFFER",
"active": True
"active": True,
}
@@ -142,14 +162,7 @@ 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
}
"auto_price_reduction": {"enabled": True, "strategy": "FIXED", "amount": 5, "min_price": 50, "delay_reposts": 0, "delay_days": 0},
}
@@ -163,38 +176,21 @@ class SparseDumpAdPartial(AdPartial):
@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
}
}
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
}
}
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
}
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
@@ -205,15 +201,7 @@ def test_prepare_ad_model_fills_missing_counters(base_ad_cfg:dict[str, object])
@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
}
}
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)
@@ -222,29 +210,13 @@ def test_min_price_must_not_exceed_price(base_ad_cfg:dict[str, object]) -> None:
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
}
}
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"
}
}
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)
@@ -252,24 +224,14 @@ def test_min_price_validation_defers_to_pydantic_for_invalid_types(base_ad_cfg:d
@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
}
}
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
}
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
@@ -280,34 +242,13 @@ def test_to_ad_stabilizes_counters_when_defaults_omit(base_ad_cfg:dict[str, obje
@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
}
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
@@ -320,9 +261,7 @@ def test_price_reduction_delay_inherited_from_defaults(complete_ad_cfg:dict[str,
amount = 5,
min_price = 50,
delay_reposts = 4,
delay_days = 0
)
)
delay_days = 0))
ad = AdPartial.model_validate(cfg).to_ad(defaults)
assert ad.auto_price_reduction.delay_reposts == 4
@@ -331,14 +270,7 @@ def test_price_reduction_delay_inherited_from_defaults(complete_ad_cfg:dict[str,
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
}
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,
@@ -346,53 +278,17 @@ def test_price_reduction_delay_override_zero(complete_ad_cfg:dict[str, object])
amount = 5,
min_price = 50,
delay_reposts = 4,
delay_days = 0
)
)
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
)
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
@@ -400,14 +296,8 @@ def test_calculate_auto_price_with_missing_strategy() -> None:
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
)
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
@@ -415,16 +305,10 @@ def test_calculate_auto_price_with_missing_amount() -> None:
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
)
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
)
calculate_auto_price(base_price = 100, auto_price_reduction = config, target_reduction_cycle = 1)
@pytest.mark.unit

81
tests/unit/test_bot.py
View File

@@ -1,81 +0,0 @@
# 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 gc, pytest # isort: skip
import pathlib
from kleinanzeigen_bot import KleinanzeigenBot
class TestKleinanzeigenBot:
@pytest.fixture
def bot(self) -> KleinanzeigenBot:
return KleinanzeigenBot()
def test_parse_args_help(self, bot:KleinanzeigenBot) -> None:
"""Test parsing of help command"""
bot.parse_args(["app", "help"])
assert bot.command == "help"
assert bot.ads_selector == "due"
assert not bot.keep_old_ads
def test_parse_args_publish(self, bot:KleinanzeigenBot) -> None:
"""Test parsing of publish command with options"""
bot.parse_args(["app", "publish", "--ads=all", "--keep-old"])
assert bot.command == "publish"
assert bot.ads_selector == "all"
assert bot.keep_old_ads
def test_parse_args_create_config(self, bot:KleinanzeigenBot) -> None:
"""Test parsing of create-config command"""
bot.parse_args(["app", "create-config"])
assert bot.command == "create-config"
def test_create_default_config_logs_error_if_exists(self, tmp_path:pathlib.Path, bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture) -> None:
"""Test that create_default_config logs an error if the config file already exists."""
config_path = tmp_path / "config.yaml"
config_path.write_text("dummy: value")
bot.config_file_path = str(config_path)
with caplog.at_level("ERROR"):
bot.create_default_config()
assert any("already exists" in m for m in caplog.messages)
def test_create_default_config_creates_file(self, tmp_path:pathlib.Path, bot:KleinanzeigenBot) -> None:
"""Test that create_default_config creates a config file if it does not exist."""
config_path = tmp_path / "config.yaml"
bot.config_file_path = str(config_path)
assert not config_path.exists()
bot.create_default_config()
assert config_path.exists()
content = config_path.read_text()
assert "username: changeme" in content
def test_load_config_handles_missing_file(self, tmp_path:pathlib.Path, bot:KleinanzeigenBot) -> None:
"""Test that load_config creates a default config file if missing. No info log is expected anymore."""
config_path = tmp_path / "config.yaml"
bot.config_file_path = str(config_path)
bot.load_config()
assert config_path.exists()
def test_get_version(self, bot:KleinanzeigenBot) -> None:
"""Test version retrieval"""
version = bot.get_version()
assert isinstance(version, str)
assert len(version) > 0
def test_file_log_closed_after_bot_shutdown(self) -> None:
"""Ensure the file log handler is properly closed after the bot is deleted"""
# Directly instantiate the bot to control its lifecycle within the test
bot = KleinanzeigenBot()
bot.configure_file_logging()
file_log = bot.file_log
assert file_log is not None
assert not file_log.is_closed()
# Delete and garbage collect the bot instance to ensure the destructor (__del__) is called
del bot
gc.collect()
assert file_log.is_closed()

195
tests/unit/test_config_model.py
View File

@@ -7,28 +7,42 @@ from kleinanzeigen_bot.model import config_model
from kleinanzeigen_bot.model.config_model import DEFAULT_DOWNLOAD_DIR, AdDefaults, Config, TimeoutConfig
def test_migrate_legacy_description_prefix() -> None:
assert AdDefaults.model_validate({}).description_prefix == "" # noqa: PLC1901 explicit empty check is clearer
@pytest.mark.parametrize("field", ["prefix", "suffix"])
def test_migrate_legacy_description(field:str) -> None:
top_key = f"description_{field}"
assert AdDefaults.model_validate({"description_prefix": "Prefix"}).description_prefix == "Prefix"
# empty default
assert getattr(AdDefaults.model_validate({}), top_key) == "" # noqa: PLC1901
assert AdDefaults.model_validate({"description_prefix": "Prefix", "description": {"prefix": "Legacy Prefix"}}).description_prefix == "Prefix"
# top-level key used directly
assert getattr(AdDefaults.model_validate({top_key: "Value"}), top_key) == "Value"
assert AdDefaults.model_validate({"description": {"prefix": "Legacy Prefix"}}).description_prefix == "Legacy Prefix"
# top-level key takes precedence over legacy nested key
assert (
getattr(
AdDefaults.model_validate({top_key: "Value", "description": {field: "Legacy"}}),
top_key,
)
== "Value"
)
assert AdDefaults.model_validate({"description_prefix": "", "description": {"prefix": "Legacy Prefix"}}).description_prefix == "Legacy Prefix"
# legacy nested key migrates when no top-level key present
assert (
getattr(
AdDefaults.model_validate({"description": {field: "Legacy"}}),
top_key,
)
== "Legacy"
)
def test_migrate_legacy_description_suffix() -> None:
assert AdDefaults.model_validate({}).description_suffix == "" # noqa: PLC1901 explicit empty check is clearer
assert AdDefaults.model_validate({"description_suffix": "Suffix"}).description_suffix == "Suffix"
assert AdDefaults.model_validate({"description_suffix": "Suffix", "description": {"suffix": "Legacy Suffix"}}).description_suffix == "Suffix"
assert AdDefaults.model_validate({"description": {"suffix": "Legacy Suffix"}}).description_suffix == "Legacy Suffix"
assert AdDefaults.model_validate({"description_suffix": "", "description": {"suffix": "Legacy Suffix"}}).description_suffix == "Legacy Suffix"
# empty top-level falls back to legacy nested key
assert (
getattr(
AdDefaults.model_validate({top_key: "", "description": {field: "Legacy"}}),
top_key,
)
== "Legacy"
)
def test_minimal_config_validation() -> None:
@@ -315,109 +329,78 @@ def test_timeout_config_resolve_falls_back_to_default() -> None:
assert timeouts.resolve("nonexistent_key") == 3.0
def test_diagnostics_pause_requires_capture_validation() -> None:
@pytest.fixture
def minimal_config() -> dict[str, object]:
"""Minimal valid config dict for diagnostics-related tests."""
return {
"ad_defaults": {"contact": {"name": "dummy", "zipcode": "12345"}},
"login": {"username": "dummy", "password": "dummy"}, # noqa: S105
}
def test_diagnostics_pause_requires_capture_validation(minimal_config:dict[str, object]) -> None:
"""
Unit: DiagnosticsConfig validator ensures pause_on_login_detection_failure
requires capture_on.login_detection to be enabled.
"""
minimal_cfg = {
"ad_defaults": {"contact": {"name": "dummy", "zipcode": "12345"}},
"login": {"username": "dummy", "password": "dummy"}, # noqa: S105
"publishing": {"delete_old_ads": "BEFORE_PUBLISH", "delete_old_ads_by_title": False},
}
valid_cfg = {**minimal_cfg, "diagnostics": {"capture_on": {"login_detection": True}, "pause_on_login_detection_failure": True}}
valid_cfg = {**minimal_config, "diagnostics": {"capture_on": {"login_detection": True}, "pause_on_login_detection_failure": True}}
config = Config.model_validate(valid_cfg)
assert config.diagnostics is not None
assert config.diagnostics.pause_on_login_detection_failure is True
assert config.diagnostics.capture_on.login_detection is True
invalid_cfg = {**minimal_cfg, "diagnostics": {"capture_on": {"login_detection": False}, "pause_on_login_detection_failure": True}}
invalid_cfg = {**minimal_config, "diagnostics": {"capture_on": {"login_detection": False}, "pause_on_login_detection_failure": True}}
with pytest.raises(ValueError, match = "pause_on_login_detection_failure requires capture_on.login_detection to be enabled"):
Config.model_validate(invalid_cfg)
def test_diagnostics_legacy_login_detection_capture_migration_when_capture_on_exists() -> None:
"""
Unit: Test that legacy login_detection_capture is removed but doesn't overwrite explicit capture_on.login_detection.
"""
minimal_cfg = {
"ad_defaults": {"contact": {"name": "dummy", "zipcode": "12345"}},
"login": {"username": "dummy", "password": "dummy"}, # noqa: S105
}
# When capture_on.login_detection is explicitly set to False, legacy True should be ignored
cfg_with_explicit = {
**minimal_cfg,
@pytest.mark.parametrize(
("legacy_key", "capture_attr", "capture_on_value", "expected"),
[
pytest.param(
"login_detection_capture",
"login_detection",
{"login_detection": False},
False,
id = "login_detection-ignored-when-capture_on-present",
),
pytest.param(
"publish_error_capture",
"publish",
{"publish": False},
False,
id = "publish-ignored-when-capture_on-present",
),
pytest.param(
"login_detection_capture",
"login_detection",
None,
True,
id = "login_detection-migrated-when-capture_on-none",
),
pytest.param(
"publish_error_capture",
"publish",
None,
True,
id = "publish-migrated-when-capture_on-none",
),
],
)
def test_diagnostics_legacy_capture_migration(
minimal_config:dict[str, object],
legacy_key:str,
capture_attr:str,
capture_on_value:dict[str, bool] | None,
expected:bool,
) -> None:
cfg = {
**minimal_config,
"diagnostics": {
"login_detection_capture": True, # legacy key
"capture_on": {"login_detection": False}, # explicit new key set to False
legacy_key: True,
"capture_on": capture_on_value,
},
}
config = Config.model_validate(cfg_with_explicit)
config = Config.model_validate(cfg)
assert config.diagnostics is not None
assert config.diagnostics.capture_on.login_detection is False # explicit value preserved
def test_diagnostics_legacy_publish_error_capture_migration_when_capture_on_exists() -> None:
"""
Unit: Test that legacy publish_error_capture is removed but doesn't overwrite explicit capture_on.publish.
"""
minimal_cfg = {
"ad_defaults": {"contact": {"name": "dummy", "zipcode": "12345"}},
"login": {"username": "dummy", "password": "dummy"}, # noqa: S105
}
# When capture_on.publish is explicitly set to False, legacy True should be ignored
cfg_with_explicit = {
**minimal_cfg,
"diagnostics": {
"publish_error_capture": True, # legacy key
"capture_on": {"publish": False}, # explicit new key set to False
},
}
config = Config.model_validate(cfg_with_explicit)
assert config.diagnostics is not None
assert config.diagnostics.capture_on.publish is False # explicit value preserved
def test_diagnostics_legacy_login_detection_capture_migration_when_capture_on_is_none() -> None:
"""
Unit: Test that legacy login_detection_capture is migrated when capture_on is None.
"""
minimal_cfg = {
"ad_defaults": {"contact": {"name": "dummy", "zipcode": "12345"}},
"login": {"username": "dummy", "password": "dummy"}, # noqa: S105
}
cfg_with_null_capture_on = {
**minimal_cfg,
"diagnostics": {
"login_detection_capture": True, # legacy key
"capture_on": None, # capture_on is explicitly None
},
}
config = Config.model_validate(cfg_with_null_capture_on)
assert config.diagnostics is not None
assert config.diagnostics.capture_on.login_detection is True # legacy value migrated
def test_diagnostics_legacy_publish_error_capture_migration_when_capture_on_is_none() -> None:
"""
Unit: Test that legacy publish_error_capture is migrated when capture_on is None.
"""
minimal_cfg = {
"ad_defaults": {"contact": {"name": "dummy", "zipcode": "12345"}},
"login": {"username": "dummy", "password": "dummy"}, # noqa: S105
}
cfg_with_null_capture_on = {
**minimal_cfg,
"diagnostics": {
"publish_error_capture": True, # legacy key
"capture_on": None, # capture_on is explicitly None
},
}
config = Config.model_validate(cfg_with_null_capture_on)
assert config.diagnostics is not None
assert config.diagnostics.capture_on.publish is True # legacy value migrated
assert getattr(config.diagnostics.capture_on, capture_attr) == expected

27
tests/unit/test_diagnostics.py
View File

@@ -112,7 +112,7 @@ class TestDiagnosticsCapture:
assert not log_path.exists()
@pytest.mark.asyncio
async def test_capture_diagnostics_handles_screenshot_exception(self, tmp_path:Path, caplog:pytest.LogCaptureFixture) -> None:
async def test_capture_diagnostics_handles_screenshot_exception(self, tmp_path:Path) -> None:
"""Test that capture_diagnostics handles screenshot capture exceptions gracefully."""
mock_page = AsyncMock()
mock_page.save_screenshot = AsyncMock(side_effect = Exception("Screenshot failed"))
@@ -126,10 +126,9 @@ class TestDiagnosticsCapture:
# Verify no artifacts were saved due to exception
assert len(result.saved_artifacts) == 0
assert "Diagnostics screenshot capture failed" in caplog.text
@pytest.mark.asyncio
async def test_capture_diagnostics_handles_json_exception(self, tmp_path:Path, caplog:pytest.LogCaptureFixture, monkeypatch:pytest.MonkeyPatch) -> None:
async def test_capture_diagnostics_handles_json_exception(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None:
"""Test that capture_diagnostics handles JSON write exceptions gracefully."""
mock_page = AsyncMock()
mock_page.get_content = AsyncMock(return_value = "<html></html>")
@@ -151,12 +150,9 @@ class TestDiagnosticsCapture:
assert any(a.suffix == ".png" for a in result.saved_artifacts)
assert any(a.suffix == ".html" for a in result.saved_artifacts)
assert not any(a.suffix == ".json" for a in result.saved_artifacts)
assert "Diagnostics JSON capture failed" in caplog.text
@pytest.mark.asyncio
async def test_capture_diagnostics_handles_log_copy_exception(
self, tmp_path:Path, caplog:pytest.LogCaptureFixture, monkeypatch:pytest.MonkeyPatch
) -> None:
async def test_capture_diagnostics_handles_log_copy_exception(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None:
"""Test that capture_diagnostics handles log copy exceptions gracefully."""
# Create a log file
log_file = tmp_path / "test.log"
@@ -179,7 +175,6 @@ class TestDiagnosticsCapture:
# Verify no artifacts were saved due to exception
assert len(result.saved_artifacts) == 0
assert "Diagnostics log copy failed" in caplog.text
finally:
monkeypatch.setattr(diagnostics_module, "_copy_log_sync", original_copy_log)
@@ -206,19 +201,3 @@ class TestDiagnosticsCapture:
# Verify no artifacts were saved
assert len(result.saved_artifacts) == 0
assert "Diagnostics capture attempted but no artifacts were saved" in caplog.text
@pytest.mark.asyncio
async def test_capture_diagnostics_logs_debug_when_no_capture_requested(self, tmp_path:Path, caplog:pytest.LogCaptureFixture) -> None:
"""Test debug is logged when no diagnostics capture is requested."""
output_dir = tmp_path / "diagnostics"
with caplog.at_level("DEBUG"):
_ = await capture_diagnostics(
output_dir = output_dir,
base_prefix = "test",
page = None,
json_payload = None,
copy_log = False,
)
assert "No diagnostics capture requested" in caplog.text

29
tests/unit/test_extract.py
View File

@@ -2507,23 +2507,21 @@ class TestRenderDownloadNameWithBudgetWarnings:
assert "Short" in rendered
assert "truncated" not in caplog.text.lower()
def test_warns_when_id_truncated(self, test_extractor:extract_module.AdExtractor, caplog:pytest.LogCaptureFixture) -> None:
"""Warning emitted when {id} is truncated."""
with caplog.at_level("WARNING"):
rendered = test_extractor._render_download_name_with_budget("{id}", 12345678901234567890, "", 5)
def test_truncates_id_when_budget_exhausted(self, test_extractor:extract_module.AdExtractor) -> None:
"""{id} is truncated to fit within budget."""
rendered = test_extractor._render_download_name_with_budget("{id}", 12345678901234567890, "", 5)
assert rendered == "12345"
assert "12345678901234567890" not in rendered
assert len(rendered) <= 5
assert "truncated {id}" in caplog.text
def test_warns_when_title_truncated(self, test_extractor:extract_module.AdExtractor, caplog:pytest.LogCaptureFixture) -> None:
"""Warning emitted when {title} is truncated."""
with caplog.at_level("WARNING"):
rendered = test_extractor._render_download_name_with_budget("{id}_{title}", 12345, "Very Long Title Here", 12)
def test_truncates_title_when_budget_exhausted(self, test_extractor:extract_module.AdExtractor) -> None:
"""{title} is truncated to fit within budget while {id} is preserved."""
rendered = test_extractor._render_download_name_with_budget("{id}_{title}", 12345, "Very Long Title Here", 12)
assert "12345" in rendered
assert "truncated {title}" in caplog.text
assert "Very Long Title Here" not in rendered
assert len(rendered) <= 12
def test_id_protected_over_literals(self, test_extractor:extract_module.AdExtractor) -> None:
"""{id} is protected over literal text with tight budget."""
@@ -2586,11 +2584,10 @@ class TestRenderDownloadNameWithBudgetWarnings:
assert rendered.endswith("_12345")
assert len(rendered) <= 20
def test_warns_when_both_id_and_title_truncated(self, test_extractor:extract_module.AdExtractor, caplog:pytest.LogCaptureFixture) -> None:
"""Warnings are emitted when both placeholders are truncated."""
with caplog.at_level("WARNING"):
rendered = test_extractor._render_download_name_with_budget("{title}_{id}", 12345678901234567890, "Very Long Title Here", 15)
def test_truncates_both_id_and_title_under_extreme_budget(self, test_extractor:extract_module.AdExtractor) -> None:
"""Both {id} and {title} are truncated to fit within a tight budget."""
rendered = test_extractor._render_download_name_with_budget("{title}_{id}", 12345678901234567890, "Very Long Title Here", 15)
assert len(rendered) <= 15
assert "truncated {id}" in caplog.text
assert "truncated {title}" in caplog.text
assert "Very Long Title Here" not in rendered
assert "12345678901234567890" not in rendered

110
tests/unit/test_init.py
View File

@@ -1,7 +1,7 @@
# 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, io, json, logging, os, tempfile # isort: skip
import asyncio, copy, fnmatch, gc, io, json, logging, os, tempfile # isort: skip
from collections.abc import Callable, Generator
from contextlib import contextmanager, redirect_stdout
from datetime import timedelta
@@ -489,28 +489,24 @@ class TestKleinanzeigenBotInitialization:
"published_ads": [{"id": 123, "state": "active"}],
"expected_active": True,
"expect_ownership_warning": False,
"expect_inactive_debug": False,
},
{
"name": "inactive ad",
"published_ads": [{"id": 123, "state": "inactive"}],
"expected_active": False,
"expect_ownership_warning": False,
"expect_inactive_debug": True,
},
{
"name": "paused ad",
"published_ads": [{"id": 123, "state": "paused"}],
"expected_active": False,
"expect_ownership_warning": False,
"expect_inactive_debug": True,
},
{
"name": "ad not in published profile",
"published_ads": [{"id": 999, "state": "active"}], # Different ID
"expected_active": False,
"expect_ownership_warning": True,
"expect_inactive_debug": False,
},
],
ids = lambda s: s["name"],
@@ -534,7 +530,7 @@ class TestKleinanzeigenBotInitialization:
extractor_mock.navigate_to_ad_page = AsyncMock(return_value = True)
extractor_mock.download_ad = AsyncMock()
caplog.set_level(logging.DEBUG)
caplog.set_level(logging.WARNING)
with (
patch.object(test_bot, "_fetch_published_ads", new_callable = AsyncMock, return_value = scenario["published_ads"]) as mock_fetch_published_ads,
@@ -548,11 +544,11 @@ class TestKleinanzeigenBotInitialization:
# Verify download_ad called with correct active parameter
extractor_mock.download_ad.assert_awaited_once_with(123, active = scenario["expected_active"])
# Verify appropriate logging
# Verify ownership warning only when expected
if scenario["expect_ownership_warning"]:
assert any("found in overview but not in published profile" in msg for msg in caplog.messages)
if scenario["expect_inactive_debug"]:
assert any("has state" in msg and "Saving as inactive" in msg for msg in caplog.messages)
else:
assert not any("found in overview but not in published profile" in msg for msg in caplog.messages)
@pytest.mark.asyncio
async def test_download_ads_all_selector_skips_invalid_ad_id(
@@ -656,7 +652,6 @@ class TestKleinanzeigenBotInitialization:
self,
test_bot:KleinanzeigenBot,
tmp_path:Path,
caplog:pytest.LogCaptureFixture,
) -> None:
"""Test that --ads=new skips already-saved ads (existing behavior unchanged)."""
test_bot.workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot")
@@ -672,8 +667,6 @@ class TestKleinanzeigenBotInitialization:
# Mock load_ads to return ad 123 as already saved
saved_ads:list[tuple[str, MagicMock, dict[str, Any]]] = [("ad_123.yaml", MagicMock(spec = Ad, id = 123), {})]
caplog.set_level(logging.INFO)
with (
patch.object(test_bot, "_fetch_published_ads", new_callable = AsyncMock, return_value = []),
patch("kleinanzeigen_bot.extract.AdExtractor", return_value = extractor_mock),
@@ -684,9 +677,6 @@ class TestKleinanzeigenBotInitialization:
# Verify download_ad was NOT called for already-saved ad
extractor_mock.download_ad.assert_not_called()
# Verify log message about already saved
assert any("already been saved" in msg for msg in caplog.messages)
@pytest.mark.asyncio
async def test_download_ads_new_selector_skips_invalid_ad_id(
self,
@@ -721,13 +711,12 @@ class TestKleinanzeigenBotInitialization:
assert not any("not in published profile" in msg for msg in caplog.messages)
@pytest.mark.asyncio
async def test_download_ads_new_selector_logs_warning_for_ad_not_in_published_profile(
async def test_download_ads_new_selector_passes_inactive_for_ad_not_in_published_profile(
self,
test_bot:KleinanzeigenBot,
tmp_path:Path,
caplog:pytest.LogCaptureFixture,
) -> None:
"""Test that --ads=new logs warning when ad is not in published profile."""
"""Test that --ads=new passes active=False when ad is not in published profile."""
test_bot.workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot")
test_bot.ads_selector = "new"
test_bot.browser = MagicMock()
@@ -741,8 +730,6 @@ class TestKleinanzeigenBotInitialization:
# Mock load_ads to return different saved ads (999 is new but not in published profile)
saved_ads:list[tuple[str, MagicMock, dict[str, Any]]] = [("ad_123.yaml", MagicMock(spec = Ad, id = 123), {})]
caplog.set_level(logging.WARNING)
with (
patch.object(test_bot, "_fetch_published_ads", new_callable = AsyncMock, return_value = []),
patch("kleinanzeigen_bot.extract.AdExtractor", return_value = extractor_mock),
@@ -753,45 +740,6 @@ class TestKleinanzeigenBotInitialization:
# Verify download_ad was called with active=False (not in profile)
extractor_mock.download_ad.assert_awaited_once_with(999, active = False)
# Verify warning about not in published profile
assert any("found in overview but not in published profile" in msg for msg in caplog.messages)
@pytest.mark.asyncio
async def test_download_ads_new_selector_logs_debug_for_inactive_state(
self,
test_bot:KleinanzeigenBot,
tmp_path:Path,
caplog:pytest.LogCaptureFixture,
) -> None:
"""Test that --ads=new logs debug message when ad has inactive state."""
test_bot.workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot")
test_bot.ads_selector = "new"
test_bot.browser = MagicMock()
extractor_mock = MagicMock()
extractor_mock.extract_own_ads_urls = AsyncMock(return_value = ["https://www.kleinanzeigen.de/s-anzeige/test-ad/999-234-5678"])
extractor_mock.extract_ad_id_from_ad_url = MagicMock(return_value = 999)
extractor_mock.navigate_to_ad_page = AsyncMock(return_value = True)
extractor_mock.download_ad = AsyncMock()
# Mock load_ads to return different saved ads (999 is new and inactive)
saved_ads:list[tuple[str, MagicMock, dict[str, Any]]] = [("ad_123.yaml", MagicMock(spec = Ad, id = 123), {})]
caplog.set_level(logging.DEBUG)
with (
patch.object(test_bot, "_fetch_published_ads", new_callable = AsyncMock, return_value = [{"id": 999, "state": "inactive"}]),
patch("kleinanzeigen_bot.extract.AdExtractor", return_value = extractor_mock),
patch.object(test_bot, "load_ads", return_value = saved_ads),
):
await test_bot.download_ads()
# Verify download_ad was called with active=False
extractor_mock.download_ad.assert_awaited_once_with(999, active = False)
# Verify debug message about inactive state
assert any("has state" in msg and "Saving as inactive" in msg for msg in caplog.messages)
@pytest.mark.asyncio
async def test_download_ads_all_selector_skips_when_navigation_fails(
self,
@@ -848,6 +796,25 @@ class TestKleinanzeigenBotInitialization:
# All non-"active" states should result in active=False
extractor_mock.download_ad.assert_awaited_once_with(123, active = False)
def test_create_default_config_preserves_existing_file(self, tmp_path:Path, test_bot:KleinanzeigenBot) -> None:
"""Test that create_default_config does not overwrite an existing config file."""
config_path = tmp_path / "config.yaml"
original_content = "dummy: value"
config_path.write_text(original_content)
test_bot.config_file_path = str(config_path)
test_bot.create_default_config()
assert config_path.read_text() == original_content
def test_create_default_config_creates_file(self, tmp_path:Path, test_bot:KleinanzeigenBot) -> None:
"""Test that create_default_config creates a config file if it does not exist."""
config_path = tmp_path / "config.yaml"
test_bot.config_file_path = str(config_path)
assert not config_path.exists()
test_bot.create_default_config()
assert config_path.exists()
content = config_path.read_text()
assert "username: changeme" in content
class TestKleinanzeigenBotLogging:
"""Tests for logging functionality."""
@@ -880,6 +847,26 @@ class TestKleinanzeigenBotLogging:
assert test_bot.file_log is None
assert list(root_logger.handlers) == initial_handlers
def test_file_log_closed_after_bot_shutdown(self, tmp_path:Path) -> None:
"""Ensure the file log handler is properly closed after the bot is deleted."""
# Directly instantiate the bot to control its lifecycle within the test
bot = KleinanzeigenBot()
log_path = tmp_path / "test.log"
bot.log_file_path = str(log_path)
bot.configure_file_logging()
file_log = bot.file_log
assert file_log is not None
assert log_path.exists()
assert not file_log.is_closed()
# Delete and garbage collect the bot instance to ensure the destructor (__del__) is called
del bot
gc.collect()
assert file_log.is_closed()
class TestKleinanzeigenBotCommandLine:
"""Tests for command line argument parsing."""
@@ -933,6 +920,11 @@ class TestKleinanzeigenBotCommandLine:
test_bot.parse_args(["dummy", "--config", str(config_path)])
assert test_bot.config_file_path == str(config_path.absolute())
def test_parse_args_create_config(self, test_bot:KleinanzeigenBot) -> None:
"""Test parsing of create-config command"""
test_bot.parse_args(["app", "create-config"])
assert test_bot.command == "create-config"
class TestKleinanzeigenBotConfiguration:
"""Tests for configuration loading and validation."""

24
tests/unit/test_json_pagination.py
View File

@@ -221,26 +221,22 @@ class TestJSONPagination:
pytest.fail(f"expected empty list when 'ads' is not a list, got: {result}")
@pytest.mark.asyncio
async def test_fetch_published_ads_filters_non_dict_entries(self, bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture) -> None:
"""Malformed entries should be filtered and logged."""
async def test_fetch_published_ads_filters_non_dict_entries(self, bot:KleinanzeigenBot) -> None:
"""Malformed entries should be filtered out."""
response_data = {"ads": [42, {"id": 1, "state": "active"}, "broken"], "paging": {"pageNum": 1, "last": 1}}
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
mock_request.return_value = {"content": json.dumps(response_data)}
with caplog.at_level("WARNING"):
result = await bot._fetch_published_ads()
result = await bot._fetch_published_ads()
if result != [{"id": 1, "state": "active"}]:
pytest.fail(f"expected malformed entries to be filtered out, got: {result}")
if "Filtered 2 malformed ad entries on page 1" not in caplog.text:
pytest.fail(f"expected malformed-entry warning in logs, got: {caplog.text}")
@pytest.mark.asyncio
async def test_fetch_published_ads_filters_dict_entries_missing_required_keys(
self,
bot:KleinanzeigenBot,
caplog:pytest.LogCaptureFixture,
) -> None:
"""Dict entries without required id/state keys should be rejected."""
response_data = {
@@ -256,13 +252,10 @@ class TestJSONPagination:
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
mock_request.return_value = {"content": json.dumps(response_data)}
with caplog.at_level("WARNING"):
result = await bot._fetch_published_ads()
result = await bot._fetch_published_ads()
if result != [{"id": 2, "state": "paused"}]:
pytest.fail(f"expected only entries with id and state to remain, got: {result}")
if "Filtered 3 malformed ad entries on page 1" not in caplog.text:
pytest.fail(f"expected malformed-entry warning in logs, got: {caplog.text}")
@pytest.mark.asyncio
async def test_fetch_published_ads_strict_raises_on_malformed_entries(self, bot:KleinanzeigenBot) -> None:
@@ -297,18 +290,15 @@ class TestJSONPagination:
await bot._fetch_published_ads(strict = True)
@pytest.mark.asyncio
async def test_fetch_published_ads_handles_non_string_content_type(self, bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture) -> None:
"""Unexpected non-string content types should stop pagination with warning."""
async def test_fetch_published_ads_handles_non_string_content_type(self, bot:KleinanzeigenBot) -> None:
"""Unexpected non-string content types should stop pagination and return empty."""
with patch.object(bot, "web_request", new_callable = AsyncMock) as mock_request:
mock_request.return_value = {"content": None}
with caplog.at_level("WARNING"):
result = await bot._fetch_published_ads()
result = await bot._fetch_published_ads()
if result != []:
pytest.fail(f"expected empty result on non-string content, got: {result}")
if "Unexpected response content type on page 1: NoneType" not in caplog.text:
pytest.fail(f"expected non-string content warning in logs, got: {caplog.text}")
@pytest.mark.asyncio
async def test_fetch_published_ads_multi_page_without_last_field(self, bot:KleinanzeigenBot) -> None:

42
tests/unit/test_price_reduction.py
View File

@@ -3,14 +3,12 @@
# 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 import AdUpdateStrategy
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
@@ -24,7 +22,7 @@ class _ApplyAutoPriceReduction(Protocol):
_ad_cfg_orig:dict[str, Any],
ad_file_relative:str,
*,
mode:AdUpdateStrategy = AdUpdateStrategy.REPLACE,
mode:kleinanzeigen_bot.AdUpdateStrategy = kleinanzeigen_bot.AdUpdateStrategy.REPLACE,
) -> None:
pass
@@ -160,7 +158,9 @@ def test_apply_auto_price_reduction_disabled_emits_no_decision_logs(
@pytest.mark.unit
def test_apply_auto_price_reduction_reduces_price_by_configured_percentage(apply_auto_price_reduction:_ApplyAutoPriceReduction) -> None:
def test_apply_auto_price_reduction_reduces_price_by_configured_percentage(
caplog:pytest.LogCaptureFixture, apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
ad_cfg = SimpleNamespace(
price = 200,
auto_price_reduction = AutoPriceReductionConfig(
@@ -178,8 +178,10 @@ def test_apply_auto_price_reduction_reduces_price_by_configured_percentage(apply
)
ad_orig:dict[str, Any] = {}
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_test.yaml")
with caplog.at_level(logging.INFO):
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_test.yaml")
assert any(r.levelname == "INFO" and "Auto price reduction applied" in r.message for r in caplog.records)
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
@@ -208,8 +210,7 @@ def test_apply_auto_price_reduction_logs_unchanged_price_at_floor(
# 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 any(r.levelname == "INFO" and "Auto price reduction applied" in r.message for r in caplog.records)
assert ad_cfg.price == 90
assert ad_cfg.price_reduction_count == 1
assert "price_reduction_count" not in ad_orig
@@ -336,8 +337,7 @@ def test_apply_auto_price_reduction_waits_when_reduction_already_applied(
with caplog.at_level(logging.DEBUG, logger = "kleinanzeigen_bot"):
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 any(r.levelname == "INFO" and "already applied" in r.message and "reductions match" in r.message for r in caplog.records)
decision_message = (
"Auto price reduction decision for [ad_already.yaml]: skipped (repost delay). "
"next reduction earliest at repost >= 4 and day delay 0/0 days. repost_count=3 eligible_cycles=3 applied_cycles=3"
@@ -377,8 +377,7 @@ def test_apply_auto_price_reduction_respects_day_delay(
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)
assert any(r.levelname == "INFO" and "waiting" in r.message and "days (elapsed" in r.message for r in caplog.records)
@pytest.mark.unit
@@ -426,8 +425,7 @@ def test_apply_auto_price_reduction_delayed_when_timestamp_missing(
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)
assert any(r.levelname == "INFO" and "publish timestamp missing" in r.message for r in caplog.records)
@pytest.mark.unit
@@ -451,8 +449,8 @@ def test_fractional_reduction_increments_counter_when_price_unchanged(
apply_auto_price_reduction(ad_cfg, ad_orig, "ad_fractional.yaml")
# Price: 100 - 0.3 = 99.7, rounds to 100 (no visible change)
expected = _("Auto price reduction kept price %s after attempting %s reduction cycles") % (100, 1)
assert any(expected in message for message in caplog.messages)
# But counter should still increment for future cumulative reductions
assert any(r.levelname == "INFO" and "kept price" in r.message for r in caplog.records)
assert ad_cfg.price == 100
assert ad_cfg.price_reduction_count == 1
assert "price_reduction_count" not in ad_orig
@@ -599,7 +597,7 @@ def test_apply_modify_mode_on_update_false_leaves_base_price_when_no_prior_reduc
created_on = None,
)
apply_auto_price_reduction(ad_cfg, {}, "ad_no_update.yaml", mode = AdUpdateStrategy.MODIFY)
apply_auto_price_reduction(ad_cfg, {}, "ad_no_update.yaml", mode = kleinanzeigen_bot.AdUpdateStrategy.MODIFY)
assert ad_cfg.price == 200
assert ad_cfg.price_reduction_count == 0
@@ -627,7 +625,7 @@ def test_apply_modify_mode_applies_reduction_when_on_update_true_and_day_delay_s
monkeypatch.setattr("kleinanzeigen_bot.misc.now", lambda: reference)
apply_auto_price_reduction(ad_cfg, {}, "ad_modify.yaml", mode = AdUpdateStrategy.MODIFY)
apply_auto_price_reduction(ad_cfg, {}, "ad_modify.yaml", mode = kleinanzeigen_bot.AdUpdateStrategy.MODIFY)
assert ad_cfg.price == 150 # 200 * 0.75
assert ad_cfg.price_reduction_count == 1
@@ -651,7 +649,7 @@ def test_apply_modify_mode_skips_new_cycle_when_day_delay_not_satisfied(
monkeypatch.setattr("kleinanzeigen_bot.misc.now", lambda: reference)
apply_auto_price_reduction(ad_cfg, {}, "ad_delay_not_met.yaml", mode = AdUpdateStrategy.MODIFY)
apply_auto_price_reduction(ad_cfg, {}, "ad_delay_not_met.yaml", mode = kleinanzeigen_bot.AdUpdateStrategy.MODIFY)
assert ad_cfg.price == 200
assert ad_cfg.price_reduction_count == 0
@@ -676,7 +674,7 @@ def test_apply_modify_mode_restores_reduced_price_with_prior_reductions(
created_on = None,
)
apply_auto_price_reduction(ad_cfg, {}, "ad_restore.yaml", mode = AdUpdateStrategy.MODIFY)
apply_auto_price_reduction(ad_cfg, {}, "ad_restore.yaml", mode = kleinanzeigen_bot.AdUpdateStrategy.MODIFY)
# base=100, 2 cycles of 10%: 100*0.9=90 → 90*0.9=81
assert ad_cfg.price == 81 # restored to reduced price, not left at base 100
@@ -707,7 +705,7 @@ def test_cross_mode_update_then_publish_preserves_reduced_price(
updated_on = None,
created_on = None,
)
apply_auto_price_reduction(ad_cfg, {}, "ad_cross.yaml", mode = AdUpdateStrategy.MODIFY)
apply_auto_price_reduction(ad_cfg, {}, "ad_cross.yaml", mode = kleinanzeigen_bot.AdUpdateStrategy.MODIFY)
assert ad_cfg.price == 90
assert ad_cfg.price_reduction_count == 1
@@ -718,7 +716,7 @@ def test_cross_mode_update_then_publish_preserves_reduced_price(
ad_cfg.price_reduction_count = 1
# --- Step 3: REPLACE mode, no new repost cycle eligible ---
apply_auto_price_reduction(ad_cfg, {}, "ad_cross.yaml", mode = AdUpdateStrategy.REPLACE)
apply_auto_price_reduction(ad_cfg, {}, "ad_cross.yaml", mode = kleinanzeigen_bot.AdUpdateStrategy.REPLACE)
# Restore-first: keep the single previously applied cycle from base 100 → 90
assert ad_cfg.price == 90
@@ -753,7 +751,7 @@ def test_modify_on_update_false_restores_price(
created_on = None,
)
apply_auto_price_reduction(ad_cfg, {}, "ad_restore.yaml", mode = AdUpdateStrategy.MODIFY)
apply_auto_price_reduction(ad_cfg, {}, "ad_restore.yaml", mode = kleinanzeigen_bot.AdUpdateStrategy.MODIFY)
# Price must be restored from base 200 with one 10% reduction → 180
assert ad_cfg.price == 180

339
tests/unit/test_update_checker.py
View File

@@ -31,29 +31,12 @@ def _freeze_update_state_datetime(monkeypatch:pytest.MonkeyPatch, fixed_now:date
@classmethod
def now(cls, tz:tzinfo | None = None) -> "FixedDateTime":
base = fixed_now.replace(tzinfo = None) if tz is None else fixed_now.astimezone(tz)
return cls(
base.year,
base.month,
base.day,
base.hour,
base.minute,
base.second,
base.microsecond,
tzinfo = base.tzinfo
)
return cls(base.year, base.month, base.day, base.hour, base.minute, base.second, base.microsecond, tzinfo = base.tzinfo)
@classmethod
def utcnow(cls) -> "FixedDateTime":
base = fixed_now.astimezone(timezone.utc).replace(tzinfo = None)
return cls(
base.year,
base.month,
base.day,
base.hour,
base.minute,
base.second,
base.microsecond
)
return cls(base.year, base.month, base.day, base.hour, base.minute, base.second, base.microsecond)
datetime_module = getattr(update_check_state_module, "datetime")
monkeypatch.setattr(datetime_module, "datetime", FixedDateTime)
@@ -61,13 +44,7 @@ def _freeze_update_state_datetime(monkeypatch:pytest.MonkeyPatch, fixed_now:date
@pytest.fixture
def config() -> Config:
return Config.model_validate({
"update_check": {
"enabled": True,
"channel": "latest",
"interval": "7d"
}
})
return Config.model_validate({"update_check": {"enabled": True, "channel": "latest", "interval": "7d"}})
@pytest.fixture
@@ -92,10 +69,7 @@ class TestUpdateChecker:
def test_resolve_commitish(self, config:Config, state_file:Path) -> None:
"""Test that a commit-ish is resolved to a full hash and date."""
checker = UpdateChecker(config, state_file)
with patch(
"requests.get",
return_value = MagicMock(json = lambda: {"sha": "e7a3d46", "commit": {"author": {"date": "2025-05-18T00:00:00Z"}}})
):
with patch("requests.get", return_value = MagicMock(json = lambda: {"sha": "e7a3d46", "commit": {"author": {"date": "2025-05-18T00:00:00Z"}}})):
commit_hash, commit_date = checker._resolve_commitish("latest")
assert commit_hash == "e7a3d46"
assert commit_date == datetime(2025, 5, 18, tzinfo = timezone.utc)
@@ -120,12 +94,7 @@ class TestUpdateChecker:
assert commit_hash == "abc"
assert commit_date is None
def test_resolve_commitish_logs_warning_on_exception(
self,
config:Config,
state_file:Path,
caplog:pytest.LogCaptureFixture
) -> None:
def test_resolve_commitish_logs_warning_on_exception(self, config:Config, state_file:Path, caplog:pytest.LogCaptureFixture) -> None:
"""Test resolving a commit-ish logs a warning when the request fails."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
checker = UpdateChecker(config, state_file)
@@ -149,40 +118,15 @@ class TestUpdateChecker:
checker.check_for_updates()
mock_get.assert_not_called()
def test_check_for_updates_no_local_version(self, config:Config, state_file:Path) -> None:
"""Test that the update checker handles the case where the local version cannot be determined."""
checker = UpdateChecker(config, state_file)
with patch.object(UpdateCheckState, "should_check", return_value = True), \
patch.object(UpdateChecker, "get_local_version", return_value = None):
checker.check_for_updates() # Should not raise exception
def test_check_for_updates_logs_missing_local_version(
self,
config:Config,
state_file:Path,
caplog:pytest.LogCaptureFixture
) -> None:
def test_check_for_updates_logs_missing_local_version(self, config:Config, state_file:Path, caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker logs a warning when the local version is missing."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
checker = UpdateChecker(config, state_file)
with patch.object(UpdateCheckState, "should_check", return_value = True), \
patch.object(UpdateChecker, "get_local_version", return_value = None):
with patch.object(UpdateCheckState, "should_check", return_value = True), patch.object(UpdateChecker, "get_local_version", return_value = None):
checker.check_for_updates()
assert any("Could not determine local version." in r.getMessage() for r in caplog.records)
def test_check_for_updates_no_commit_hash(self, config:Config, state_file:Path) -> None:
"""Test that the update checker handles the case where the commit hash cannot be extracted."""
checker = UpdateChecker(config, state_file)
with patch.object(UpdateChecker, "get_local_version", return_value = "2025"):
checker.check_for_updates() # Should not raise exception
def test_check_for_updates_no_releases(self, config:Config, state_file:Path) -> None:
"""Test that the update checker handles the case where no releases are found."""
checker = UpdateChecker(config, state_file)
with patch("requests.get", return_value = MagicMock(json = list)):
checker.check_for_updates() # Should not raise exception
def test_check_for_updates_api_error(self, config:Config, state_file:Path) -> None:
"""Test that the update checker handles API errors gracefully."""
checker = UpdateChecker(config, state_file)
@@ -190,22 +134,14 @@ class TestUpdateChecker:
checker.check_for_updates() # Should not raise exception
def test_check_for_updates_latest_prerelease_warning(
self,
config:Config,
state_file:Path,
mocker:"MockerFixture",
caplog:pytest.LogCaptureFixture
self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture
) -> None:
"""Test that the update checker warns when latest points to a prerelease."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
mocker.patch.object(UpdateChecker, "_get_commit_hash", return_value = "fb00f11")
mocker.patch.object(
requests,
"get",
return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": True})
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": True}))
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
@@ -221,32 +157,16 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [
("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)),
("e7a3d46", datetime(2025, 5, 16, tzinfo = timezone.utc))
]
)
mocker.patch.object(
requests,
"get",
return_value = mocker.Mock(
json = lambda: {"tag_name": "latest", "prerelease": False}
)
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 16, tzinfo = timezone.utc))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": False}))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
print("LOG RECORDS:")
for r in caplog.records:
print(f"{r.levelname}: {r.getMessage()}")
expected = (
"You are on a different commit than the release for channel 'latest' (tag: latest). This may mean you are ahead, behind, or on a different branch. "
"Local commit: fb00f11 (2025-05-18 00:00:00 UTC), Release commit: e7a3d46 (2025-05-16 00:00:00 UTC)"
)
assert any(expected in r.getMessage() for r in caplog.records)
msgs = [r.getMessage() for r in caplog.records if r.levelno >= logging.INFO]
assert any("different commit" in m and "channel 'latest'" in m for m in msgs)
def test_check_for_updates_preview(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker correctly handles preview releases."""
@@ -257,40 +177,19 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [
("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)),
("e7a3d46", datetime(2025, 5, 16, tzinfo = timezone.utc))
]
)
mocker.patch.object(
requests,
"get",
return_value = mocker.Mock(
json = lambda: [{"tag_name": "preview", "prerelease": True, "draft": False}]
)
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 16, tzinfo = timezone.utc))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: [{"tag_name": "preview", "prerelease": True, "draft": False}]))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
print("LOG RECORDS:")
for r in caplog.records:
print(f"{r.levelname}: {r.getMessage()}")
expected = (
"You are on a different commit than the release for channel 'preview' (tag: preview). "
"This may mean you are ahead, behind, or on a different branch. "
"Local commit: fb00f11 (2025-05-18 00:00:00 UTC), Release commit: e7a3d46 (2025-05-16 00:00:00 UTC)"
)
assert any(expected in r.getMessage() for r in caplog.records)
msgs = [r.getMessage() for r in caplog.records if r.levelno >= logging.INFO]
assert any("different commit" in m and "channel 'preview'" in m for m in msgs)
def test_check_for_updates_preview_missing_prerelease(
self,
config:Config,
state_file:Path,
mocker:"MockerFixture",
caplog:pytest.LogCaptureFixture
self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture
) -> None:
"""Test that the update checker warns when no preview prerelease is available."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
@@ -298,11 +197,7 @@ class TestUpdateChecker:
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
mocker.patch.object(UpdateChecker, "_get_commit_hash", return_value = "fb00f11")
mocker.patch.object(
requests,
"get",
return_value = mocker.Mock(json = lambda: [{"tag_name": "v1", "prerelease": False, "draft": False}])
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: [{"tag_name": "v1", "prerelease": False, "draft": False}]))
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
@@ -317,37 +212,18 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [
("fb00f11", datetime(2025, 5, 16, tzinfo = timezone.utc)),
("e7a3d46", datetime(2025, 5, 18, tzinfo = timezone.utc))
]
)
mocker.patch.object(
requests,
"get",
return_value = mocker.Mock(
json = lambda: {"tag_name": "latest", "prerelease": False}
)
side_effect = [("fb00f11", datetime(2025, 5, 16, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 18, tzinfo = timezone.utc))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": False}))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
print("LOG RECORDS:")
for r in caplog.records:
print(f"{r.levelname}: {r.getMessage()}")
msgs = [r.getMessage() for r in caplog.records if r.levelno >= logging.INFO]
assert any("new version is available" in m and "channel: latest" in m for m in msgs)
expected = "A new version is available: e7a3d46 from 2025-05-18 00:00:00 UTC (current: 2025+fb00f11 from 2025-05-16 00:00:00 UTC, channel: latest)"
assert any(expected in r.getMessage() for r in caplog.records)
def test_check_for_updates_logs_release_notes(
self,
config:Config,
state_file:Path,
mocker:"MockerFixture",
caplog:pytest.LogCaptureFixture
) -> None:
def test_check_for_updates_logs_release_notes(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
"""Test that release notes are logged when present."""
caplog.set_level("INFO", logger = "kleinanzeigen_bot.update_checker")
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
@@ -355,19 +231,17 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [
("fb00f11", datetime(2025, 5, 16, tzinfo = timezone.utc)),
("e7a3d46", datetime(2025, 5, 18, tzinfo = timezone.utc))
]
side_effect = [("fb00f11", datetime(2025, 5, 16, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 18, tzinfo = timezone.utc))],
)
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
mocker.patch.object(
requests,
"get",
return_value = mocker.Mock(
json = lambda: {"tag_name": "latest", "prerelease": False, "body": "Release notes here"}
)
)
json = lambda: {
"tag_name": "latest",
"prerelease": False,
"body": "Release notes here"}))
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
@@ -382,37 +256,18 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [
("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)),
("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc))
]
)
mocker.patch.object(
requests,
"get",
return_value = mocker.Mock(
json = lambda: {"tag_name": "latest", "prerelease": False}
)
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)), ("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": False}))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
print("LOG RECORDS:")
for r in caplog.records:
print(f"{r.levelname}: {r.getMessage()}")
msgs = [r.getMessage() for r in caplog.records if r.levelno >= logging.INFO]
assert any("on the latest version" in m and "channel latest" in m for m in msgs)
expected = "You are on the latest version: 2025+fb00f11 (compared to fb00f11 in channel latest)"
assert any(expected in r.getMessage() for r in caplog.records)
def test_check_for_updates_unknown_channel(
self,
config:Config,
state_file:Path,
mocker:"MockerFixture",
caplog:pytest.LogCaptureFixture
) -> None:
def test_check_for_updates_unknown_channel(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker warns on unknown update channels."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
cast(Any, config.update_check).channel = "unknown"
@@ -427,18 +282,15 @@ class TestUpdateChecker:
mock_get.assert_not_called()
assert any("Unknown update channel: unknown" in r.getMessage() for r in caplog.records)
def test_check_for_updates_respects_interval_gate(
self,
config:Config,
state_file:Path,
caplog:pytest.LogCaptureFixture
) -> None:
def test_check_for_updates_respects_interval_gate(self, config:Config, state_file:Path, caplog:pytest.LogCaptureFixture) -> None:
"""Ensure the interval guard short-circuits update checks without touching the network."""
caplog.set_level(logging.WARNING)
with patch.object(UpdateCheckState, "should_check", return_value = False) as should_check_mock, \
patch.object(UpdateCheckState, "update_last_check") as update_last_check_mock, \
patch("requests.get") as mock_get:
with (
patch.object(UpdateCheckState, "should_check", return_value = False) as should_check_mock,
patch.object(UpdateCheckState, "update_last_check") as update_last_check_mock,
patch("requests.get") as mock_get,
):
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
@@ -465,18 +317,6 @@ class TestUpdateChecker:
state = UpdateCheckState.load(state_file)
assert state.last_check is None
def test_update_check_state_save_error(self, state_file:Path) -> None:
"""Test that saving state handles errors gracefully."""
state = UpdateCheckState()
state.last_check = datetime.now(timezone.utc)
# Make the file read-only to cause a save error
state_file.touch()
state_file.chmod(0o444)
# Should not raise an exception
state.save(state_file)
def test_update_check_state_interval_units(self, monkeypatch:pytest.MonkeyPatch) -> None:
"""Test that different interval units are handled correctly."""
state = UpdateCheckState()
@@ -542,59 +382,59 @@ class TestUpdateChecker:
assert state.should_check("2d") is False # Valid, but only 1 day elapsed
# Test maximum value (30d)
assert state.should_check("31d") is False # Too long, fallback to 7d, only 1 day elapsed
assert state.should_check("60d") is False # Too long, fallback to 7d, only 1 day elapsed
assert state.should_check("31d") is False # Too long, fallback to 7d, only 1 day elapsed
assert state.should_check("60d") is False # Too long, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 30)
assert state.should_check("30d") is False # Exactly 30 days, should_check is False
state.last_check = now - timedelta(days = 30, seconds = 1)
assert state.should_check("30d") is True # Should check if just over interval
assert state.should_check("30d") is True # Should check if just over interval
state.last_check = now - timedelta(days = 21)
assert state.should_check("21d") is False # Exactly 21 days, should_check is False
state.last_check = now - timedelta(days = 21, seconds = 1)
assert state.should_check("21d") is True # Should check if just over interval
assert state.should_check("21d") is True # Should check if just over interval
state.last_check = now - timedelta(days = 7)
assert state.should_check("7d") is False # 7 days, should_check is False
assert state.should_check("7d") is False # 7 days, should_check is False
state.last_check = now - timedelta(days = 7, seconds = 1)
assert state.should_check("7d") is True # Should check if just over interval
assert state.should_check("7d") is True # Should check if just over interval
# Test negative values
state.last_check = now - timedelta(days = 1)
assert state.should_check("-1d") is False # Negative value, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("-1d") is True # Negative value, fallback to 7d, 8 days elapsed
assert state.should_check("-1d") is True # Negative value, fallback to 7d, 8 days elapsed
# Test zero value
state.last_check = now - timedelta(days = 1)
assert state.should_check("0d") is False # Zero value, fallback to 7d, only 1 day elapsed
assert state.should_check("0d") is False # Zero value, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("0d") is True # Zero value, fallback to 7d, 8 days elapsed
assert state.should_check("0d") is True # Zero value, fallback to 7d, 8 days elapsed
# Test invalid formats
state.last_check = now - timedelta(days = 1)
assert state.should_check("invalid") is False # Invalid format, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("invalid") is True # Invalid format, fallback to 7d, 8 days elapsed
assert state.should_check("invalid") is True # Invalid format, fallback to 7d, 8 days elapsed
state.last_check = now - timedelta(days = 1)
assert state.should_check("1") is False # Missing unit, fallback to 7d, only 1 day elapsed
assert state.should_check("1") is False # Missing unit, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("1") is True # Missing unit, fallback to 7d, 8 days elapsed
assert state.should_check("1") is True # Missing unit, fallback to 7d, 8 days elapsed
state.last_check = now - timedelta(days = 1)
assert state.should_check("d") is False # Missing value, fallback to 7d, only 1 day elapsed
assert state.should_check("d") is False # Missing value, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("d") is True # Missing value, fallback to 7d, 8 days elapsed
assert state.should_check("d") is True # Missing value, fallback to 7d, 8 days elapsed
# Test unit conversions (all sub-day intervals are too short)
state.last_check = now - timedelta(days = 1)
assert state.should_check("24h") is False # 1 day in hours, fallback to 7d, only 1 day elapsed
assert state.should_check("24h") is False # 1 day in hours, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("24h") is True # 1 day in hours, fallback to 7d, 8 days elapsed
assert state.should_check("24h") is True # 1 day in hours, fallback to 7d, 8 days elapsed
state.last_check = now - timedelta(days = 1)
assert state.should_check("1440m") is False # 1 day in minutes, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("1440m") is True # 1 day in minutes, fallback to 7d, 8 days elapsed
assert state.should_check("1440m") is True # 1 day in minutes, fallback to 7d, 8 days elapsed
state.last_check = now - timedelta(days = 1)
assert state.should_check("86400s") is False # 1 day in seconds, fallback to 7d, only 1 day elapsed
state.last_check = now - timedelta(days = 8)
assert state.should_check("86400s") is True # 1 day in seconds, fallback to 7d, 8 days elapsed
assert state.should_check("86400s") is True # 1 day in seconds, fallback to 7d, 8 days elapsed
def test_update_check_state_invalid_date(self, state_file:Path) -> None:
"""Test that loading a state file with an invalid date string for last_check returns a new state (triggers ValueError)."""
@@ -602,14 +442,6 @@ class TestUpdateChecker:
state = UpdateCheckState.load(state_file)
assert state.last_check is None
def test_update_check_state_save_permission_error(self, mocker:"MockerFixture", state_file:Path) -> None:
"""Test that save handles PermissionError from dicts.save_dict."""
state = UpdateCheckState()
state.last_check = datetime.now(timezone.utc)
mocker.patch("kleinanzeigen_bot.utils.dicts.save_dict", side_effect = PermissionError)
# Should not raise
state.save(state_file)
def test_resolve_commitish_no_author(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
"""Test resolving a commit-ish when the API returns no author key."""
checker = UpdateChecker(config, state_file)
@@ -640,10 +472,7 @@ class TestUpdateChecker:
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
mocker.patch.object(UpdateChecker, "_get_commit_hash", return_value = "fb00f11")
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
mocker.patch(
"requests.get",
return_value = mocker.Mock(json = lambda: {"prerelease": False})
)
mocker.patch("requests.get", return_value = mocker.Mock(json = lambda: {"prerelease": False}))
checker.check_for_updates() # Should not raise
def test_check_for_updates_no_releases_empty(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
@@ -668,12 +497,7 @@ class TestUpdateChecker:
mocker.patch.object(UpdateChecker, "_resolve_commitish", return_value = (None, None))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
# Patch requests.get to avoid any real HTTP requests
mocker.patch(
"requests.get",
return_value = mocker.Mock(
json = lambda: {"tag_name": "latest", "prerelease": False}
)
)
mocker.patch("requests.get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": False}))
checker = UpdateChecker(config, state_file)
checker.check_for_updates()
assert any("Could not determine commit dates for comparison." in r.getMessage() for r in caplog.records)
@@ -681,9 +505,7 @@ class TestUpdateChecker:
def test_update_check_state_version_tracking(self, state_file:Path) -> None:
"""Test that version tracking works correctly."""
# Create a state with version 0 (old format)
state_file.write_text(json.dumps({
"last_check": datetime.now(timezone.utc).isoformat()
}), encoding = "utf-8")
state_file.write_text(json.dumps({"last_check": datetime.now(timezone.utc).isoformat()}), encoding = "utf-8")
# Load the state - should migrate to version 1
state = UpdateCheckState.load(state_file)
@@ -700,9 +522,7 @@ class TestUpdateChecker:
"""Test that state migration works correctly."""
# Create a state with version 0 (old format)
old_time = datetime.now(timezone.utc)
state_file.write_text(json.dumps({
"last_check": old_time.isoformat()
}), encoding = "utf-8")
state_file.write_text(json.dumps({"last_check": old_time.isoformat()}), encoding = "utf-8")
# Load the state - should migrate to version 1
state = UpdateCheckState.load(state_file)
@@ -731,40 +551,25 @@ class TestUpdateChecker:
mocker.patch("kleinanzeigen_bot.utils.dicts.save_dict", side_effect = Exception("Test error"))
state.save(state_file) # Should not raise
def test_update_check_state_load_errors(self, state_file:Path) -> None:
"""Test that load errors are handled gracefully."""
# Test invalid JSON
state_file.write_text("invalid json", encoding = "utf-8")
state = UpdateCheckState.load(state_file)
assert state.version == 1
assert state.last_check is None
# Test invalid date format
state_file.write_text(json.dumps({
"version": 1,
"last_check": "invalid-date"
}), encoding = "utf-8")
state = UpdateCheckState.load(state_file)
assert state.version == 1
assert state.last_check is None
def test_update_check_state_timezone_handling(self, state_file:Path) -> None:
"""Test that timezone handling works correctly."""
# Test loading timestamp without timezone (should assume UTC)
state_file.write_text(json.dumps({
"version": 1,
"last_check": "2024-03-20T12:00:00"
}), encoding = "utf-8")
state_file.write_text(json.dumps({"version": 1, "last_check": "2024-03-20T12:00:00"}), encoding = "utf-8")
state = UpdateCheckState.load(state_file)
assert state.last_check is not None
assert state.last_check.tzinfo == timezone.utc
assert state.last_check.hour == 12
# Test loading timestamp with different timezone (should convert to UTC)
state_file.write_text(json.dumps({
"version": 1,
"last_check": "2024-03-20T12:00:00+02:00" # 2 hours ahead of UTC
}), encoding = "utf-8")
state_file.write_text(
json.dumps(
{
"version": 1,
"last_check": "2024-03-20T12:00:00+02:00", # 2 hours ahead of UTC
}
),
encoding = "utf-8",
)
state = UpdateCheckState.load(state_file)
assert state.last_check is not None
assert state.last_check.tzinfo == timezone.utc

380
tests/unit/test_web_scraping_mixin_chrome_version.py
View File

@@ -10,6 +10,24 @@ import pytest
from kleinanzeigen_bot.utils.chrome_version_detector import ChromeVersionInfo
from kleinanzeigen_bot.utils.web_scraping_mixin import WebScrapingMixin
_ORIG_ENV_GET = os.environ.get
def _hide_pytest_current_test(key:str, default:object = None) -> object:
return default if key == "PYTEST_CURRENT_TEST" else _ORIG_ENV_GET(key, default)
@pytest.fixture
def no_pytest_guard(monkeypatch:pytest.MonkeyPatch) -> None:
r"""Make os.environ.get("PYTEST_CURRENT_TEST") return None for this test.
``monkeypatch.delenv`` cannot be used here because pytest re-sets
``PYTEST_CURRENT_TEST`` from ``(setup)`` to ``(call)`` *after* fixture
setup but before the test body runs. Patching the ``get`` method
sidesteps that timing issue.
"""
monkeypatch.setattr(os.environ, "get", _hide_pytest_current_test)
class TestWebScrapingMixinChromeVersionValidation:
"""Test Chrome version validation in WebScrapingMixin."""
@@ -20,7 +38,9 @@ class TestWebScrapingMixinChromeVersionValidation:
return WebScrapingMixin()
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_binary")
async def test_validate_chrome_version_configuration_chrome_136_plus_valid(self, mock_detect:Mock, scraper:WebScrapingMixin) -> None:
async def test_validate_chrome_version_configuration_chrome_136_plus_valid(
self, mock_detect:Mock, scraper:WebScrapingMixin, no_pytest_guard:None
) -> None:
"""Test Chrome 136+ validation with valid configuration."""
# Setup mocks
mock_detect.return_value = ChromeVersionInfo("136.0.6778.0", 136, "Chrome")
@@ -30,31 +50,21 @@ class TestWebScrapingMixinChromeVersionValidation:
scraper.browser_config.arguments = ["--remote-debugging-port=9222", "--user-data-dir=/tmp/chrome-debug"] # noqa: S108
scraper.browser_config.user_data_dir = "/tmp/chrome-debug" # noqa: S108
# Temporarily unset PYTEST_CURRENT_TEST to allow validation to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test validation
await scraper._validate_chrome_version_configuration()
try:
# Test validation
await scraper._validate_chrome_version_configuration()
# Verify detection was called correctly with timeout
assert mock_detect.call_count == 1
args, kwargs = mock_detect.call_args
assert args[0] == "/path/to/chrome"
assert kwargs["timeout"] == pytest.approx(10.0)
# Verify detection was called correctly with timeout
assert mock_detect.call_count == 1
args, kwargs = mock_detect.call_args
assert args[0] == "/path/to/chrome"
assert kwargs["timeout"] == pytest.approx(10.0)
# Verify validation passed (no exception raised)
# The validation is now done internally in _validate_chrome_136_configuration
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify validation passed (no exception raised)
# The validation is now done internally in _validate_chrome_136_configuration
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_binary")
async def test_validate_chrome_version_configuration_chrome_136_plus_invalid(
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome 136+ validation with invalid configuration."""
# Setup mocks
@@ -65,28 +75,18 @@ class TestWebScrapingMixinChromeVersionValidation:
scraper.browser_config.arguments = ["--remote-debugging-port=9222"]
scraper.browser_config.user_data_dir = None
# Temporarily unset PYTEST_CURRENT_TEST to allow validation to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test validation should log error but not raise exception due to error handling
await scraper._validate_chrome_version_configuration()
try:
# Test validation should log error but not raise exception due to error handling
await scraper._validate_chrome_version_configuration()
# Verify detection call and logged error
assert mock_detect.call_count == 1
_, kwargs = mock_detect.call_args
assert kwargs["timeout"] == pytest.approx(10.0)
assert "Chrome 136+ configuration validation failed" in caplog.text
assert "Chrome 136+ requires --user-data-dir" in caplog.text
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify detection call and logged error
assert mock_detect.call_count == 1
_, kwargs = mock_detect.call_args
assert kwargs["timeout"] == pytest.approx(10.0)
assert "Chrome 136+ configuration validation failed" in caplog.text
assert "Chrome 136+ requires --user-data-dir" in caplog.text
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_binary")
async def test_validate_chrome_version_configuration_chrome_pre_136(self, mock_detect:Mock, scraper:WebScrapingMixin) -> None:
async def test_validate_chrome_version_configuration_chrome_pre_136(self, mock_detect:Mock, scraper:WebScrapingMixin, no_pytest_guard:None) -> None:
"""Test Chrome pre-136 validation (no special requirements)."""
# Setup mocks
mock_detect.return_value = ChromeVersionInfo("120.0.6099.109", 120, "Chrome")
@@ -96,28 +96,18 @@ class TestWebScrapingMixinChromeVersionValidation:
scraper.browser_config.arguments = ["--remote-debugging-port=9222"]
scraper.browser_config.user_data_dir = None
# Temporarily unset PYTEST_CURRENT_TEST to allow validation to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test validation should pass without validation
await scraper._validate_chrome_version_configuration()
try:
# Test validation should pass without validation
await scraper._validate_chrome_version_configuration()
# Verify detection was called but no validation
assert mock_detect.call_count == 1
_, kwargs = mock_detect.call_args
assert kwargs["timeout"] == pytest.approx(10.0)
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify detection was called but no validation
assert mock_detect.call_count == 1
_, kwargs = mock_detect.call_args
assert kwargs["timeout"] == pytest.approx(10.0)
@patch("kleinanzeigen_bot.utils.chrome_version_detector.detect_chrome_version_from_binary")
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_remote_debugging")
async def test_validate_chrome_version_logs_remote_detection(
self, mock_remote:Mock, mock_binary:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_remote:Mock, mock_binary:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""When a remote browser responds, the detected version should be logged."""
mock_remote.return_value = ChromeVersionInfo("136.0.6778.0", 136, "Chrome")
@@ -126,7 +116,7 @@ class TestWebScrapingMixinChromeVersionValidation:
scraper.browser_config.binary_location = "/path/to/chrome"
caplog.set_level("DEBUG")
with patch.dict(os.environ, {}, clear = True), patch.object(scraper, "_check_port_with_retry", return_value = True):
with patch.object(scraper, "_check_port_with_retry", return_value = True):
await scraper._validate_chrome_version_configuration()
assert "Detected version from existing browser" in caplog.text
@@ -146,7 +136,7 @@ class TestWebScrapingMixinChromeVersionValidation:
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_binary")
async def test_validate_chrome_version_configuration_detection_fails(
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome version validation when detection fails."""
# Setup mocks
@@ -155,26 +145,16 @@ class TestWebScrapingMixinChromeVersionValidation:
# Configure scraper
scraper.browser_config.binary_location = "/path/to/chrome"
# Temporarily unset PYTEST_CURRENT_TEST to allow validation to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test validation should pass without validation
await scraper._validate_chrome_version_configuration()
try:
# Test validation should pass without validation
await scraper._validate_chrome_version_configuration()
# Verify detection was called
assert mock_detect.call_count == 1
_, kwargs = mock_detect.call_args
assert kwargs["timeout"] == pytest.approx(10.0)
# Verify detection was called
assert mock_detect.call_count == 1
_, kwargs = mock_detect.call_args
assert kwargs["timeout"] == pytest.approx(10.0)
# Verify debug log message (line 824)
assert "Could not detect browser version, skipping validation" in caplog.text
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify debug log message (line 824)
assert "Could not detect browser version, skipping validation" in caplog.text
class TestWebScrapingMixinChromeVersionDiagnostics:
@@ -188,7 +168,7 @@ class TestWebScrapingMixinChromeVersionDiagnostics:
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.get_chrome_version_diagnostic_info")
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.validate_chrome_136_configuration")
def test_diagnose_chrome_version_issues_binary_detection(
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome version diagnostics with binary detection."""
# Setup mocks
@@ -204,36 +184,26 @@ class TestWebScrapingMixinChromeVersionDiagnostics:
scraper.browser_config.binary_location = "/path/to/chrome"
scraper.browser_config.arguments = ["--remote-debugging-port=9222", "--user-data-dir=/tmp/chrome-debug"]
# Temporarily unset PYTEST_CURRENT_TEST to allow diagnostics to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test diagnostics
scraper._diagnose_chrome_version_issues(9222)
try:
# Test diagnostics
scraper._diagnose_chrome_version_issues(9222)
# Verify logs
assert "Chrome version from binary: 136.0.6778.0 (major: 136)" in caplog.text
assert "Chrome 136+ detected - security validation required" in caplog.text
# Verify logs
assert "Chrome version from binary: 136.0.6778.0 (major: 136)" in caplog.text
assert "Chrome 136+ detected - security validation required" in caplog.text
# Verify mocks were called
assert mock_get_diagnostic.call_count == 1
kwargs = mock_get_diagnostic.call_args.kwargs
assert kwargs["binary_path"] == "/path/to/chrome"
assert kwargs["remote_port"] == 9222
assert kwargs["remote_host"] == "127.0.0.1"
assert kwargs["remote_timeout"] > 0
assert kwargs["binary_timeout"] > 0
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify mocks were called
assert mock_get_diagnostic.call_count == 1
kwargs = mock_get_diagnostic.call_args.kwargs
assert kwargs["binary_path"] == "/path/to/chrome"
assert kwargs["remote_port"] == 9222
assert kwargs["remote_host"] == "127.0.0.1"
assert kwargs["remote_timeout"] > 0
assert kwargs["binary_timeout"] > 0
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.get_chrome_version_diagnostic_info")
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.validate_chrome_136_configuration")
def test_diagnose_chrome_version_issues_remote_detection(
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome version diagnostics with remote detection."""
# Setup mocks
@@ -249,29 +219,21 @@ class TestWebScrapingMixinChromeVersionDiagnostics:
scraper.browser_config.binary_location = "/path/to/chrome"
scraper.browser_config.arguments = ["--remote-debugging-port=9222"]
# Temporarily unset PYTEST_CURRENT_TEST to allow diagnostics to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test diagnostics
scraper._diagnose_chrome_version_issues(9222)
try:
# Test diagnostics
scraper._diagnose_chrome_version_issues(9222)
# Verify logs
assert "(info) Chrome version from remote debugging: 136.0.6778.0 (major: 136)" in caplog.text
assert "Remote Chrome 136+ detected - validating configuration" in caplog.text
assert "Chrome 136+ configuration validation failed" in caplog.text
# Verify logs
assert "(info) Chrome version from remote debugging: 136.0.6778.0 (major: 136)" in caplog.text
assert "Remote Chrome 136+ detected - validating configuration" in caplog.text
assert "Chrome 136+ configuration validation failed" in caplog.text
# Verify validation was called
mock_validate.assert_called_once_with(["--remote-debugging-port=9222"], None)
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify validation was called
mock_validate.assert_called_once_with(["--remote-debugging-port=9222"], None)
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.get_chrome_version_diagnostic_info")
def test_diagnose_chrome_version_issues_no_detection(self, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture) -> None:
def test_diagnose_chrome_version_issues_no_detection(
self, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome version diagnostics with no detection."""
# Setup mocks
mock_get_diagnostic.return_value = {"binary_detection": None, "remote_detection": None, "chrome_136_plus_detected": False, "recommendations": []}
@@ -279,26 +241,16 @@ class TestWebScrapingMixinChromeVersionDiagnostics:
# Configure scraper
scraper.browser_config.binary_location = "/path/to/chrome"
# Temporarily unset PYTEST_CURRENT_TEST to allow diagnostics to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test diagnostics
scraper._diagnose_chrome_version_issues(0)
try:
# Test diagnostics
scraper._diagnose_chrome_version_issues(0)
# Verify no Chrome version logs
assert "Chrome version from binary" not in caplog.text
assert "Chrome version from remote debugging" not in caplog.text
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify no Chrome version logs
assert "Chrome version from binary" not in caplog.text
assert "Chrome version from remote debugging" not in caplog.text
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.get_chrome_version_diagnostic_info")
def test_diagnose_chrome_version_issues_chrome_136_plus_recommendations(
self, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome version diagnostics with Chrome 136+ recommendations."""
# Setup mocks
@@ -312,27 +264,17 @@ class TestWebScrapingMixinChromeVersionDiagnostics:
# Configure scraper
scraper.browser_config.binary_location = "/path/to/chrome"
# Temporarily unset PYTEST_CURRENT_TEST to allow diagnostics to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test diagnostics
scraper._diagnose_chrome_version_issues(0)
try:
# Test diagnostics
scraper._diagnose_chrome_version_issues(0)
# Verify recommendations
assert "Chrome/Edge 136+ security changes require --user-data-dir for remote debugging" in caplog.text
assert "https://developer.chrome.com/blog/remote-debugging-port" in caplog.text
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify recommendations
assert "Chrome/Edge 136+ security changes require --user-data-dir for remote debugging" in caplog.text
assert "https://developer.chrome.com/blog/remote-debugging-port" in caplog.text
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.get_chrome_version_diagnostic_info")
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.validate_chrome_136_configuration")
def test_diagnose_chrome_version_issues_binary_pre_136(
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome version diagnostics with pre-136 binary detection (lines 832-849)."""
# Setup mocks to ensure exact branch coverage
@@ -351,34 +293,24 @@ class TestWebScrapingMixinChromeVersionDiagnostics:
# Configure scraper
scraper.browser_config.binary_location = "/path/to/chrome"
# Temporarily unset PYTEST_CURRENT_TEST to allow diagnostics to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test diagnostics
scraper._diagnose_chrome_version_issues(0)
try:
# Test diagnostics
scraper._diagnose_chrome_version_issues(0)
# Verify pre-136 log message (lines 832-849)
assert "Chrome pre-136 detected - no special security requirements" in caplog.text
# Verify pre-136 log message (lines 832-849)
assert "Chrome pre-136 detected - no special security requirements" in caplog.text
# Verify that the diagnostic function was called with correct parameters
assert mock_get_diagnostic.call_count == 1
kwargs = mock_get_diagnostic.call_args.kwargs
assert kwargs["binary_path"] == "/path/to/chrome"
assert kwargs["remote_port"] is None
assert kwargs["remote_timeout"] > 0
assert kwargs["binary_timeout"] > 0
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify that the diagnostic function was called with correct parameters
assert mock_get_diagnostic.call_count == 1
kwargs = mock_get_diagnostic.call_args.kwargs
assert kwargs["binary_path"] == "/path/to/chrome"
assert kwargs["remote_port"] is None
assert kwargs["remote_timeout"] > 0
assert kwargs["binary_timeout"] > 0
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.get_chrome_version_diagnostic_info")
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.validate_chrome_136_configuration")
def test_diagnose_chrome_version_issues_remote_validation_passes(
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_validate:Mock, mock_get_diagnostic:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome version diagnostics with remote validation passing (line 846)."""
# Setup mocks
@@ -395,27 +327,17 @@ class TestWebScrapingMixinChromeVersionDiagnostics:
scraper.browser_config.arguments = ["--remote-debugging-port=9222", "--user-data-dir=/tmp/chrome-debug"] # noqa: S108
scraper.browser_config.user_data_dir = "/tmp/chrome-debug" # noqa: S108
# Temporarily unset PYTEST_CURRENT_TEST to allow diagnostics to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test diagnostics
scraper._diagnose_chrome_version_issues(9222)
try:
# Test diagnostics
scraper._diagnose_chrome_version_issues(9222)
# Verify validation passed log message (line 846)
assert "Chrome 136+ configuration validation passed" in caplog.text
# Verify validation passed log message (line 846)
assert "Chrome 136+ configuration validation passed" in caplog.text
# Verify validation was called with correct arguments
mock_validate.assert_called_once_with(
["--remote-debugging-port=9222", "--user-data-dir=/tmp/chrome-debug"], # noqa: S108
"/tmp/chrome-debug", # noqa: S108
)
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify validation was called with correct arguments
mock_validate.assert_called_once_with(
["--remote-debugging-port=9222", "--user-data-dir=/tmp/chrome-debug"], # noqa: S108
"/tmp/chrome-debug", # noqa: S108
)
class TestWebScrapingMixinIntegration:
@@ -465,7 +387,7 @@ class TestWebScrapingMixinIntegration:
# Verify Chrome diagnostics was called
mock_diagnose.assert_called_once_with(9222)
def test_backward_compatibility_old_configs_still_work(self) -> None:
def test_backward_compatibility_old_configs_still_work(self, no_pytest_guard:None) -> None:
"""Test that old configurations without Chrome 136+ validation still work."""
# Create a scraper with old-style config (no user_data_dir)
scraper = WebScrapingMixin()
@@ -478,26 +400,12 @@ class TestWebScrapingMixinIntegration:
with patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_binary") as mock_detect:
mock_detect.return_value = ChromeVersionInfo("120.0.6099.109", 120, "Chrome")
# Temporarily unset PYTEST_CURRENT_TEST to allow validation to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
try:
# This should not raise an exception for pre-136 Chrome
asyncio.run(scraper._validate_chrome_version_configuration())
# Verify that the validation passed (no exception raised)
# The method should log that pre-136 Chrome was detected
# and no special validation is required
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# This should not raise an exception for pre-136 Chrome
asyncio.run(scraper._validate_chrome_version_configuration())
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_binary")
async def test_validate_chrome_136_configuration_with_whitespace_user_data_dir(
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome 136+ validation correctly handles whitespace-only user_data_dir."""
# Setup mocks
@@ -508,29 +416,19 @@ class TestWebScrapingMixinIntegration:
scraper.browser_config.arguments = ["--remote-debugging-port=9222"]
scraper.browser_config.user_data_dir = " " # Only whitespace
# Temporarily unset PYTEST_CURRENT_TEST to allow validation to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test validation should fail because whitespace-only is treated as empty
await scraper._validate_chrome_version_configuration()
try:
# Test validation should fail because whitespace-only is treated as empty
await scraper._validate_chrome_version_configuration()
# Verify detection was called
assert mock_detect.call_count == 1
# Verify detection was called
assert mock_detect.call_count == 1
# Verify error was logged
assert "Chrome 136+ configuration validation failed" in caplog.text
assert "Chrome 136+ requires --user-data-dir" in caplog.text
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify error was logged
assert "Chrome 136+ configuration validation failed" in caplog.text
assert "Chrome 136+ requires --user-data-dir" in caplog.text
@patch("kleinanzeigen_bot.utils.web_scraping_mixin.detect_chrome_version_from_binary")
async def test_validate_chrome_136_configuration_with_valid_user_data_dir(
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture
self, mock_detect:Mock, scraper:WebScrapingMixin, caplog:pytest.LogCaptureFixture, no_pytest_guard:None
) -> None:
"""Test Chrome 136+ validation passes with valid user_data_dir."""
# Setup mocks
@@ -541,21 +439,11 @@ class TestWebScrapingMixinIntegration:
scraper.browser_config.arguments = ["--remote-debugging-port=9222"]
scraper.browser_config.user_data_dir = "/tmp/valid-profile" # noqa: S108
# Temporarily unset PYTEST_CURRENT_TEST to allow validation to run
original_env = os.environ.get("PYTEST_CURRENT_TEST")
if "PYTEST_CURRENT_TEST" in os.environ:
del os.environ["PYTEST_CURRENT_TEST"]
# Test validation should pass
await scraper._validate_chrome_version_configuration()
try:
# Test validation should pass
await scraper._validate_chrome_version_configuration()
# Verify detection was called
assert mock_detect.call_count == 1
# Verify detection was called
assert mock_detect.call_count == 1
# Verify success was logged
assert "Chrome 136+ configuration validation passed" in caplog.text
finally:
# Restore environment
if original_env:
os.environ["PYTEST_CURRENT_TEST"] = original_env
# Verify success was logged
assert "Chrome 136+ configuration validation passed" in caplog.text