# 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 from collections.abc import Callable, Generator from contextlib import redirect_stdout from datetime import timedelta from pathlib import Path, PureWindowsPath from typing import Any, cast from unittest.mock import AsyncMock, MagicMock, patch import pytest from nodriver.core.connection import ProtocolException from pydantic import ValidationError from kleinanzeigen_bot import LOG, PUBLISH_MAX_RETRIES, AdUpdateStrategy, KleinanzeigenBot, LoginState, misc from kleinanzeigen_bot._version import __version__ from kleinanzeigen_bot.model.ad_model import Ad from kleinanzeigen_bot.model.config_model import AdDefaults, Config, DiagnosticsConfig, PublishingConfig from kleinanzeigen_bot.utils import dicts, loggers, xdg_paths from kleinanzeigen_bot.utils.exceptions import PublishSubmissionUncertainError from kleinanzeigen_bot.utils.web_scraping_mixin import By, Element @pytest.fixture def mock_page() -> MagicMock: """Provide a mock page object for testing.""" mock = MagicMock() mock.sleep = AsyncMock() mock.evaluate = AsyncMock() mock.click = AsyncMock() mock.type = AsyncMock() mock.select = AsyncMock() mock.wait_for_selector = AsyncMock() mock.wait_for_navigation = AsyncMock() mock.wait_for_load_state = AsyncMock() mock.content = AsyncMock(return_value = "") mock.goto = AsyncMock() mock.close = AsyncMock() return mock @pytest.fixture def base_ad_config() -> dict[str, Any]: """Provide a base ad configuration that can be used across tests.""" return { "id": None, "title": "Test Title", "description": "Test Description", "type": "OFFER", "price_type": "FIXED", "price": 100, "shipping_type": "SHIPPING", "shipping_options": [], "category": "160", "special_attributes": {}, "sell_directly": False, "images": [], "active": True, "republication_interval": 7, "created_on": None, "contact": {"name": "Test User", "zipcode": "12345", "location": "Test City", "street": "", "phone": ""}, } def remove_fields(config:dict[str, Any], *fields:str) -> dict[str, Any]: """Create a new ad configuration with specified fields removed. Args: config: The configuration to remove fields from *fields: Field names to remove Returns: A new ad configuration dictionary with specified fields removed """ result = copy.deepcopy(config) for field in fields: if "." in field: # Handle nested fields (e.g., "contact.phone") parts = field.split(".", maxsplit = 1) current = result for part in parts[:-1]: if part in current: current = current[part] if parts[-1] in current: del current[parts[-1]] elif field in result: del result[field] return result @pytest.fixture def minimal_ad_config(base_ad_config:dict[str, Any]) -> dict[str, Any]: """Provide a minimal ad configuration with only required fields.""" return remove_fields(base_ad_config, "id", "created_on", "shipping_options", "special_attributes", "contact.street", "contact.phone") @pytest.fixture def mock_config_setup(test_bot:KleinanzeigenBot) -> Generator[None]: """Provide a centralized mock configuration setup for tests. This fixture mocks load_config and other essential configuration-related methods.""" with ( patch.object(test_bot, "load_config"), patch.object(test_bot, "create_browser_session", new_callable = AsyncMock), patch.object(test_bot, "login", new_callable = AsyncMock), patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, ): # Mock the web request for published ads mock_request.return_value = {"content": '{"ads": []}'} yield def _make_fake_resolve_workspace( captured_mode:dict[str, xdg_paths.InstallationMode | None], workspace:xdg_paths.Workspace, ) -> Callable[..., xdg_paths.Workspace]: """Create a fake resolve_workspace that captures the workspace_mode argument.""" def fake_resolve_workspace( config_arg:str | None, logfile_arg:str | None, *, workspace_mode:xdg_paths.InstallationMode | None, logfile_explicitly_provided:bool, log_basename:str, ) -> xdg_paths.Workspace: captured_mode["value"] = workspace_mode return workspace return fake_resolve_workspace class TestKleinanzeigenBotInitialization: """Tests for KleinanzeigenBot initialization and basic functionality.""" def test_constructor_initializes_default_values(self, test_bot:KleinanzeigenBot) -> None: """Verify that constructor sets all default values correctly.""" assert test_bot.root_url == "https://www.kleinanzeigen.de" assert isinstance(test_bot.config, Config) assert test_bot.command == "help" assert test_bot.ads_selector == "due" assert test_bot.keep_old_ads is False assert test_bot.log_file_path is not None assert test_bot.file_log is None def test_get_version_returns_correct_version(self, test_bot:KleinanzeigenBot) -> None: """Verify version retrieval works correctly.""" with patch("kleinanzeigen_bot.__version__", "1.2.3"): assert test_bot.get_version() == "1.2.3" def test_resolve_workspace_skips_help(self, test_bot:KleinanzeigenBot) -> None: """Ensure workspace resolution returns early for help.""" test_bot.command = "help" test_bot.workspace = None test_bot._resolve_workspace() assert test_bot.workspace is None def test_resolve_workspace_skips_create_config(self, test_bot:KleinanzeigenBot) -> None: """Ensure workspace resolution returns early for create-config.""" test_bot.command = "create-config" test_bot.workspace = None test_bot._resolve_workspace() assert test_bot.workspace is None def test_resolve_workspace_exits_on_workspace_resolution_error(self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture) -> None: """Workspace resolution errors should terminate with code 2.""" caplog.set_level(logging.ERROR) test_bot.command = "verify" with ( patch("kleinanzeigen_bot.xdg_paths.resolve_workspace", side_effect = ValueError("workspace error")), pytest.raises(SystemExit) as exc_info, ): test_bot._resolve_workspace() assert exc_info.value.code == 2 assert "workspace error" in caplog.text def test_resolve_workspace_fails_fast_when_config_parent_cannot_be_created(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Workspace resolution should fail immediately when config directory creation fails.""" test_bot.command = "verify" workspace = xdg_paths.Workspace.for_config(tmp_path / "blocked" / "config.yaml", "kleinanzeigen-bot") with ( patch("kleinanzeigen_bot.xdg_paths.resolve_workspace", return_value = workspace), patch("kleinanzeigen_bot.xdg_paths.ensure_directory", side_effect = OSError("mkdir denied")), pytest.raises(OSError, match = "mkdir denied"), ): test_bot._resolve_workspace() def test_resolve_workspace_programmatic_config_in_xdg_defaults_to_xdg(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Programmatic config_file_path in XDG config tree should default workspace mode to xdg.""" test_bot.command = "verify" xdg_dirs = { "config": tmp_path / "xdg-config" / xdg_paths.APP_NAME, "state": tmp_path / "xdg-state" / xdg_paths.APP_NAME, "cache": tmp_path / "xdg-cache" / xdg_paths.APP_NAME, } for path in xdg_dirs.values(): path.mkdir(parents = True, exist_ok = True) config_path = xdg_dirs["config"] / "config.yaml" config_path.touch() test_bot.config_file_path = str(config_path) workspace = xdg_paths.Workspace.for_config(tmp_path / "resolved" / "config.yaml", "kleinanzeigen-bot") captured_mode:dict[str, xdg_paths.InstallationMode | None] = {"value": None} with ( patch("kleinanzeigen_bot.xdg_paths.get_xdg_base_dir", side_effect = lambda category: xdg_dirs[category]), patch("kleinanzeigen_bot.xdg_paths.resolve_workspace", side_effect = _make_fake_resolve_workspace(captured_mode, workspace)), patch("kleinanzeigen_bot.xdg_paths.ensure_directory"), ): test_bot._resolve_workspace() assert captured_mode["value"] == "xdg" def test_resolve_workspace_programmatic_config_outside_xdg_defaults_to_portable(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Programmatic config_file_path outside XDG config tree should default workspace mode to portable.""" test_bot.command = "verify" xdg_dirs = { "config": tmp_path / "xdg-config" / xdg_paths.APP_NAME, "state": tmp_path / "xdg-state" / xdg_paths.APP_NAME, "cache": tmp_path / "xdg-cache" / xdg_paths.APP_NAME, } for path in xdg_dirs.values(): path.mkdir(parents = True, exist_ok = True) config_path = tmp_path / "external" / "config.yaml" config_path.parent.mkdir(parents = True, exist_ok = True) config_path.touch() test_bot.config_file_path = str(config_path) workspace = xdg_paths.Workspace.for_config(tmp_path / "resolved" / "config.yaml", "kleinanzeigen-bot") captured_mode:dict[str, xdg_paths.InstallationMode | None] = {"value": None} with ( patch("kleinanzeigen_bot.xdg_paths.get_xdg_base_dir", side_effect = lambda category: xdg_dirs[category]), patch("kleinanzeigen_bot.xdg_paths.resolve_workspace", side_effect = _make_fake_resolve_workspace(captured_mode, workspace)), patch("kleinanzeigen_bot.xdg_paths.ensure_directory"), ): test_bot._resolve_workspace() assert captured_mode["value"] == "portable" def test_create_default_config_creates_parent_without_workspace(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """create_default_config should create parent directories when no workspace is set.""" config_path = tmp_path / "nested" / "config.yaml" test_bot.workspace = None test_bot.config_file_path = str(config_path) test_bot.create_default_config() assert config_path.exists() @pytest.mark.asyncio @pytest.mark.parametrize("command", ["verify", "update-check", "update-content-hash", "publish", "delete", "download"]) async def test_run_uses_workspace_state_file_for_update_checker(self, test_bot:KleinanzeigenBot, command:str, tmp_path:Path) -> None: """Ensure UpdateChecker is initialized with the workspace state file.""" update_checker_calls:list[tuple[Config, Path]] = [] class DummyUpdateChecker: def __init__(self, config:Config, state_file:Path) -> None: update_checker_calls.append((config, state_file)) def check_for_updates(self, *_args:Any, **_kwargs:Any) -> None: return None def set_workspace() -> None: test_bot.workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot") with ( patch.object(test_bot, "configure_file_logging"), patch.object(test_bot, "load_config"), patch.object(test_bot, "load_ads", return_value = []), patch.object(test_bot, "create_browser_session", new_callable = AsyncMock), patch.object(test_bot, "login", new_callable = AsyncMock), patch.object(test_bot, "download_ads", new_callable = AsyncMock), patch.object(test_bot, "close_browser_session"), patch.object(test_bot, "_resolve_workspace", side_effect = set_workspace), patch("kleinanzeigen_bot.UpdateChecker", DummyUpdateChecker), ): await test_bot.run(["app", command]) expected_state_path = (tmp_path / "config.yaml").resolve().parent / ".temp" / "update_check_state.json" assert update_checker_calls == [(test_bot.config, expected_state_path)] @pytest.mark.asyncio async def test_download_ads_passes_download_dir_and_published_ads(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Ensure download_ads wires download_dir and published_ads_by_id into AdExtractor.""" test_bot.workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot") test_bot.ads_selector = "all" test_bot.browser = MagicMock() extractor_mock = MagicMock() extractor_mock.extract_own_ads_urls = AsyncMock(return_value = []) mock_published_ads = [{"id": 123, "buyNowEligible": True}, {"id": 456, "buyNowEligible": False}] with ( patch.object(test_bot, "_fetch_published_ads", new_callable = AsyncMock, return_value = mock_published_ads), patch("kleinanzeigen_bot.extract.AdExtractor", return_value = extractor_mock) as mock_extractor, ): await test_bot.download_ads() # Verify published_ads_by_id is built correctly and passed to extractor mock_extractor.assert_called_once_with( test_bot.browser, test_bot.config, test_bot.workspace.download_dir, published_ads_by_id = {123: mock_published_ads[0], 456: mock_published_ads[1]}, ) class TestKleinanzeigenBotLogging: """Tests for logging functionality.""" def test_configure_file_logging_adds_and_removes_handlers(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Ensure file logging registers a handler and cleans it up afterward.""" log_path = tmp_path / "bot.log" test_bot.log_file_path = str(log_path) root_logger = logging.getLogger() initial_handlers = list(root_logger.handlers) test_bot.configure_file_logging() assert test_bot.file_log is not None assert log_path.exists() assert len(root_logger.handlers) == len(initial_handlers) + 1 test_bot.file_log.close() assert test_bot.file_log.is_closed() assert len(root_logger.handlers) == len(initial_handlers) def test_configure_file_logging_skips_when_path_missing(self, test_bot:KleinanzeigenBot) -> None: """Ensure no handler is added when no log path is configured.""" root_logger = logging.getLogger() initial_handlers = list(root_logger.handlers) test_bot.log_file_path = None test_bot.configure_file_logging() assert test_bot.file_log is None assert list(root_logger.handlers) == initial_handlers class TestKleinanzeigenBotCommandLine: """Tests for command line argument parsing.""" @pytest.mark.parametrize( ("args", "expected_command", "expected_selector", "expected_keep_old"), [ (["publish", "--ads=all"], "publish", "all", False), (["verify"], "verify", "due", False), (["download", "--ads=12345"], "download", "12345", False), (["publish", "--force"], "publish", "all", False), (["publish", "--keep-old"], "publish", "due", True), (["publish", "--ads=all", "--keep-old"], "publish", "all", True), (["download", "--ads=new"], "download", "new", False), (["publish", "--ads=changed"], "publish", "changed", False), (["publish", "--ads=changed,due"], "publish", "changed,due", False), (["publish", "--ads=changed,new"], "publish", "changed,new", False), (["version"], "version", "due", False), ], ) def test_parse_args_handles_valid_arguments( self, test_bot:KleinanzeigenBot, args:list[str], expected_command:str, expected_selector:str, expected_keep_old:bool ) -> None: """Verify that valid command line arguments are parsed correctly.""" test_bot.parse_args(["dummy"] + args) # Add dummy arg to simulate sys.argv[0] assert test_bot.command == expected_command assert test_bot.ads_selector == expected_selector assert test_bot.keep_old_ads == expected_keep_old def test_parse_args_handles_help_command(self, test_bot:KleinanzeigenBot) -> None: """Verify that help command is handled correctly.""" buf = io.StringIO() with pytest.raises(SystemExit) as exc_info, redirect_stdout(buf): test_bot.parse_args(["dummy", "--help"]) assert exc_info.value.code == 0 stdout = buf.getvalue() assert "publish" in stdout assert "verify" in stdout assert "help" in stdout assert "version" in stdout assert "--verbose" in stdout def test_parse_args_handles_invalid_arguments(self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture) -> None: """Verify that invalid arguments are handled correctly.""" caplog.set_level(logging.ERROR) with pytest.raises(SystemExit) as exc_info: test_bot.parse_args(["dummy", "--invalid-option"]) assert exc_info.value.code == 2 assert any( record.levelno == logging.ERROR and ("--invalid-option not recognized" in record.getMessage() or "Option --invalid-option unbekannt" in record.getMessage()) for record in caplog.records ) assert any(("--invalid-option not recognized" in m) or ("Option --invalid-option unbekannt" in m) for m in caplog.messages) def test_parse_args_handles_verbose_flag(self, test_bot:KleinanzeigenBot) -> None: """Verify that verbose flag sets correct log level.""" test_bot.parse_args(["dummy", "--verbose"]) assert loggers.is_debug(LOG) def test_parse_args_handles_config_path(self, test_bot:KleinanzeigenBot, test_data_dir:str) -> None: """Verify that config path is set correctly.""" config_path = Path(test_data_dir) / "custom_config.yaml" test_bot.parse_args(["dummy", "--config", str(config_path)]) assert test_bot.config_file_path == str(config_path.absolute()) class TestKleinanzeigenBotConfiguration: """Tests for configuration loading and validation.""" def test_load_config_handles_missing_file(self, test_bot:KleinanzeigenBot, test_data_dir:str) -> None: """Verify that loading a missing config file creates default config. No info log is expected anymore.""" config_path = Path(test_data_dir) / "missing_config.yaml" config_path.unlink(missing_ok = True) test_bot.config_file_path = str(config_path) test_bot.load_config() assert config_path.exists() def test_load_config_validates_required_fields(self, test_bot:KleinanzeigenBot, test_data_dir:str) -> None: """Verify that config validation checks required fields.""" config_path = Path(test_data_dir) / "config.yaml" config_content = """ login: username: dummy_user # Missing password """ with open(config_path, "w", encoding = "utf-8") as f: f.write(config_content) test_bot.config_file_path = str(config_path) with pytest.raises(ValidationError) as exc_info: test_bot.load_config() assert "login.username" not in str(exc_info.value) assert "login.password" in str(exc_info.value) class TestKleinanzeigenBotAuthentication: """Tests for login and authentication functionality.""" @pytest.mark.asyncio async def test_is_logged_in_returns_true_when_logged_in(self, test_bot:KleinanzeigenBot) -> None: """Verify that login check returns true when logged in.""" with patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, return_value = ("Welcome dummy_user", 0), ): assert await test_bot.is_logged_in() is True @pytest.mark.asyncio async def test_is_logged_in_returns_true_with_alternative_element(self, test_bot:KleinanzeigenBot) -> None: """Verify that login check returns true when logged in with alternative element.""" with patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, return_value = ("angemeldet als: dummy_user", 1), ): assert await test_bot.is_logged_in() is True @pytest.mark.asyncio async def test_is_logged_in_returns_false_when_not_logged_in(self, test_bot:KleinanzeigenBot) -> None: """Verify that login check returns false when not logged in.""" with ( patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, side_effect = [("nicht-eingeloggt", 0), ("kein user signal", 0)], ), patch.object(test_bot, "_has_logged_out_cta", new_callable = AsyncMock, return_value = False), ): assert await test_bot.is_logged_in() is False @pytest.mark.asyncio async def test_has_logged_out_cta_requires_visible_candidate(self, test_bot:KleinanzeigenBot) -> None: matched_element = MagicMock(spec = Element) with ( patch.object(test_bot, "web_find_first_available", new_callable = AsyncMock, return_value = (matched_element, 0)), patch.object(test_bot, "_extract_visible_text", new_callable = AsyncMock, return_value = ""), ): assert await test_bot._has_logged_out_cta() is False @pytest.mark.asyncio async def test_has_logged_out_cta_accepts_visible_candidate(self, test_bot:KleinanzeigenBot) -> None: matched_element = MagicMock(spec = Element) with ( patch.object(test_bot, "web_find_first_available", new_callable = AsyncMock, return_value = (matched_element, 0)), patch.object(test_bot, "_extract_visible_text", new_callable = AsyncMock, return_value = "Einloggen"), ): assert await test_bot._has_logged_out_cta() is True @pytest.mark.asyncio async def test_is_logged_in_uses_selector_group_timeout_key(self, test_bot:KleinanzeigenBot) -> None: """Verify login detection uses selector-group lookup with login_detection timeout key.""" with patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, side_effect = [TimeoutError(), ("Welcome dummy_user", 0)], ) as group_text: assert await test_bot.is_logged_in(include_probe = False) is True group_text.assert_awaited() assert any(call.kwargs.get("timeout") == test_bot._timeout("login_detection") for call in group_text.await_args_list) @pytest.mark.asyncio async def test_is_logged_in_runs_full_selector_group_before_cta_precheck(self, test_bot:KleinanzeigenBot) -> None: """Quick CTA checks must not short-circuit before full logged-in selector checks.""" with patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, side_effect = [TimeoutError(), ("Welcome dummy_user", 0)], ) as group_text: assert await test_bot.is_logged_in(include_probe = False) is True group_text.assert_awaited() assert group_text.await_count >= 1 @pytest.mark.asyncio async def test_is_logged_in_short_circuits_before_cta_check_when_quick_user_signal_matches(self, test_bot:KleinanzeigenBot) -> None: """Logged-in quick pre-check should win even if incidental login links exist elsewhere.""" with patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, return_value = ("angemeldet als: dummy_user", 0), ) as group_text: assert await test_bot.is_logged_in(include_probe = False) is True group_text.assert_awaited() assert group_text.await_count >= 1 @pytest.mark.asyncio async def test_is_logged_in_logs_matched_raw_selector( self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture ) -> None: """Login detection logs should show the matched raw selector.""" caplog.set_level("DEBUG") with ( caplog.at_level("DEBUG"), patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, return_value = ("angemeldet als: dummy_user", 0), ), ): assert await test_bot.is_logged_in(include_probe = False) is True assert "Login detected via login detection selector" in caplog.text assert "CLASS_NAME=mr-medium" in caplog.text @pytest.mark.asyncio async def test_is_logged_in_logs_generic_message_when_selector_group_does_not_match( self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture ) -> None: """Missing selector-group match should log the tried selectors when probe is disabled.""" caplog.set_level("DEBUG") with ( caplog.at_level("DEBUG"), patch.object(test_bot, "web_text_first_available", side_effect = [TimeoutError(), TimeoutError()]), patch.object(test_bot, "_has_logged_out_cta", new_callable = AsyncMock, return_value = False), ): assert await test_bot.is_logged_in(include_probe = False) is False assert "No login detected via configured login detection selectors" in caplog.text assert "CLASS_NAME=mr-medium" in caplog.text assert "ID=user-email" in caplog.text @pytest.mark.asyncio async def test_is_logged_in_logs_raw_selectors_when_dom_checks_fail_and_probe_disabled( self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture ) -> None: """Final failure should report selectors and disabled-probe state.""" caplog.set_level("DEBUG") with ( caplog.at_level("DEBUG"), patch.object(test_bot, "web_text_first_available", side_effect = [TimeoutError(), TimeoutError()]), patch.object(test_bot, "_has_logged_out_cta", new_callable = AsyncMock, return_value = False), ): assert await test_bot.is_logged_in() is False assert "No login detected via configured login detection selectors" in caplog.text assert "auth probe is disabled" in caplog.text @pytest.mark.asyncio async def test_get_login_state_prefers_dom_checks(self, test_bot:KleinanzeigenBot) -> None: with ( patch.object( test_bot, "web_text_first_available", new_callable = AsyncMock, return_value = ("Welcome dummy_user", 0), ) as web_text, ): assert await test_bot.get_login_state() == LoginState.LOGGED_IN web_text.assert_awaited_once() def test_current_page_url_strips_query_and_fragment(self, test_bot:KleinanzeigenBot) -> None: page = MagicMock() page.url = "https://login.kleinanzeigen.de/u/login/password?state=secret&code=abc#frag" test_bot.page = page assert test_bot._current_page_url() == "https://login.kleinanzeigen.de/u/login/password" def test_is_valid_post_auth0_destination_filters_invalid_urls(self, test_bot:KleinanzeigenBot) -> None: assert test_bot._is_valid_post_auth0_destination("https://www.kleinanzeigen.de/") is True assert test_bot._is_valid_post_auth0_destination("https://www.kleinanzeigen.de/m-meine-anzeigen.html") is True assert test_bot._is_valid_post_auth0_destination("https://foo.kleinanzeigen.de/") is True assert test_bot._is_valid_post_auth0_destination("unknown") is False assert test_bot._is_valid_post_auth0_destination("about:blank") is False assert test_bot._is_valid_post_auth0_destination("https://evilkleinanzeigen.de/") is False assert test_bot._is_valid_post_auth0_destination("https://kleinanzeigen.de.evil.com/") is False assert test_bot._is_valid_post_auth0_destination("https://login.kleinanzeigen.de/u/login/password") is False assert test_bot._is_valid_post_auth0_destination("https://www.kleinanzeigen.de/login-error-500") is False @pytest.mark.asyncio async def test_get_login_state_returns_unknown_when_dom_checks_are_inconclusive(self, test_bot:KleinanzeigenBot) -> None: with ( patch.object(test_bot, "web_text_first_available", side_effect = [TimeoutError(), TimeoutError()]) as web_text, patch.object(test_bot, "web_find_first_available", side_effect = TimeoutError()) as cta_find, ): assert await test_bot.get_login_state() == LoginState.UNKNOWN assert web_text.await_count == 2 assert cta_find.await_count == 2 @pytest.mark.asyncio async def test_get_login_state_returns_logged_out_when_cta_detected(self, test_bot:KleinanzeigenBot) -> None: matched_element = MagicMock(spec = Element) with ( patch.object( test_bot, "web_text_first_available", side_effect = [TimeoutError(), TimeoutError()], ) as web_text, patch.object(test_bot, "web_find_first_available", new_callable = AsyncMock, return_value = (matched_element, 0)), patch.object(test_bot, "_extract_visible_text", new_callable = AsyncMock, return_value = "Hier einloggen"), ): assert await test_bot.get_login_state() == LoginState.LOGGED_OUT assert web_text.await_count == 2 @pytest.mark.asyncio async def test_get_login_state_unknown_captures_diagnostics_when_enabled(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: test_bot.config.diagnostics = DiagnosticsConfig.model_validate({"capture_on": {"login_detection": True}, "output_dir": str(tmp_path)}) page = MagicMock() page.save_screenshot = AsyncMock() page.get_content = AsyncMock(return_value = "") test_bot.page = page with ( patch.object(test_bot, "web_text_first_available", side_effect = [TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError()]), patch.object(test_bot, "web_find_first_available", side_effect = TimeoutError()), ): assert await test_bot.get_login_state() == LoginState.UNKNOWN page.save_screenshot.assert_awaited_once() page.get_content.assert_awaited_once() @pytest.mark.asyncio async def test_get_login_state_unknown_does_not_capture_diagnostics_when_disabled(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: test_bot.config.diagnostics = DiagnosticsConfig.model_validate({"capture_on": {"login_detection": False}, "output_dir": str(tmp_path)}) page = MagicMock() page.save_screenshot = AsyncMock() page.get_content = AsyncMock(return_value = "") test_bot.page = page with ( patch.object(test_bot, "web_text_first_available", side_effect = [TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError()]), patch.object(test_bot, "web_find_first_available", side_effect = TimeoutError()), ): assert await test_bot.get_login_state() == LoginState.UNKNOWN page.save_screenshot.assert_not_called() page.get_content.assert_not_called() @pytest.mark.asyncio async def test_get_login_state_unknown_pauses_for_inspection_when_enabled_and_interactive(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: test_bot.config.diagnostics = DiagnosticsConfig.model_validate( {"capture_on": {"login_detection": True}, "pause_on_login_detection_failure": True, "output_dir": str(tmp_path)} ) page = MagicMock() page.save_screenshot = AsyncMock() page.get_content = AsyncMock(return_value = "") test_bot.page = page stdin_mock = MagicMock() stdin_mock.isatty.return_value = True with ( patch.object( test_bot, "web_text_first_available", side_effect = [ TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError(), ], ), patch.object(test_bot, "web_find_first_available", side_effect = TimeoutError()), patch("kleinanzeigen_bot.sys.stdin", stdin_mock), patch("kleinanzeigen_bot.ainput", new_callable = AsyncMock) as mock_ainput, ): assert await test_bot.get_login_state() == LoginState.UNKNOWN # Call twice to ensure the capture/pause guard triggers only once per process. assert await test_bot.get_login_state() == LoginState.UNKNOWN page.save_screenshot.assert_awaited_once() page.get_content.assert_awaited_once() mock_ainput.assert_awaited_once() @pytest.mark.asyncio async def test_get_login_state_unknown_does_not_pause_when_non_interactive(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: test_bot.config.diagnostics = DiagnosticsConfig.model_validate( {"capture_on": {"login_detection": True}, "pause_on_login_detection_failure": True, "output_dir": str(tmp_path)} ) page = MagicMock() page.save_screenshot = AsyncMock() page.get_content = AsyncMock(return_value = "") test_bot.page = page stdin_mock = MagicMock() stdin_mock.isatty.return_value = False with ( patch.object(test_bot, "web_text_first_available", side_effect = [TimeoutError(), TimeoutError(), TimeoutError(), TimeoutError()]), patch.object(test_bot, "web_find_first_available", side_effect = TimeoutError()), patch("kleinanzeigen_bot.sys.stdin", stdin_mock), patch("kleinanzeigen_bot.ainput", new_callable = AsyncMock) as mock_ainput, ): assert await test_bot.get_login_state() == LoginState.UNKNOWN mock_ainput.assert_not_called() @pytest.mark.asyncio async def test_login_flow_completes_successfully(self, test_bot:KleinanzeigenBot) -> None: """Verify that normal login flow completes successfully.""" with ( patch.object(test_bot, "web_open") as mock_open, patch.object(test_bot, "get_login_state", new_callable = AsyncMock, side_effect = [LoginState.LOGGED_OUT, LoginState.LOGGED_IN]) as mock_logged_in, patch.object(test_bot, "_click_gdpr_banner", new_callable = AsyncMock), patch.object(test_bot, "fill_login_data_and_send", new_callable = AsyncMock) as mock_fill, patch.object(test_bot, "handle_after_login_logic", new_callable = AsyncMock) as mock_after_login, patch.object(test_bot, "_dismiss_consent_banner", new_callable = AsyncMock), ): await test_bot.login() opened_urls = [call.args[0] for call in mock_open.call_args_list] assert any(url.startswith(test_bot.root_url) for url in opened_urls) assert any(url.endswith("/m-einloggen-sso.html") for url in opened_urls) mock_logged_in.assert_awaited() mock_fill.assert_awaited_once() mock_after_login.assert_awaited_once() @pytest.mark.asyncio async def test_login_flow_returns_early_when_already_logged_in(self, test_bot:KleinanzeigenBot) -> None: """Login should return early when state is already LOGGED_IN.""" with ( patch.object(test_bot, "web_open") as mock_open, patch.object(test_bot, "get_login_state", new_callable = AsyncMock, return_value = LoginState.LOGGED_IN) as mock_state, patch.object(test_bot, "_click_gdpr_banner", new_callable = AsyncMock), patch.object(test_bot, "fill_login_data_and_send", new_callable = AsyncMock) as mock_fill, patch.object(test_bot, "handle_after_login_logic", new_callable = AsyncMock) as mock_after_login, ): await test_bot.login() mock_open.assert_awaited_once() assert mock_open.await_args is not None assert mock_open.await_args.args[0] == test_bot.root_url mock_state.assert_awaited_once() mock_fill.assert_not_called() mock_after_login.assert_not_called() @pytest.mark.asyncio async def test_login_flow_raises_when_state_remains_unknown(self, test_bot:KleinanzeigenBot) -> None: """Post-login UNKNOWN state should fail fast with diagnostics.""" with ( patch.object(test_bot, "web_open"), patch.object(test_bot, "get_login_state", new_callable = AsyncMock, side_effect = [LoginState.LOGGED_OUT, LoginState.UNKNOWN]) as mock_state, patch.object(test_bot, "_click_gdpr_banner", new_callable = AsyncMock), patch.object(test_bot, "fill_login_data_and_send", new_callable = AsyncMock), patch.object(test_bot, "handle_after_login_logic", new_callable = AsyncMock), patch.object(test_bot, "_dismiss_consent_banner", new_callable = AsyncMock), patch.object(test_bot, "_capture_login_detection_diagnostics_if_enabled", new_callable = AsyncMock) as mock_diagnostics, ): with pytest.raises(AssertionError, match = "Login could not be confirmed"): await test_bot.login() mock_diagnostics.assert_awaited_once() mock_state.assert_awaited() @pytest.mark.asyncio async def test_login_flow_raises_when_sso_navigation_times_out(self, test_bot:KleinanzeigenBot) -> None: """SSO navigation timeout should trigger diagnostics and re-raise.""" with ( patch.object(test_bot, "web_open", new_callable = AsyncMock, side_effect = [None, TimeoutError("sso timeout")]), patch.object(test_bot, "get_login_state", new_callable = AsyncMock, return_value = LoginState.LOGGED_OUT) as mock_state, patch.object(test_bot, "_click_gdpr_banner", new_callable = AsyncMock), patch.object(test_bot, "_capture_login_detection_diagnostics_if_enabled", new_callable = AsyncMock) as mock_diagnostics, ): with pytest.raises(TimeoutError, match = "sso timeout"): await test_bot.login() mock_diagnostics.assert_awaited_once() mock_state.assert_awaited_once() @pytest.mark.asyncio async def test_check_and_wait_for_captcha(self, test_bot:KleinanzeigenBot) -> None: """Verify that captcha detection works correctly.""" with patch.object(test_bot, "web_find") as mock_find, patch("kleinanzeigen_bot.ainput", new_callable = AsyncMock) as mock_ainput: # Test case 1: Captcha found mock_find.return_value = AsyncMock() mock_ainput.return_value = "" await test_bot.check_and_wait_for_captcha(is_login_page = True) assert mock_find.call_count == 1 assert mock_ainput.call_count == 1 # Test case 2: No captcha mock_find.side_effect = TimeoutError() mock_ainput.reset_mock() await test_bot.check_and_wait_for_captcha(is_login_page = True) assert mock_find.call_count == 2 assert mock_ainput.call_count == 0 @pytest.mark.asyncio async def test_fill_login_data_and_send(self, test_bot:KleinanzeigenBot) -> None: """Verify that login form filling works correctly.""" with ( patch.object(test_bot, "_wait_for_auth0_login_context", new_callable = AsyncMock) as wait_context, patch.object(test_bot, "_wait_for_auth0_password_step", new_callable = AsyncMock) as wait_password, patch.object(test_bot, "_wait_for_post_auth0_submit_transition", new_callable = AsyncMock) as wait_transition, patch.object(test_bot, "web_input") as mock_input, patch.object(test_bot, "web_click") as mock_click, patch.object(test_bot, "check_and_wait_for_captcha", new_callable = AsyncMock) as mock_captcha, ): await test_bot.fill_login_data_and_send() wait_context.assert_awaited_once() wait_password.assert_awaited_once() wait_transition.assert_awaited_once() assert mock_captcha.call_count == 1 assert mock_input.call_count == 2 assert mock_click.call_count == 2 @pytest.mark.asyncio async def test_fill_login_data_and_send_logs_generic_start_message( self, test_bot:KleinanzeigenBot, caplog:pytest.LogCaptureFixture ) -> None: with ( caplog.at_level("INFO"), patch.object(test_bot, "_wait_for_auth0_login_context", new_callable = AsyncMock), patch.object(test_bot, "_wait_for_auth0_password_step", new_callable = AsyncMock), patch.object(test_bot, "_wait_for_post_auth0_submit_transition", new_callable = AsyncMock), patch.object(test_bot, "web_input"), patch.object(test_bot, "web_click"), patch.object(test_bot, "check_and_wait_for_captcha", new_callable = AsyncMock), ): await test_bot.fill_login_data_and_send() assert "Logging in..." in caplog.text assert test_bot.config.login.username not in caplog.text @pytest.mark.asyncio async def test_fill_login_data_and_send_fails_when_password_step_missing(self, test_bot:KleinanzeigenBot) -> None: """Missing Auth0 password step should fail fast.""" with ( patch.object(test_bot, "_wait_for_auth0_login_context", new_callable = AsyncMock), patch.object(test_bot, "_wait_for_auth0_password_step", new_callable = AsyncMock, side_effect = AssertionError("missing password")), patch.object(test_bot, "web_input") as mock_input, patch.object(test_bot, "web_click") as mock_click, ): with pytest.raises(AssertionError, match = "missing password"): await test_bot.fill_login_data_and_send() assert mock_input.call_count == 1 assert mock_click.call_count == 1 @pytest.mark.asyncio async def test_wait_for_post_auth0_submit_transition_url_branch(self, test_bot:KleinanzeigenBot) -> None: """URL transition success should return without fallback checks.""" with ( patch.object(test_bot, "web_await", new_callable = AsyncMock, return_value = True) as mock_wait, patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as mock_sleep, ): await test_bot._wait_for_post_auth0_submit_transition() mock_wait.assert_awaited_once() mock_sleep.assert_not_called() @pytest.mark.asyncio async def test_wait_for_post_auth0_submit_transition_dom_fallback_branch(self, test_bot:KleinanzeigenBot) -> None: """DOM fallback should run when URL transition is inconclusive.""" with ( patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]) as mock_wait, patch.object(test_bot, "is_logged_in", new_callable = AsyncMock, return_value = True) as mock_is_logged_in, patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as mock_sleep, ): await test_bot._wait_for_post_auth0_submit_transition() mock_wait.assert_awaited_once() mock_is_logged_in.assert_awaited_once() mock_sleep.assert_not_called() @pytest.mark.asyncio async def test_wait_for_post_auth0_submit_transition_sleep_fallback_branch(self, test_bot:KleinanzeigenBot) -> None: """Sleep fallback should run when bounded login check times out.""" with ( patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]) as mock_wait, patch.object(test_bot, "is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError) as mock_is_logged_in, patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as mock_sleep, ): with pytest.raises(TimeoutError, match = "Auth0 post-submit verification remained inconclusive"): await test_bot._wait_for_post_auth0_submit_transition() mock_wait.assert_awaited_once() assert mock_is_logged_in.await_count == 2 mock_sleep.assert_awaited_once() assert mock_sleep.await_args is not None sleep_kwargs = cast(Any, mock_sleep.await_args).kwargs assert sleep_kwargs["min_ms"] < sleep_kwargs["max_ms"] @pytest.mark.asyncio async def test_wait_for_post_auth0_submit_transition_sleep_fallback_when_login_not_confirmed( self, test_bot:KleinanzeigenBot ) -> None: """Sleep fallback should run when bounded login check returns False.""" with ( patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]) as mock_wait, patch.object(test_bot, "is_logged_in", new_callable = AsyncMock, return_value = False) as mock_is_logged_in, patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as mock_sleep, ): with pytest.raises(TimeoutError, match = "Auth0 post-submit verification remained inconclusive"): await test_bot._wait_for_post_auth0_submit_transition() mock_wait.assert_awaited_once() assert mock_is_logged_in.await_count == 2 mock_sleep.assert_awaited_once() @pytest.mark.asyncio async def test_click_gdpr_banner_uses_quick_dom_timeout_and_passes_click_timeout(self, test_bot:KleinanzeigenBot) -> None: with ( patch.object(test_bot, "_timeout", return_value = 1.25) as mock_timeout, patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, patch.object(test_bot, "web_click", new_callable = AsyncMock) as mock_click, ): await test_bot._click_gdpr_banner() mock_timeout.assert_called_once_with("quick_dom") mock_find.assert_awaited_once_with(By.ID, "gdpr-banner-accept", timeout = 1.25) mock_click.assert_awaited_once_with(By.ID, "gdpr-banner-accept", timeout = 1.25) @pytest.mark.asyncio async def test_handle_after_login_logic(self, test_bot:KleinanzeigenBot) -> None: """Verify that post-login handling works correctly.""" with ( patch.object(test_bot, "_check_sms_verification", new_callable = AsyncMock, side_effect = TimeoutError()) as mock_sms, patch.object(test_bot, "_check_email_verification", new_callable = AsyncMock, side_effect = TimeoutError()) as mock_email, patch.object(test_bot, "_click_gdpr_banner", new_callable = AsyncMock, side_effect = TimeoutError()) as mock_gdpr, ): await test_bot.handle_after_login_logic() mock_sms.assert_awaited_once() mock_email.assert_awaited_once() mock_gdpr.assert_awaited_once() class TestKleinanzeigenBotDiagnostics: @pytest.fixture def diagnostics_ad_config(self) -> dict[str, Any]: return { "active": True, "type": "OFFER", "title": "Test ad title", "description": "Test description", "category": "161/176/sonstige", "price_type": "NEGOTIABLE", "shipping_type": "PICKUP", "sell_directly": False, "contact": { "name": "Tester", "zipcode": "12345", }, "republication_interval": 7, } @pytest.mark.unit @pytest.mark.asyncio async def test_publish_ads_captures_diagnostics_on_failures( self, test_bot:KleinanzeigenBot, tmp_path:Path, diagnostics_ad_config:dict[str, Any], ) -> None: """Ensure publish failures capture diagnostics artifacts.""" log_file_path = tmp_path / "test.log" log_file_path.write_text("Test log content\n", encoding = "utf-8") test_bot.log_file_path = str(log_file_path) test_bot.config.diagnostics = DiagnosticsConfig.model_validate({"capture_on": {"publish": True}, "output_dir": str(tmp_path)}) page = MagicMock() page.save_screenshot = AsyncMock() page.get_content = AsyncMock(return_value = "") page.sleep = AsyncMock() page.url = "https://example.com/fail" test_bot.page = page ad_cfg = Ad.model_validate(diagnostics_ad_config) ad_cfg_orig = copy.deepcopy(diagnostics_ad_config) ad_file = str(tmp_path / "ad_000001_Test.yml") ads_response = {"content": json.dumps({"ads": [], "paging": {"pageNum": 1, "last": 1}})} with ( patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = ads_response), patch.object(test_bot, "publish_ad", new_callable = AsyncMock, side_effect = TimeoutError("boom")), ): await test_bot.publish_ads([(ad_file, ad_cfg, ad_cfg_orig)]) expected_retries = PUBLISH_MAX_RETRIES assert page.save_screenshot.await_count == expected_retries assert page.get_content.await_count == expected_retries entries = os.listdir(tmp_path) html_files = [name for name in entries if fnmatch.fnmatch(name, "publish_error_*_attempt*_ad_000001_Test.html")] json_files = [name for name in entries if fnmatch.fnmatch(name, "publish_error_*_attempt*_ad_000001_Test.json")] assert len(html_files) == expected_retries assert len(json_files) == expected_retries @pytest.mark.unit @pytest.mark.asyncio async def test_publish_ads_captures_log_copy_when_enabled( self, test_bot:KleinanzeigenBot, tmp_path:Path, diagnostics_ad_config:dict[str, Any], ) -> None: """Ensure publish failures copy log file when capture_log_copy is enabled.""" log_file_path = tmp_path / "test.log" log_file_path.write_text("Test log content\n", encoding = "utf-8") test_bot.log_file_path = str(log_file_path) test_bot.config.diagnostics = DiagnosticsConfig.model_validate({"capture_on": {"publish": True}, "capture_log_copy": True, "output_dir": str(tmp_path)}) page = MagicMock() page.save_screenshot = AsyncMock() page.get_content = AsyncMock(return_value = "") page.sleep = AsyncMock() page.url = "https://example.com/fail" test_bot.page = page ad_cfg = Ad.model_validate(diagnostics_ad_config) ad_cfg_orig = copy.deepcopy(diagnostics_ad_config) ad_file = str(tmp_path / "ad_000001_Test.yml") ads_response = {"content": json.dumps({"ads": [], "paging": {"pageNum": 1, "last": 1}})} with ( patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = ads_response), patch.object(test_bot, "publish_ad", new_callable = AsyncMock, side_effect = TimeoutError("boom")), ): await test_bot.publish_ads([(ad_file, ad_cfg, ad_cfg_orig)]) entries = os.listdir(tmp_path) log_files = [name for name in entries if fnmatch.fnmatch(name, "publish_error_*_attempt*_ad_000001_Test.log")] assert len(log_files) == PUBLISH_MAX_RETRIES @pytest.mark.unit @pytest.mark.asyncio async def test_publish_ads_does_not_capture_diagnostics_when_disabled( self, test_bot:KleinanzeigenBot, tmp_path:Path, diagnostics_ad_config:dict[str, Any], ) -> None: """Ensure diagnostics are not captured when disabled.""" test_bot.config.diagnostics = DiagnosticsConfig.model_validate({"capture_on": {"publish": False}, "output_dir": str(tmp_path)}) page = MagicMock() page.save_screenshot = AsyncMock() page.get_content = AsyncMock(return_value = "") page.sleep = AsyncMock() page.url = "https://example.com/fail" test_bot.page = page ad_cfg = Ad.model_validate(diagnostics_ad_config) ad_cfg_orig = copy.deepcopy(diagnostics_ad_config) ad_file = str(tmp_path / "ad_000001_Test.yml") with ( patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = {"content": json.dumps({"ads": []})}), patch.object(test_bot, "publish_ad", new_callable = AsyncMock, side_effect = TimeoutError("boom")), ): await test_bot.publish_ads([(ad_file, ad_cfg, ad_cfg_orig)]) page.save_screenshot.assert_not_called() page.get_content.assert_not_called() entries = os.listdir(tmp_path) html_files = [name for name in entries if fnmatch.fnmatch(name, "publish_error_*_attempt*_ad_000001_Test.html")] json_files = [name for name in entries if fnmatch.fnmatch(name, "publish_error_*_attempt*_ad_000001_Test.json")] assert not html_files assert not json_files class TestKleinanzeigenBotLocalization: """Tests for localization and help text.""" def test_show_help_displays_german_text(self, test_bot:KleinanzeigenBot) -> None: """Verify that help text is displayed in German when language is German.""" with patch("kleinanzeigen_bot.get_current_locale") as mock_locale, patch("builtins.print") as mock_print: mock_locale.return_value.language = "de" test_bot.show_help() printed_text = "".join(str(call.args[0]) for call in mock_print.call_args_list) assert "Verwendung:" in printed_text assert "Befehle:" in printed_text def test_show_help_displays_english_text(self, test_bot:KleinanzeigenBot) -> None: """Verify that help text is displayed in English when language is English.""" with patch("kleinanzeigen_bot.get_current_locale") as mock_locale, patch("builtins.print") as mock_print: mock_locale.return_value.language = "en" test_bot.show_help() printed_text = "".join(str(call.args[0]) for call in mock_print.call_args_list) assert "Usage:" in printed_text assert "Commands:" in printed_text class TestKleinanzeigenBotBasics: """Basic tests for KleinanzeigenBot.""" def test_get_version(self, test_bot:KleinanzeigenBot) -> None: """Test version retrieval.""" assert test_bot.get_version() == __version__ @pytest.mark.asyncio async def test_publish_ads_triggers_publish_and_cleanup( self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], mock_page:MagicMock, ) -> None: """Simulate publish job wiring without hitting the live site.""" test_bot.page = mock_page test_bot.config.publishing.delete_old_ads = "AFTER_PUBLISH" test_bot.keep_old_ads = False payload:dict[str, list[Any]] = {"ads": []} ad_cfgs:list[tuple[str, Ad, dict[str, Any]]] = [("ad.yaml", Ad.model_validate(base_ad_config), {})] with ( patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = {"content": json.dumps(payload)}) as web_request_mock, patch.object(test_bot, "publish_ad", new_callable = AsyncMock) as publish_ad_mock, patch.object(test_bot, "web_await", new_callable = AsyncMock, return_value = True) as web_await_mock, patch.object(test_bot, "delete_ad", new_callable = AsyncMock) as delete_ad_mock, ): await test_bot.publish_ads(ad_cfgs) # web_request is called once for initial published-ads snapshot expected_url = f"{test_bot.root_url}/m-meine-anzeigen-verwalten.json?sort=DEFAULT&pageNum=1" web_request_mock.assert_awaited_once_with(expected_url) publish_ad_mock.assert_awaited_once_with("ad.yaml", ad_cfgs[0][1], {}, [], AdUpdateStrategy.REPLACE) web_await_mock.assert_awaited_once() delete_ad_mock.assert_awaited_once_with(ad_cfgs[0][1], [], delete_old_ads_by_title = False) @pytest.mark.asyncio async def test_publish_ads_uses_millisecond_retry_delay_on_retryable_failure( self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], mock_page:MagicMock, ) -> None: """Retry branch should sleep with explicit millisecond delay.""" test_bot.page = mock_page test_bot.keep_old_ads = True ad_cfg = Ad.model_validate(base_ad_config) ad_cfg_orig = copy.deepcopy(base_ad_config) ad_file = "ad.yaml" ads_response = {"content": json.dumps({"ads": [], "paging": {"pageNum": 1, "last": 1}})} with ( patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = ads_response), patch.object(test_bot, "publish_ad", new_callable = AsyncMock, side_effect = [TimeoutError("transient"), None]) as publish_mock, patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as sleep_mock, patch.object(test_bot, "web_await", new_callable = AsyncMock, return_value = True), ): await test_bot.publish_ads([(ad_file, ad_cfg, ad_cfg_orig)]) assert publish_mock.await_count == 2 sleep_mock.assert_awaited_once_with(2_000) @pytest.mark.asyncio async def test_publish_ads_does_not_retry_when_submission_state_is_uncertain( self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], mock_page:MagicMock, ) -> None: """Post-submit uncertainty must fail closed and skip retries.""" test_bot.page = mock_page test_bot.keep_old_ads = True ad_cfg = Ad.model_validate(base_ad_config) ad_cfg_orig = copy.deepcopy(base_ad_config) ad_file = "ad.yaml" with ( patch.object( test_bot, "web_request", new_callable = AsyncMock, return_value = {"content": json.dumps({"ads": [], "paging": {"pageNum": 1, "last": 1}})}, ), patch.object( test_bot, "publish_ad", new_callable = AsyncMock, side_effect = PublishSubmissionUncertainError("submission may have succeeded before failure"), ) as publish_mock, patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as sleep_mock, ): await test_bot.publish_ads([(ad_file, ad_cfg, ad_cfg_orig)]) assert publish_mock.await_count == 1 sleep_mock.assert_not_awaited() @pytest.mark.asyncio async def test_publish_ad_keeps_pre_submit_timeouts_retryable( self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], ) -> None: """Timeouts before submit boundary should remain plain retryable failures.""" ad_cfg = Ad.model_validate(base_ad_config | {"id": 12345, "shipping_type": "NOT_APPLICABLE", "price_type": "NOT_APPLICABLE"}) ad_cfg_orig = copy.deepcopy(base_ad_config) with ( patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "_dismiss_consent_banner", new_callable = AsyncMock), patch.object(test_bot, "_KleinanzeigenBot__set_category", new_callable = AsyncMock, side_effect = TimeoutError("image upload timeout")), pytest.raises(TimeoutError, match = "image upload timeout"), ): await test_bot.publish_ad("ad.yaml", ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.MODIFY) @pytest.mark.asyncio async def test_publish_ad_marks_post_submit_timeout_as_uncertain( self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], mock_page:MagicMock, ) -> None: """Timeouts after submit click should be converted to non-retryable uncertainty.""" test_bot.page = mock_page ad_cfg = Ad.model_validate(base_ad_config | {"id": 12345, "shipping_type": "NOT_APPLICABLE", "price_type": "NOT_APPLICABLE"}) ad_cfg_orig = copy.deepcopy(base_ad_config) async def find_side_effect(selector_type:By, selector_value:str, **_:Any) -> MagicMock: if selector_type == By.ID and selector_value == "myftr-shppngcrt-frm": raise TimeoutError("no payment form") return MagicMock() with ( patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "_dismiss_consent_banner", new_callable = AsyncMock), patch.object(test_bot, "_KleinanzeigenBot__set_category", new_callable = AsyncMock), patch.object(test_bot, "_KleinanzeigenBot__set_special_attributes", new_callable = AsyncMock), patch.object(test_bot, "_KleinanzeigenBot__set_contact_fields", new_callable = AsyncMock), patch.object(test_bot, "check_and_wait_for_captcha", new_callable = AsyncMock), patch.object(test_bot, "web_input", new_callable = AsyncMock), patch.object(test_bot, "web_click", new_callable = AsyncMock), patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = False), patch.object(test_bot, "web_execute", new_callable = AsyncMock), patch.object(test_bot, "web_find", new_callable = AsyncMock, side_effect = find_side_effect), patch.object(test_bot, "web_find_all", new_callable = AsyncMock, return_value = []), patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = TimeoutError("confirmation timeout")), pytest.raises(PublishSubmissionUncertainError, match = "submission may have succeeded before failure"), ): await test_bot.publish_ad("ad.yaml", ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.MODIFY) @pytest.mark.asyncio async def test_publish_ad_marks_post_submit_protocol_exception_as_uncertain( self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], mock_page:MagicMock, ) -> None: """Protocol exceptions after submit click should be converted to uncertainty.""" test_bot.page = mock_page ad_cfg = Ad.model_validate(base_ad_config | {"id": 12345, "shipping_type": "NOT_APPLICABLE", "price_type": "NOT_APPLICABLE"}) ad_cfg_orig = copy.deepcopy(base_ad_config) async def find_side_effect(selector_type:By, selector_value:str, **_:Any) -> MagicMock: if selector_type == By.ID and selector_value == "myftr-shppngcrt-frm": raise TimeoutError("no payment form") return MagicMock() with ( patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "_dismiss_consent_banner", new_callable = AsyncMock), patch.object(test_bot, "_KleinanzeigenBot__set_category", new_callable = AsyncMock), patch.object(test_bot, "_KleinanzeigenBot__set_special_attributes", new_callable = AsyncMock), patch.object(test_bot, "_KleinanzeigenBot__set_contact_fields", new_callable = AsyncMock), patch.object(test_bot, "check_and_wait_for_captcha", new_callable = AsyncMock), patch.object(test_bot, "web_input", new_callable = AsyncMock), patch.object(test_bot, "web_click", new_callable = AsyncMock), patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = False), patch.object(test_bot, "web_execute", new_callable = AsyncMock), patch.object(test_bot, "web_find", new_callable = AsyncMock, side_effect = find_side_effect), patch.object(test_bot, "web_find_all", new_callable = AsyncMock, return_value = []), patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = ProtocolException(MagicMock(), "connection lost", 0)), pytest.raises(PublishSubmissionUncertainError, match = "submission may have succeeded before failure"), ): await test_bot.publish_ad("ad.yaml", ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.MODIFY) def test_get_root_url(self, test_bot:KleinanzeigenBot) -> None: """Test root URL retrieval.""" assert test_bot.root_url == "https://www.kleinanzeigen.de" def test_get_config_defaults(self, test_bot:KleinanzeigenBot) -> None: """Test default configuration values.""" assert isinstance(test_bot.config, Config) assert test_bot.command == "help" assert test_bot.ads_selector == "due" assert test_bot.keep_old_ads is False def test_get_log_level(self, test_bot:KleinanzeigenBot) -> None: """Test log level configuration.""" # Reset log level to default LOG.setLevel(loggers.INFO) assert not loggers.is_debug(LOG) test_bot.parse_args(["script.py", "-v"]) assert loggers.is_debug(LOG) def test_get_config_file_path(self, test_bot:KleinanzeigenBot) -> None: """Test config file path handling.""" default_path = os.path.abspath("config.yaml") assert test_bot.config_file_path == default_path test_path = os.path.abspath("custom_config.yaml") test_bot.config_file_path = test_path assert test_bot.config_file_path == test_path def test_get_log_file_path(self, test_bot:KleinanzeigenBot) -> None: """Test log file path handling.""" default_path = os.path.abspath("kleinanzeigen_bot.log") assert test_bot.log_file_path == default_path test_path = os.path.abspath("custom.log") test_bot.log_file_path = test_path assert test_bot.log_file_path == test_path def test_get_categories(self, test_bot:KleinanzeigenBot) -> None: """Test categories handling.""" test_categories = {"test_cat": "test_id"} test_bot.categories = test_categories assert test_bot.categories == test_categories class TestKleinanzeigenBotArgParsing: """Tests for command line argument parsing.""" def test_parse_args_help(self, test_bot:KleinanzeigenBot) -> None: """Test parsing help command.""" test_bot.parse_args(["script.py", "help"]) assert test_bot.command == "help" def test_parse_args_version(self, test_bot:KleinanzeigenBot) -> None: """Test parsing version command.""" test_bot.parse_args(["script.py", "version"]) assert test_bot.command == "version" def test_parse_args_verbose(self, test_bot:KleinanzeigenBot) -> None: """Test parsing verbose flag.""" test_bot.parse_args(["script.py", "-v", "help"]) assert loggers.is_debug(loggers.get_logger("kleinanzeigen_bot")) def test_parse_args_config_path(self, test_bot:KleinanzeigenBot) -> None: """Test parsing config path.""" test_bot.parse_args(["script.py", "--config=test.yaml", "help"]) assert test_bot.config_file_path.endswith("test.yaml") def test_parse_args_logfile(self, test_bot:KleinanzeigenBot) -> None: """Test parsing log file path.""" test_bot.parse_args(["script.py", "--logfile=test.log", "help"]) assert test_bot.log_file_path is not None assert "test.log" in test_bot.log_file_path def test_parse_args_workspace_mode(self, test_bot:KleinanzeigenBot) -> None: """Test parsing workspace mode option.""" test_bot.parse_args(["script.py", "--workspace-mode=xdg", "help"]) assert test_bot._workspace_mode_arg == "xdg" def test_parse_args_workspace_mode_invalid(self, test_bot:KleinanzeigenBot) -> None: """Test invalid workspace mode exits with error.""" with pytest.raises(SystemExit) as exc_info: test_bot.parse_args(["script.py", "--workspace-mode=invalid", "help"]) assert exc_info.value.code == 2 def test_parse_args_ads_selector(self, test_bot:KleinanzeigenBot) -> None: """Test parsing ads selector.""" test_bot.parse_args(["script.py", "--ads=all", "publish"]) assert test_bot.ads_selector == "all" def test_parse_args_force(self, test_bot:KleinanzeigenBot) -> None: """Test parsing force flag.""" test_bot.parse_args(["script.py", "--force", "publish"]) assert test_bot.ads_selector == "all" def test_parse_args_keep_old(self, test_bot:KleinanzeigenBot) -> None: """Test parsing keep-old flag.""" test_bot.parse_args(["script.py", "--keep-old", "publish"]) assert test_bot.keep_old_ads is True def test_parse_args_logfile_empty(self, test_bot:KleinanzeigenBot) -> None: """Test parsing empty log file path.""" test_bot.parse_args(["script.py", "--logfile=", "help"]) assert test_bot.log_file_path is None def test_parse_args_lang_option(self, test_bot:KleinanzeigenBot) -> None: """Test parsing language option.""" test_bot.parse_args(["script.py", "--lang=en", "help"]) assert test_bot.command == "help" def test_parse_args_no_arguments(self, test_bot:KleinanzeigenBot) -> None: """Test parsing no arguments defaults to help.""" test_bot.parse_args(["script.py"]) assert test_bot.command == "help" def test_parse_args_multiple_commands(self, test_bot:KleinanzeigenBot) -> None: """Test parsing multiple commands raises error.""" with pytest.raises(SystemExit) as exc_info: test_bot.parse_args(["script.py", "help", "version"]) assert exc_info.value.code == 2 def test_parse_args_explicit_flags(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Test that explicit flags are set when config/logfile/workspace options are provided.""" config_path = tmp_path / "custom_config.yaml" log_path = tmp_path / "custom.log" # Test --config flag stores raw config arg test_bot.parse_args(["script.py", "--config", str(config_path), "help"]) assert test_bot._config_arg == str(config_path) assert str(config_path.absolute()) == test_bot.config_file_path # Test --logfile flag sets explicit logfile values test_bot.parse_args(["script.py", "--logfile", str(log_path), "help"]) assert test_bot._logfile_explicitly_provided is True assert test_bot._logfile_arg == str(log_path) assert str(log_path.absolute()) == test_bot.log_file_path # Test both flags together test_bot._config_arg = None test_bot._logfile_explicitly_provided = False test_bot._workspace_mode_arg = None test_bot.parse_args(["script.py", "--config", str(config_path), "--logfile", str(log_path), "--workspace-mode", "portable", "help"]) assert test_bot._config_arg == str(config_path) assert test_bot._logfile_explicitly_provided is True assert test_bot._workspace_mode_arg == "portable" class TestKleinanzeigenBotCommands: """Tests for command execution.""" @pytest.mark.asyncio async def test_run_version_command(self, test_bot:KleinanzeigenBot, capsys:Any) -> None: """Test running version command.""" await test_bot.run(["script.py", "version"]) captured = capsys.readouterr() assert __version__ in captured.out @pytest.mark.asyncio async def test_run_help_command(self, test_bot:KleinanzeigenBot, capsys:Any) -> None: """Test running help command.""" await test_bot.run(["script.py", "help"]) captured = capsys.readouterr() assert "Usage:" in captured.out @pytest.mark.asyncio async def test_run_unknown_command(self, test_bot:KleinanzeigenBot) -> None: """Test running unknown command.""" with pytest.raises(SystemExit) as exc_info: await test_bot.run(["script.py", "unknown"]) assert exc_info.value.code == 2 @pytest.mark.asyncio async def test_verify_command(self, test_bot:KleinanzeigenBot, tmp_path:Any) -> None: """Test verify command with minimal config.""" config_path = Path(tmp_path) / "config.yaml" config_path.write_text( """ login: username: test password: test """, encoding = "utf-8", ) test_bot.config_file_path = str(config_path) await test_bot.run(["script.py", "verify"]) assert test_bot.config.login.username == "test" class TestKleinanzeigenBotAdOperations: """Tests for ad-related operations.""" @pytest.mark.asyncio async def test_run_delete_command_no_ads(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running delete command with no ads.""" with patch.object(test_bot, "load_ads", return_value = []): await test_bot.run(["script.py", "delete"]) assert test_bot.command == "delete" @pytest.mark.asyncio async def test_run_publish_command_no_ads(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running publish command with no ads.""" with patch.object(test_bot, "load_ads", return_value = []): await test_bot.run(["script.py", "publish"]) assert test_bot.command == "publish" @pytest.mark.asyncio async def test_run_download_command_default_selector(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running download command with default selector.""" with patch.object(test_bot, "download_ads", new_callable = AsyncMock): await test_bot.run(["script.py", "download"]) assert test_bot.ads_selector == "new" @pytest.mark.asyncio async def test_run_update_default_selector(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running update command with default selector falls back to changed.""" with patch.object(test_bot, "load_ads", return_value = []): await test_bot.run(["script.py", "update"]) assert test_bot.ads_selector == "changed" @pytest.mark.asyncio async def test_run_extend_default_selector(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running extend command with default selector falls back to all.""" with patch.object(test_bot, "load_ads", return_value = []): await test_bot.run(["script.py", "extend"]) assert test_bot.ads_selector == "all" def test_load_ads_no_files(self, test_bot:KleinanzeigenBot) -> None: """Test loading ads with no files.""" test_bot.config.ad_files = ["nonexistent/*.yaml"] ads = test_bot.load_ads() assert len(ads) == 0 class TestKleinanzeigenBotAdManagement: """Tests for ad management functionality.""" @pytest.mark.asyncio async def test_download_ads_with_specific_ids(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test downloading ads with specific IDs.""" test_bot.ads_selector = "123,456" with patch.object(test_bot, "download_ads", new_callable = AsyncMock): await test_bot.run(["script.py", "download", "--ads=123,456"]) assert test_bot.ads_selector == "123,456" @pytest.mark.asyncio async def test_run_publish_invalid_selector(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running publish with invalid selector exits with error.""" with pytest.raises(SystemExit) as exc_info: await test_bot.run(["script.py", "publish", "--ads=invalid"]) assert exc_info.value.code == 2 @pytest.mark.asyncio async def test_run_download_invalid_selector(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running download with invalid selector exits with error.""" with pytest.raises(SystemExit) as exc_info: await test_bot.run(["script.py", "download", "--ads=invalid"]) assert exc_info.value.code == 2 @pytest.mark.asyncio async def test_run_update_invalid_selector(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running update with invalid selector exits with error.""" with pytest.raises(SystemExit) as exc_info: await test_bot.run(["script.py", "update", "--ads=invalid"]) assert exc_info.value.code == 2 @pytest.mark.asyncio async def test_run_extend_invalid_selector(self, test_bot:KleinanzeigenBot, mock_config_setup:None) -> None: # pylint: disable=unused-argument """Test running extend with invalid selector exits with error.""" with pytest.raises(SystemExit) as exc_info: await test_bot.run(["script.py", "extend", "--ads=invalid"]) assert exc_info.value.code == 2 class TestKleinanzeigenBotAdConfiguration: """Tests for ad configuration functionality.""" def test_load_config_with_categories(self, test_bot:KleinanzeigenBot, tmp_path:Any) -> None: """Test loading config with custom categories.""" config_path = Path(tmp_path) / "config.yaml" with open(config_path, "w", encoding = "utf-8") as f: f.write(""" login: username: test password: test categories: custom_cat: custom_id """) test_bot.config_file_path = str(config_path) test_bot.load_config() assert "custom_cat" in test_bot.categories assert test_bot.categories["custom_cat"] == "custom_id" def test_load_ads_with_missing_title(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None: """Test loading ads with missing title.""" temp_path = Path(tmp_path) ad_dir = temp_path / "ads" ad_dir.mkdir() ad_file = ad_dir / "test_ad.yaml" # Create a minimal config with empty title to trigger validation ad_cfg = minimal_ad_config | {"title": ""} dicts.save_dict(ad_file, ad_cfg) # Set config file path to tmp_path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] with pytest.raises(ValidationError) as exc_info: test_bot.load_ads() assert "title" in str(exc_info.value) def test_load_ads_with_invalid_price_type(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None: """Test loading ads with invalid price type.""" temp_path = Path(tmp_path) ad_dir = temp_path / "ads" ad_dir.mkdir() ad_file = ad_dir / "test_ad.yaml" # Create config with invalid price type ad_cfg = minimal_ad_config | {"price_type": "INVALID_TYPE"} dicts.save_dict(ad_file, ad_cfg) # Set config file path to tmp_path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] with pytest.raises(ValidationError) as exc_info: test_bot.load_ads() assert "price_type" in str(exc_info.value) def test_load_ads_with_invalid_shipping_type(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None: """Test loading ads with invalid shipping type.""" temp_path = Path(tmp_path) ad_dir = temp_path / "ads" ad_dir.mkdir() ad_file = ad_dir / "test_ad.yaml" # Create config with invalid shipping type ad_cfg = minimal_ad_config | {"shipping_type": "INVALID_TYPE"} dicts.save_dict(ad_file, ad_cfg) # Set config file path to tmp_path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] with pytest.raises(ValidationError) as exc_info: test_bot.load_ads() assert "shipping_type" in str(exc_info.value) def test_load_ads_with_invalid_price_config(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None: """Test loading ads with invalid price configuration.""" temp_path = Path(tmp_path) ad_dir = temp_path / "ads" ad_dir.mkdir() ad_file = ad_dir / "test_ad.yaml" # Create config with price for GIVE_AWAY type ad_cfg = minimal_ad_config | { "price_type": "GIVE_AWAY", "price": 100, # Price should not be set for GIVE_AWAY } dicts.save_dict(ad_file, ad_cfg) # Set config file path to tmp_path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] with pytest.raises(ValidationError) as exc_info: test_bot.load_ads() assert "price" in str(exc_info.value) def test_load_ads_with_missing_price(self, test_bot:KleinanzeigenBot, tmp_path:Any, minimal_ad_config:dict[str, Any]) -> None: """Test loading ads with missing price for FIXED price type.""" temp_path = Path(tmp_path) ad_dir = temp_path / "ads" ad_dir.mkdir() ad_file = ad_dir / "test_ad.yaml" # Create config with FIXED price type but no price ad_cfg = minimal_ad_config | { "price_type": "FIXED", "price": None, # Missing required price for FIXED type } dicts.save_dict(ad_file, ad_cfg) # Set config file path to tmp_path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] with pytest.raises(ValidationError) as exc_info: test_bot.load_ads() assert "price is required when price_type is FIXED" in str(exc_info.value) class TestKleinanzeigenBotAdDeletion: """Tests for ad deletion functionality.""" @pytest.mark.asyncio async def test_delete_ad_by_title(self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any]) -> None: """Test deleting an ad by title.""" test_bot.page = MagicMock() test_bot.page.evaluate = AsyncMock(return_value = {"statusCode": 200, "statusMessage": "OK", "content": "{}"}) test_bot.page.sleep = AsyncMock() # Use minimal config since we only need title for deletion by title ad_cfg = Ad.model_validate( minimal_ad_config | { "title": "Test Title", "id": None, # Explicitly set id to None for title-based deletion } ) published_ads = [{"title": "Test Title", "id": "67890"}, {"title": "Other Title", "id": "11111"}] with ( patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, patch.object(test_bot, "web_click", new_callable = AsyncMock), patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = True), ): mock_find.return_value.attrs = {"content": "some-token"} result = await test_bot.delete_ad(ad_cfg, published_ads, delete_old_ads_by_title = True) assert result is True @pytest.mark.asyncio async def test_delete_ad_by_id(self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any]) -> None: """Test deleting an ad by ID.""" test_bot.page = MagicMock() test_bot.page.evaluate = AsyncMock(return_value = {"statusCode": 200, "statusMessage": "OK", "content": "{}"}) test_bot.page.sleep = AsyncMock() # Create config with ID for deletion by ID ad_cfg = Ad.model_validate( minimal_ad_config | { "id": "12345" # Fixed: use proper dict key syntax } ) published_ads = [{"title": "Different Title", "id": "12345"}, {"title": "Other Title", "id": "11111"}] with ( patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, patch.object(test_bot, "web_click", new_callable = AsyncMock), patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = True), ): mock_find.return_value.attrs = {"content": "some-token"} result = await test_bot.delete_ad(ad_cfg, published_ads, delete_old_ads_by_title = False) assert result is True @pytest.mark.asyncio async def test_delete_ad_by_id_with_non_string_csrf_token(self, test_bot:KleinanzeigenBot, minimal_ad_config:dict[str, Any]) -> None: """Test deleting an ad by ID with non-string CSRF token to cover str() conversion.""" test_bot.page = MagicMock() test_bot.page.evaluate = AsyncMock(return_value = {"statusCode": 200, "statusMessage": "OK", "content": "{}"}) test_bot.page.sleep = AsyncMock() # Create config with ID for deletion by ID ad_cfg = Ad.model_validate(minimal_ad_config | {"id": "12345"}) published_ads = [{"title": "Different Title", "id": "12345"}, {"title": "Other Title", "id": "11111"}] with ( patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, patch.object(test_bot, "web_click", new_callable = AsyncMock), patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = True), patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, ): # Mock non-string CSRF token to test str() conversion mock_find.return_value.attrs = {"content": 12345} # Non-string token result = await test_bot.delete_ad(ad_cfg, published_ads, delete_old_ads_by_title = False) assert result is True # Verify that str() was called on the CSRF token mock_request.assert_called_once() call_args = mock_request.call_args assert call_args[1]["headers"]["x-csrf-token"] == "12345" # Should be converted to string class TestKleinanzeigenBotAdRepublication: """Tests for ad republication functionality.""" def test_check_ad_republication_with_changes(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """Test that ads with changes are marked for republication.""" # Mock the description config to prevent modification of the description test_bot.config.ad_defaults = AdDefaults.model_validate({"description": {"prefix": "", "suffix": ""}}) # Create ad config with all necessary fields for republication ad_cfg = Ad.model_validate( base_ad_config | {"id": "12345", "updated_on": "2024-01-01T00:00:01", "created_on": "2024-01-01T00:00:01", "description": "Changed description"} ) # Create a temporary directory and file with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) ad_dir = temp_path / "ads" ad_dir.mkdir() ad_file = ad_dir / "test_ad.yaml" dicts.save_dict(ad_file, ad_cfg.model_dump()) # Set config file path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] ads_to_publish = test_bot.load_ads() assert len(ads_to_publish) == 1 def test_check_ad_republication_no_changes(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """Test that unchanged ads within interval are not marked for republication.""" current_time = misc.now() three_days_ago = (current_time - timedelta(days = 3)).isoformat() # Create ad config with timestamps for republication check ad_cfg = Ad.model_validate(base_ad_config | {"id": "12345", "updated_on": three_days_ago, "created_on": three_days_ago}) # Calculate hash before making the copy to ensure they match ad_cfg_orig = ad_cfg.model_dump() current_hash = ad_cfg.update_content_hash().content_hash ad_cfg_orig["content_hash"] = current_hash # Mock the config to prevent actual file operations test_bot.config.ad_files = ["test.yaml"] with ( patch("kleinanzeigen_bot.utils.dicts.load_dict_if_exists", return_value = ad_cfg_orig), patch("kleinanzeigen_bot.utils.dicts.load_dict", return_value = {}), ): # Mock ad_fields.yaml ads_to_publish = test_bot.load_ads() assert len(ads_to_publish) == 0 # No ads should be marked for republication class TestKleinanzeigenBotShippingOptions: """Tests for shipping options functionality.""" @pytest.mark.asyncio async def test_shipping_options_mapping(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], tmp_path:Any) -> None: """Test that shipping options are mapped correctly.""" # Create a mock page to simulate browser context test_bot.page = MagicMock() test_bot.page.url = "https://www.kleinanzeigen.de/p-anzeige-aufgeben-bestaetigung.html?adId=12345" test_bot.page.evaluate = AsyncMock() # Create ad config with specific shipping options ad_cfg = Ad.model_validate( base_ad_config | { "shipping_options": ["DHL_2", "Hermes_Päckchen"], "updated_on": "2024-01-01T00:00:00", # Add created_on to prevent KeyError "created_on": "2024-01-01T00:00:00", # Add updated_on for consistency } ) # Create the original ad config and published ads list ad_cfg.update_content_hash() # Add content hash to prevent republication ad_cfg_orig = ad_cfg.model_dump() published_ads:list[dict[str, Any]] = [] # Set up default config values needed for the test test_bot.config.publishing = PublishingConfig.model_validate({"delete_old_ads": "BEFORE_PUBLISH", "delete_old_ads_by_title": False}) # Create temporary file path ad_file = Path(tmp_path) / "test_ad.yaml" # Mock web_execute to handle all JavaScript calls async def mock_web_execute(script:str) -> Any: if script == "document.body.scrollHeight": return 0 # Return integer to prevent scrolling loop return None # Create mock elements csrf_token_elem = MagicMock() csrf_token_elem.attrs = {"content": "csrf-token-123"} shipping_form_elem = MagicMock() shipping_form_elem.attrs = {} shipping_size_radio = MagicMock() shipping_size_radio.attrs = {"checked": False} category_path_elem = MagicMock() category_path_elem.apply = AsyncMock(return_value = "Test Category") # Mock the necessary web interaction methods with ( patch.object(test_bot, "web_execute", side_effect = mock_web_execute), patch.object(test_bot, "web_click", new_callable = AsyncMock), patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, patch.object(test_bot, "web_select", new_callable = AsyncMock), patch.object(test_bot, "web_input", new_callable = AsyncMock), patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "web_sleep", new_callable = AsyncMock), patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = True), patch.object(test_bot, "web_request", new_callable = AsyncMock), patch.object(test_bot, "web_find_all", new_callable = AsyncMock), patch.object(test_bot, "web_await", new_callable = AsyncMock), patch("builtins.input", return_value = ""), patch.object(test_bot, "web_scroll_page_down", new_callable = AsyncMock), ): # Mock web_find to simulate element detection async def mock_find_side_effect(selector_type:By, selector_value:str, **_:Any) -> Element | None: if selector_value == "meta[name=_csrf]": return csrf_token_elem if selector_value == "myftr-shppngcrt-frm": return shipping_form_elem if selector_type == By.ID and selector_value.startswith("radio-button-"): return shipping_size_radio if selector_value == "postad-category-path": return category_path_elem return None mock_find.side_effect = mock_find_side_effect # Mock web_check to return True for radio button checked state with patch.object(test_bot, "web_check", new_callable = AsyncMock) as mock_check: mock_check.return_value = True # Test through the public interface by publishing an ad await test_bot.publish_ad(str(ad_file), ad_cfg, ad_cfg_orig, published_ads) # Verify that web_find was called the expected number of times assert mock_find.await_count >= 3 # Verify the file was created in the temporary directory assert ad_file.exists() @pytest.mark.asyncio async def test_cross_drive_path_fallback_windows(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """Test that cross-drive path handling falls back to absolute path on Windows.""" # Create ad config ad_cfg = Ad.model_validate( base_ad_config | { "updated_on": "2024-01-01T00:00:00", "created_on": "2024-01-01T00:00:00", "auto_price_reduction": {"enabled": True, "strategy": "FIXED", "amount": 10, "min_price": 50, "delay_reposts": 0, "delay_days": 0}, "price": 100, "repost_count": 1, "price_reduction_count": 0, } ) ad_cfg.update_content_hash() ad_cfg_orig = ad_cfg.model_dump() # Simulate Windows cross-drive scenario # Config on D:, ad file on C: test_bot.config_file_path = "D:\\project\\config.yaml" ad_file = "C:\\temp\\test_ad.yaml" # Create a sentinel exception to abort publish_ad early class _SentinelException(Exception): pass # Track what path argument __apply_auto_price_reduction receives recorded_path:list[str] = [] def mock_apply_auto_price_reduction(ad_cfg:Ad, ad_cfg_orig:dict[str, Any], ad_file_relative:str) -> None: recorded_path.append(ad_file_relative) raise _SentinelException("Abort early for test") # Mock Path to use PureWindowsPath for testing cross-drive behavior with ( patch("kleinanzeigen_bot.Path", PureWindowsPath), patch("kleinanzeigen_bot.apply_auto_price_reduction", side_effect = mock_apply_auto_price_reduction), patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "delete_ad", new_callable = AsyncMock), ): # Call publish_ad and expect sentinel exception try: await test_bot.publish_ad(ad_file, ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.REPLACE) pytest.fail("Expected _SentinelException to be raised") except _SentinelException: # This is expected - the test aborts early pass # Verify the path argument is the absolute path (fallback behavior) assert len(recorded_path) == 1 assert recorded_path[0] == ad_file, f"Expected absolute path fallback, got: {recorded_path[0]}" @pytest.mark.asyncio async def test_auto_price_reduction_only_on_replace_not_update(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any], tmp_path:Path) -> None: """Test that auto price reduction is ONLY applied on REPLACE mode, not UPDATE.""" # Create ad with auto price reduction enabled ad_cfg = Ad.model_validate( base_ad_config | { "id": 12345, "price": 200, "auto_price_reduction": {"enabled": True, "strategy": "FIXED", "amount": 50, "min_price": 50, "delay_reposts": 0, "delay_days": 0}, "repost_count": 1, "price_reduction_count": 0, "updated_on": "2024-01-01T00:00:00", "created_on": "2024-01-01T00:00:00", } ) ad_cfg.update_content_hash() ad_cfg_orig = ad_cfg.model_dump() # Mock the private __apply_auto_price_reduction method with patch("kleinanzeigen_bot.apply_auto_price_reduction") as mock_apply: # Mock other dependencies mock_response = {"statusCode": 200, "statusMessage": "OK", "content": "{}"} with ( patch.object(test_bot, "web_find", new_callable = AsyncMock), patch.object(test_bot, "web_input", new_callable = AsyncMock), patch.object(test_bot, "web_click", new_callable = AsyncMock), patch.object(test_bot, "web_open", new_callable = AsyncMock), patch.object(test_bot, "web_select", new_callable = AsyncMock), patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = False), patch.object(test_bot, "web_await", new_callable = AsyncMock), patch.object(test_bot, "web_sleep", new_callable = AsyncMock), patch.object(test_bot, "web_execute", new_callable = AsyncMock, return_value = mock_response), patch.object(test_bot, "web_request", new_callable = AsyncMock, return_value = mock_response), patch.object(test_bot, "web_scroll_page_down", new_callable = AsyncMock), patch.object(test_bot, "web_find_all", new_callable = AsyncMock, return_value = []), patch.object(test_bot, "check_and_wait_for_captcha", new_callable = AsyncMock), patch("builtins.input", return_value = ""), patch("kleinanzeigen_bot.utils.misc.ainput", new_callable = AsyncMock, return_value = ""), ): test_bot.page = MagicMock() test_bot.page.url = "https://www.kleinanzeigen.de/p-anzeige-aufgeben-bestaetigung.html?adId=12345" test_bot.config.publishing.delete_old_ads = "BEFORE_PUBLISH" # Test REPLACE mode - should call __apply_auto_price_reduction await test_bot.publish_ad(str(tmp_path / "ad.yaml"), ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.REPLACE) assert mock_apply.call_count == 1, "Auto price reduction should be called on REPLACE" # Reset mock mock_apply.reset_mock() # Test MODIFY mode - should NOT call __apply_auto_price_reduction await test_bot.publish_ad(str(tmp_path / "ad.yaml"), ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.MODIFY) assert mock_apply.call_count == 0, "Auto price reduction should NOT be called on MODIFY" @pytest.mark.asyncio async def test_special_attributes_with_non_string_values(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """Test that special attributes with non-string values are converted to strings.""" # Create ad config with string special attributes first (to pass validation) ad_cfg = Ad.model_validate( base_ad_config | { "special_attributes": { "art_s": "12345", # String value initially "condition_s": "67890", # String value initially "color_s": "red", # String value }, "updated_on": "2024-01-01T00:00:00", "created_on": "2024-01-01T00:00:00", } ) # Now modify the special attributes to non-string values to test str() conversion # This simulates the scenario where the values come from external sources as non-strings # We need to cast to Any to bypass type checking for this test special_attrs = cast(Any, ad_cfg.special_attributes) special_attrs["art_s"] = 12345 # Non-string value special_attrs["condition_s"] = 67890 # Non-string value # Mock special attribute elements art_s_elem = MagicMock() art_s_attrs = MagicMock() art_s_attrs.id = "art_s" art_s_attrs.name = "art_s" art_s_elem.attrs = art_s_attrs art_s_elem.local_name = "select" condition_s_elem = MagicMock() condition_s_attrs = MagicMock() condition_s_attrs.id = "condition_s" condition_s_attrs.name = "condition_s" condition_s_elem.attrs = condition_s_attrs condition_s_elem.local_name = "select" color_s_elem = MagicMock() color_s_attrs = MagicMock() color_s_attrs.id = "color_s" color_s_attrs.name = "color_s" color_s_elem.attrs = color_s_attrs color_s_elem.local_name = "select" # Mock the necessary web interaction methods with ( patch.object(test_bot, "web_find", new_callable = AsyncMock) as mock_find, patch.object(test_bot, "web_select", new_callable = AsyncMock) as mock_select, patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = True), patch.object(test_bot, "_KleinanzeigenBot__set_condition", new_callable = AsyncMock) as mock_set_condition, ): # Mock web_find to simulate element detection async def mock_find_side_effect(selector_type:By, selector_value:str, **_:Any) -> Element | None: # Handle XPath queries for special attributes if selector_type == By.XPATH and "contains(@name" in selector_value: if "art_s" in selector_value: return art_s_elem if "condition_s" in selector_value: return condition_s_elem if "color_s" in selector_value: return color_s_elem return None mock_find.side_effect = mock_find_side_effect # Test the __set_special_attributes method directly await getattr(test_bot, "_KleinanzeigenBot__set_special_attributes")(ad_cfg) # Verify that web_select was called with string values (str() conversion) mock_select.assert_any_call(By.ID, "art_s", "12345") # Converted to string mock_select.assert_any_call(By.ID, "color_s", "red") # Already string # Verify that __set_condition was called with string value mock_set_condition.assert_called_once_with("67890") # Converted to string class TestShippingSelectorTimeout: """Regression tests for commercial shipping selector (versand_s) timeout handling. Ensures that TimeoutError from web_check (element absent) is caught gracefully, while TimeoutError from web_select (element found but interaction fails) propagates. """ @pytest.mark.asyncio async def test_missing_versand_s_falls_back_to_dialog(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """When versand_s selector is absent, web_check raises TimeoutError and the bot falls through to dialog-based shipping.""" ad_cfg = Ad.model_validate(base_ad_config | {"shipping_type": "SHIPPING"}) with ( patch.object(test_bot, "web_check", new_callable = AsyncMock, side_effect = TimeoutError("element not found")) as mock_check, patch.object(test_bot, "web_select", new_callable = AsyncMock) as mock_select, patch.object(test_bot, "web_click", new_callable = AsyncMock) as mock_click, patch.object(test_bot, "web_find", new_callable = AsyncMock), patch.object(test_bot, "web_input", new_callable = AsyncMock), ): await getattr(test_bot, "_KleinanzeigenBot__set_shipping")(ad_cfg) # Probe must have been awaited with quick_dom timeout mock_check.assert_awaited_once() assert mock_check.await_args is not None assert mock_check.await_args.kwargs["timeout"] == test_bot._timeout("quick_dom") # web_select must NOT have been called with versand_s (commercial path was skipped) for call in mock_select.call_args_list: assert "versand_s" not in str(call), "web_select should not be called for versand_s when element is absent" # Dialog-based fallback should have been triggered (click on "Versandmethoden auswählen") clicked_selectors = [str(c) for c in mock_click.call_args_list] assert any("Versandmethoden" in s for s in clicked_selectors), \ "Expected dialog-based shipping fallback when versand_s is absent" @pytest.mark.asyncio async def test_visible_versand_s_uses_commercial_select(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """When versand_s selector is present, web_check succeeds and web_select sets the value.""" ad_cfg = Ad.model_validate(base_ad_config | {"shipping_type": "SHIPPING"}) with ( patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = True) as mock_check, patch.object(test_bot, "web_select", new_callable = AsyncMock) as mock_select, patch.object(test_bot, "web_click", new_callable = AsyncMock) as mock_click, ): await getattr(test_bot, "_KleinanzeigenBot__set_shipping")(ad_cfg) # Probe must have been awaited with quick_dom timeout mock_check.assert_awaited_once() assert mock_check.await_args is not None assert mock_check.await_args.kwargs["timeout"] == test_bot._timeout("quick_dom") # web_select must have been awaited with versand_s and "ja" (SHIPPING) mock_select.assert_awaited_once_with(By.XPATH, '//select[contains(@id, ".versand_s")]', "ja") # Dialog-based fallback should NOT have been triggered clicked_selectors = [str(c) for c in mock_click.call_args_list] assert not any("Versandmethoden" in s for s in clicked_selectors), \ "Dialog-based shipping should not be triggered when versand_s is present" @pytest.mark.asyncio async def test_web_select_timeout_propagates_after_successful_probe(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """When web_check succeeds but web_select raises TimeoutError, the error must propagate (not be swallowed).""" ad_cfg = Ad.model_validate(base_ad_config | {"shipping_type": "SHIPPING"}) with ( patch.object(test_bot, "web_check", new_callable = AsyncMock, return_value = True) as mock_check, patch.object(test_bot, "web_select", new_callable = AsyncMock, side_effect = TimeoutError("select timed out")), pytest.raises(TimeoutError, match = "select timed out"), ): await getattr(test_bot, "_KleinanzeigenBot__set_shipping")(ad_cfg) # Probe must have been awaited with quick_dom timeout mock_check.assert_awaited_once() assert mock_check.await_args is not None assert mock_check.await_args.kwargs["timeout"] == test_bot._timeout("quick_dom") class TestKleinanzeigenBotUrlConstruction: """Tests for URL construction functionality.""" def test_url_construction(self, test_bot:KleinanzeigenBot) -> None: """Test that URLs are constructed correctly.""" # Test login URL expected_login_url = "https://www.kleinanzeigen.de/m-einloggen.html?targetUrl=/" assert f"{test_bot.root_url}/m-einloggen.html?targetUrl=/" == expected_login_url # Test ad management URL expected_manage_url = "https://www.kleinanzeigen.de/m-meine-anzeigen.html" assert f"{test_bot.root_url}/m-meine-anzeigen.html" == expected_manage_url # Test ad publishing URL expected_publish_url = "https://www.kleinanzeigen.de/p-anzeige-aufgeben-schritt2.html" assert f"{test_bot.root_url}/p-anzeige-aufgeben-schritt2.html" == expected_publish_url class TestKleinanzeigenBotPrefixSuffix: """Tests for description prefix and suffix functionality.""" # pylint: disable=protected-access def test_description_prefix_suffix_handling(self, test_bot_config:Config, description_test_cases:list[tuple[dict[str, Any], str, str]]) -> None: """Test handling of description prefix/suffix in various configurations.""" for config, raw_description, expected_description in description_test_cases: test_bot = KleinanzeigenBot() test_bot.config = test_bot_config.with_values(config) ad_cfg = test_bot.load_ad( { "description": raw_description, "active": True, "title": "0123456789", "category": "whatever", } ) # Access private method using the correct name mangling description = getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert description == expected_description def test_description_length_validation(self, test_bot_config:Config) -> None: """Test that long descriptions with affixes raise appropriate error.""" test_bot = KleinanzeigenBot() test_bot.config = test_bot_config.with_values({"ad_defaults": {"description_prefix": "P" * 1000, "description_suffix": "S" * 1000}}) ad_cfg = test_bot.load_ad( { "description": "D" * 2001, # This plus affixes will exceed 4000 chars "active": True, "title": "0123456789", "category": "whatever", } ) with pytest.raises(AssertionError) as exc_info: getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert "Length of ad description including prefix and suffix exceeds 4000 chars" in str(exc_info.value) assert "Description length: 4001" in str(exc_info.value) class TestKleinanzeigenBotDescriptionHandling: """Tests for description handling functionality.""" def test_description_without_main_config_description(self, test_bot_config:Config) -> None: """Test that description works correctly when description is missing from main config.""" test_bot = KleinanzeigenBot() test_bot.config = test_bot_config # Test with a simple ad config ad_cfg = test_bot.load_ad( { "description": "Test Description", "active": True, "title": "0123456789", "category": "whatever", } ) # The description should be returned as-is without any prefix/suffix description = getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert description == "Test Description" def test_description_with_only_new_format_affixes(self, test_bot_config:Config) -> None: """Test that description works with only new format affixes in config.""" test_bot = KleinanzeigenBot() test_bot.config = test_bot_config.with_values({"ad_defaults": {"description_prefix": "Prefix: ", "description_suffix": " :Suffix"}}) ad_cfg = test_bot.load_ad( { "description": "Test Description", "active": True, "title": "0123456789", "category": "whatever", } ) description = getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert description == "Prefix: Test Description :Suffix" def test_description_with_mixed_config_formats(self, test_bot_config:Config) -> None: """Test that description works with both old and new format affixes in config.""" test_bot = KleinanzeigenBot() test_bot.config = test_bot_config.with_values( { "ad_defaults": { "description_prefix": "New Prefix: ", "description_suffix": " :New Suffix", "description": {"prefix": "Old Prefix: ", "suffix": " :Old Suffix"}, } } ) ad_cfg = test_bot.load_ad( { "description": "Test Description", "active": True, "title": "0123456789", "category": "whatever", } ) description = getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert description == "New Prefix: Test Description :New Suffix" def test_description_with_ad_level_affixes(self, test_bot_config:Config) -> None: """Test that ad-level affixes take precedence over config affixes.""" test_bot = KleinanzeigenBot() test_bot.config = test_bot_config.with_values({"ad_defaults": {"description_prefix": "Config Prefix: ", "description_suffix": " :Config Suffix"}}) ad_cfg = test_bot.load_ad( { "description": "Test Description", "description_prefix": "Ad Prefix: ", "description_suffix": " :Ad Suffix", "active": True, "title": "0123456789", "category": "whatever", } ) description = getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert description == "Ad Prefix: Test Description :Ad Suffix" def test_description_with_none_values(self, test_bot_config:Config) -> None: """Test that None values in affixes are handled correctly.""" test_bot = KleinanzeigenBot() test_bot.config = test_bot_config.with_values( {"ad_defaults": {"description_prefix": None, "description_suffix": None, "description": {"prefix": None, "suffix": None}}} ) ad_cfg = test_bot.load_ad( { "description": "Test Description", "active": True, "title": "0123456789", "category": "whatever", } ) description = getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert description == "Test Description" def test_description_with_email_replacement(self, test_bot_config:Config) -> None: """Test that @ symbols in description are replaced with (at).""" test_bot = KleinanzeigenBot() test_bot.config = test_bot_config ad_cfg = test_bot.load_ad( { "description": "Contact: test@example.com", "active": True, "title": "0123456789", "category": "whatever", } ) description = getattr(test_bot, "_KleinanzeigenBot__get_description")(ad_cfg, with_affixes = True) assert description == "Contact: test(at)example.com" class TestKleinanzeigenBotChangedAds: """Tests for the 'changed' ads selector functionality.""" def test_load_ads_with_changed_selector(self, test_bot_config:Config, base_ad_config:dict[str, Any]) -> None: """Test that only changed ads are loaded when using the 'changed' selector.""" # Set up the bot with the 'changed' selector test_bot = KleinanzeigenBot() test_bot.ads_selector = "changed" test_bot.config = test_bot_config.with_values({"ad_defaults": {"description": {"prefix": "", "suffix": ""}}}) # Create a changed ad ad_cfg = Ad.model_validate( base_ad_config | {"id": "12345", "title": "Changed Ad", "updated_on": "2024-01-01T00:00:00", "created_on": "2024-01-01T00:00:00", "active": True} ) # Calculate hash for changed_ad and add it to the config # Then modify the ad to simulate a change changed_ad = ad_cfg.model_dump() changed_hash = ad_cfg.update_content_hash().content_hash changed_ad["content_hash"] = changed_hash # Now modify the ad to make it "changed" changed_ad["title"] = "Changed Ad - Modified" # Create temporary directory and file with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) ad_dir = temp_path / "ads" ad_dir.mkdir() # Write the ad file dicts.save_dict(ad_dir / "changed_ad.yaml", changed_ad) # Set config file path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] # Mock the loading of the ad configuration with patch( "kleinanzeigen_bot.utils.dicts.load_dict", side_effect = [ changed_ad, # First call returns the changed ad {}, # Second call for ad_fields.yaml ], ): ads_to_publish = test_bot.load_ads() # The changed ad should be loaded assert len(ads_to_publish) == 1 assert ads_to_publish[0][1].title == "Changed Ad - Modified" def test_load_ads_with_due_selector_includes_all_due_ads(self, test_bot:KleinanzeigenBot, base_ad_config:dict[str, Any]) -> None: """Test that 'due' selector includes all ads that are due for republication, regardless of changes.""" # Set up the bot with the 'due' selector test_bot.ads_selector = "due" # Create a changed ad that is also due for republication current_time = misc.now() old_date = (current_time - timedelta(days = 10)).isoformat() # Past republication interval ad_cfg = Ad.model_validate( base_ad_config | { "id": "12345", "title": "Changed Ad", "updated_on": old_date, "created_on": old_date, "republication_interval": 7, # Due for republication after 7 days "active": True, } ) changed_ad = ad_cfg.model_dump() # Create temporary directory and file with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) ad_dir = temp_path / "ads" ad_dir.mkdir() # Write the ad file dicts.save_dict(ad_dir / "changed_ad.yaml", changed_ad) # Set config file path and use relative path for ad_files test_bot.config_file_path = str(temp_path / "config.yaml") test_bot.config.ad_files = ["ads/*.yaml"] # Mock the loading of the ad configuration with patch( "kleinanzeigen_bot.utils.dicts.load_dict", side_effect = [ changed_ad, # First call returns the changed ad {}, # Second call for ad_fields.yaml ], ): ads_to_publish = test_bot.load_ads() # The changed ad should be loaded with 'due' selector because it's due for republication assert len(ads_to_publish) == 1 def test_file_logger_writes_message(tmp_path:Path, caplog:pytest.LogCaptureFixture) -> None: """ Unit: Logger can be initialized and used, robust to pytest log capture. """ log_path = tmp_path / "logger_test.log" logger_name = "logger_test_logger_unique" logger = logging.getLogger(logger_name) logger.setLevel(logging.DEBUG) logger.propagate = False for h in list(logger.handlers): logger.removeHandler(h) handle = logging.FileHandler(str(log_path), encoding = "utf-8") handle.setLevel(logging.DEBUG) formatter = logging.Formatter("%(levelname)s:%(name)s:%(message)s") handle.setFormatter(formatter) logger.addHandler(handle) logger.info("Logger test log message") handle.flush() handle.close() logger.removeHandler(handle) assert log_path.exists() with open(log_path, "r", encoding = "utf-8") as f: contents = f.read() assert "Logger test log message" in contents def _apply_price_reduction_persistence(count:int | None) -> dict[str, Any]: """Return a dict with price_reduction_count only when count is positive (count -> dict[str, Any]).""" ad_cfg_orig:dict[str, Any] = {} if count is not None and count > 0: ad_cfg_orig["price_reduction_count"] = count return ad_cfg_orig class TestPriceReductionPersistence: """Tests for price_reduction_count persistence logic.""" @pytest.mark.unit def test_persistence_logic_saves_when_count_positive(self) -> None: """Test the conditional logic that decides whether to persist price_reduction_count.""" # Simulate the logic from publish_ad lines 1076-1079 # Test case 1: price_reduction_count = 3 (should persist) ad_cfg_orig = _apply_price_reduction_persistence(3) assert "price_reduction_count" in ad_cfg_orig assert ad_cfg_orig["price_reduction_count"] == 3 @pytest.mark.unit def test_persistence_logic_skips_when_count_zero(self) -> None: """Test that price_reduction_count == 0 does not get persisted.""" # Test case 2: price_reduction_count = 0 (should NOT persist) ad_cfg_orig = _apply_price_reduction_persistence(0) assert "price_reduction_count" not in ad_cfg_orig @pytest.mark.unit def test_persistence_logic_skips_when_count_none(self) -> None: """Test that price_reduction_count == None does not get persisted.""" # Test case 3: price_reduction_count = None (should NOT persist) ad_cfg_orig = _apply_price_reduction_persistence(None) assert "price_reduction_count" not in ad_cfg_orig