diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index d3e54d8..6278561 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -51,7 +51,7 @@ This section provides quick reference commands for common development tasks. See pdm run app |-> executes 'python -m kleinanzeigen_bot' |-> executes 'kleinanzeigen_bot/__main__.py' - |-> executes main() function of 'kleinanzeigen_bot/__init__.py' + |-> executes main() function of 'kleinanzeigen_bot/cli.py' |-> executes KleinanzeigenBot().run() ``` diff --git a/pyproject.toml b/pyproject.toml index 81cce44..d4f090b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -107,11 +107,11 @@ format = { composite = ["format:py", "format:yaml"] } "format:yaml" = "yamlfix scripts/ src/ tests/" lint = { composite = ["lint:ruff", "lint:mypy", "lint:pyright", "lint:actions"] } -"lint:ruff" = "ruff check --preview" -"lint:mypy" = "mypy" -"lint:pyright" = "basedpyright" -"lint:actions" = "actionlint" -"lint:fix" = {shell = "ruff check --preview --fix" } +"lint:ruff" = { shell = "echo [ruff] && ruff check --preview" } +"lint:mypy" = { shell = "echo [mypy] && mypy" } +"lint:pyright" = { shell = "echo [pyright] && basedpyright" } +"lint:actions" = { shell = "echo [actionlint] && actionlint" } +"lint:fix" = { shell = "echo [ruff:fix] && ruff check --preview --fix" } # tests # Public test commands diff --git a/src/kleinanzeigen_bot/__init__.py b/src/kleinanzeigen_bot/__init__.py index 3cc4187..6b63141 100644 --- a/src/kleinanzeigen_bot/__init__.py +++ b/src/kleinanzeigen_bot/__init__.py @@ -1,16 +1,15 @@ # SPDX-FileCopyrightText: © Sebastian Thomschke and contributors # SPDX-License-Identifier: AGPL-3.0-or-later # SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ -import atexit, asyncio, enum, json, os, re, signal, sys, textwrap # isort: skip -import getopt # pylint: disable=deprecated-module +import asyncio, enum, importlib, json, os, re, sys # isort: skip import urllib.parse as urllib_parse from dataclasses import dataclass, replace from datetime import datetime from gettext import gettext as _ from pathlib import Path -from typing import Any, Final, Sequence, cast +from typing import TYPE_CHECKING, Any, Final, Sequence, cast -import certifi, colorama, nodriver # isort: skip +import certifi, colorama # isort: skip from nodriver.core.connection import ProtocolException from ruamel.yaml import YAML from wcmatch import glob @@ -21,7 +20,7 @@ from . import download_selection as _download_selection from . import extract from . import local_path_renaming as _local_path_renaming from . import price_reduction as _price_reduction -from . import resources as _resources +from . import runtime_config as _runtime_config from ._version import __version__ from .ad_description import get_ad_description from .model.ad_model import ( @@ -39,24 +38,24 @@ from .model.config_model import DEFAULT_DOWNLOAD_DIR, Config from .update_checker import UpdateChecker from .utils import diagnostics as _diagnostics from .utils import dicts as _dicts -from .utils import error_handlers as _error_handlers from .utils import loggers as _loggers from .utils import misc as _misc from .utils import xdg_paths as _xdg_paths from .utils.exceptions import CaptchaEncountered, CategoryResolutionError, PublishedAdsFetchIncompleteError, PublishSubmissionUncertainError from .utils.files import abspath -from .utils.i18n import Locale, get_current_locale, pluralize, set_current_locale +from .utils.i18n import pluralize from .utils.misc import ainput, ensure, is_frozen -from .utils.timing_collector import TimingCollector from .utils.web_scraping_mixin import By, Element, Is, WebScrapingMixin +if TYPE_CHECKING: + from .utils.timing_collector import TimingCollector + # W0406: possibly a bug, see https://github.com/PyCQA/pylint/issues/3933 LOG:Final[_loggers.Logger] = _loggers.get_logger(__name__) LOG.setLevel(_loggers.INFO) SUBMISSION_MAX_RETRIES:Final[int] = 3 -_LOGIN_ENV_PATTERN:Final[re.Pattern[str]] = re.compile(r"^\$\{(?P\w+)(?::-(?P.*))?\}$") _LOGIN_DETECTION_SELECTORS:Final[list[tuple["By", str]]] = [ (By.CLASS_NAME, "mr-medium"), (By.ID, "user-email"), @@ -132,7 +131,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 self.keep_old_ads = False self._login_detection_diagnostics_captured:bool = False - self._timing_collector:TimingCollector | None = None + self._timing_collector:"TimingCollector | None" = None def __del__(self) -> None: if self.file_log: @@ -190,73 +189,80 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 await ad_extractor.download_ad(ad_id, active = resolved.active) - def _resolve_workspace(self) -> None: - """ - Resolve workspace paths after CLI args are parsed. - """ - if self.command in {"help", "version", "create-config"}: - return - effective_config_arg = self._config_arg - effective_workspace_mode = self._workspace_mode_arg - if not effective_config_arg: - default_config = (Path.cwd() / "config.yaml").resolve() - if self.config_file_path and Path(self.config_file_path).resolve() != default_config: - effective_config_arg = self.config_file_path - if effective_workspace_mode is None: - # Backward compatibility for tests/programmatic assignment of config_file_path: - # infer a stable default from the configured path location. - config_path = Path(self.config_file_path).resolve() - xdg_config_dir = _xdg_paths.get_xdg_base_dir("config").resolve() - effective_workspace_mode = "xdg" if config_path.is_relative_to(xdg_config_dir) else "portable" + async def run(self, args:list[str]) -> None: # noqa: PLR0915 + _cli = importlib.import_module(".cli", __name__) + parsed = _cli.parse_args(args) + self.command = parsed.command + self.ads_selector = parsed.ads_selector + self._ads_selector_explicit = parsed.ads_selector_explicit + self.keep_old_ads = parsed.keep_old_ads + self._config_arg = parsed.config_arg + self._workspace_mode_arg = cast(_xdg_paths.InstallationMode, parsed.workspace_mode) if parsed.workspace_mode else None + self._logfile_arg = parsed.logfile_arg + self._logfile_explicitly_provided = parsed.logfile_explicitly_provided + if parsed.config_file_path is not None: + self.config_file_path = parsed.config_file_path + if parsed.logfile_explicitly_provided: + self.log_file_path = parsed.log_file_path - try: - self.workspace = _xdg_paths.resolve_workspace( - config_arg = effective_config_arg, - logfile_arg = self._logfile_arg, - workspace_mode = effective_workspace_mode, - logfile_explicitly_provided = self._logfile_explicitly_provided, - log_basename = self._log_basename, + self.workspace = _runtime_config.resolve_workspace( + command = self.command, + config_file_path = self.config_file_path, + config_arg = self._config_arg, + logfile_arg = self._logfile_arg, + workspace_mode = self._workspace_mode_arg, + logfile_explicitly_provided = self._logfile_explicitly_provided, + log_basename = self._log_basename, + ) + if self.workspace is not None: + self.config_file_path = str(self.workspace.config_file) + self.log_file_path = str(self.workspace.log_file) if self.workspace.log_file else None + + def _bootstrap_runtime() -> None: + self.file_log = _runtime_config.configure_file_logging( + self.log_file_path, + self.workspace, + self.file_log, + self.get_version(), ) - except ValueError as exc: - LOG.error(str(exc)) - sys.exit(2) + runtime_state = _runtime_config.load_config(self.config_file_path, self.workspace, self.command) + self.config = runtime_state.config + self.categories = runtime_state.categories + self._timing_collector = runtime_state.timing_collector + _runtime_config.apply_browser_config(self.browser_config, self.config, self.workspace, self.config_file_path) - _xdg_paths.ensure_directory(self.workspace.config_file.parent, "config directory") - - self.config_file_path = str(self.workspace.config_file) - self.log_file_path = str(self.workspace.log_file) if self.workspace.log_file else None - - LOG.info("Config: %s", self.workspace.config_file) - LOG.info("Workspace mode: %s", self.workspace.mode) - LOG.info("Workspace: %s", self.workspace.config_dir) - if _loggers.is_debug(LOG): - LOG.debug("Log file: %s", self.workspace.log_file) - LOG.debug("State dir: %s", self.workspace.state_dir) - LOG.debug("Download dir: %s", self.workspace.download_dir) - LOG.debug("Browser profile: %s", self.workspace.browser_profile_dir) - LOG.debug("Diagnostics dir: %s", self.workspace.diagnostics_dir) - - async def run(self, args:list[str]) -> None: - self.parse_args(args) - self._resolve_workspace() try: + # When adding/removing a case, also update runtime_config.VALID_COMMANDS. match self.command: case "help": - self.show_help() + _cli.show_help() return case "version": print(self.get_version()) case "create-config": - self.create_default_config() + if self.workspace is None and self._workspace_mode_arg is not None: + try: + workspace = _xdg_paths.resolve_workspace( + config_arg = self._config_arg, + logfile_arg = self._logfile_arg, + workspace_mode = self._workspace_mode_arg, + logfile_explicitly_provided = self._logfile_explicitly_provided, + log_basename = self._log_basename, + ) + self.workspace = workspace + self.config_file_path = str(workspace.config_file) + self.log_file_path = str(workspace.log_file) if workspace.log_file else None + except ValueError as exc: + LOG.error(str(exc)) + sys.exit(2) + _runtime_config.create_default_config(self.config_file_path, self.workspace) return case "diagnose": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() self.diagnose_browser_issues() return case "verify": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() # Check for updates on startup checker = UpdateChecker(self.config, self._update_check_state_path) checker.check_for_updates() @@ -273,13 +279,11 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 LOG.info("DONE: No configuration errors found.") LOG.info("############################################") case "update-check": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() checker = UpdateChecker(self.config, self._update_check_state_path) checker.check_for_updates(skip_interval_check = True) case "update-content-hash": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() # Check for updates on startup checker = UpdateChecker(self.config, self._update_check_state_path) checker.check_for_updates() @@ -291,8 +295,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 LOG.info("DONE: No active ads found.") LOG.info("############################################") case "publish": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() # Check for updates on startup checker = UpdateChecker(self.config, self._update_check_state_path) checker.check_for_updates() @@ -315,8 +318,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 LOG.info("DONE: No new/outdated ads found.") LOG.info("############################################") case "update": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() if not self._is_valid_ads_selector({"all", "changed"}): if self._ads_selector_explicit: @@ -333,8 +335,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 LOG.info("DONE: No changed ads found.") LOG.info("############################################") case "delete": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() # Check for updates on startup checker = UpdateChecker(self.config, self._update_check_state_path) checker.check_for_updates() @@ -347,8 +348,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 LOG.info("DONE: No ads to delete found.") LOG.info("############################################") case "extend": - self.configure_file_logging() - self.load_config() + _bootstrap_runtime() # Check for updates on startup checker = UpdateChecker(self.config, self._update_check_state_path) checker.check_for_updates() @@ -370,14 +370,13 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 LOG.info("DONE: No ads found to extend.") LOG.info("############################################") case "download": - self.configure_file_logging() # ad IDs depends on selector if not self._is_valid_ads_selector({"all", "new"}): if self._ads_selector_explicit: LOG.error('Invalid --ads selector: "%s". Valid values: comma-separated keywords (all, new) or numeric IDs.', self.ads_selector) sys.exit(2) self.ads_selector = "new" - self.load_config() + _bootstrap_runtime() # Check for updates on startup checker = UpdateChecker(self.config, self._update_check_state_path) checker.check_for_updates() @@ -397,127 +396,6 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 except Exception as exc: # noqa: BLE001 LOG.warning("Timing collector flush failed: %s", exc) - def show_help(self) -> None: - if is_frozen(): - exe = sys.argv[0] - elif os.getenv("PDM_PROJECT_ROOT", ""): - exe = "pdm run app" - else: - exe = "python -m kleinanzeigen_bot" - - if get_current_locale().language == "de": - print( - textwrap.dedent( - f"""\ - Verwendung: {colorama.Fore.LIGHTMAGENTA_EX}{exe} BEFEHL [OPTIONEN]{colorama.Style.RESET_ALL} - - Befehle: - publish - (Wieder-)Veröffentlicht Anzeigen - verify - Überprüft die Konfigurationsdateien - delete - Löscht Anzeigen - update - Aktualisiert bestehende Anzeigen - extend - Verlängert Anzeigen innerhalb des 8-Tage-Zeitfensters - download - Lädt eine oder mehrere Anzeigen herunter - update-check - Prüft auf verfügbare Updates - update-content-hash - Berechnet den content_hash aller Anzeigen anhand der aktuellen ad_defaults neu; - nach Änderungen an den config.yaml/ad_defaults verhindert es, dass alle Anzeigen als - "geändert" gelten und neu veröffentlicht werden. - create-config - Erstellt eine neue Standard-Konfigurationsdatei, falls noch nicht vorhanden - diagnose - Diagnostiziert Browser-Verbindungsprobleme und zeigt Troubleshooting-Informationen - -- - help - Zeigt diese Hilfe an (Standardbefehl) - version - Zeigt die Version der Anwendung an - - Optionen: - --ads=all|due|new|changed| (publish) - Gibt an, welche Anzeigen (erneut) veröffentlicht werden sollen (STANDARD: due) - Mögliche Werte: - * all: Veröffentlicht alle Anzeigen erneut, ignoriert republication_interval - * due: Veröffentlicht alle neuen Anzeigen und erneut entsprechend dem republication_interval - * new: Veröffentlicht nur neue Anzeigen (d.h. Anzeigen ohne ID in der Konfigurationsdatei) - * changed: Veröffentlicht nur Anzeigen, die seit der letzten Veröffentlichung geändert wurden - * : Gibt eine oder mehrere Anzeigen-IDs an, die veröffentlicht werden sollen, z. B. "--ads=1,2,3", ignoriert republication_interval - * Kombinationen: Sie können mehrere Selektoren mit Kommas kombinieren, z. B. "--ads=changed,due" um sowohl geänderte als auch - fällige Anzeigen zu veröffentlichen - --ads=all|new| (download) - Gibt an, welche Anzeigen heruntergeladen werden sollen (STANDARD: new) - Mögliche Werte: - * all: Lädt alle Anzeigen aus Ihrem Profil herunter - * new: Lädt Anzeigen aus Ihrem Profil herunter, die lokal noch nicht gespeichert sind - * : Gibt eine oder mehrere Anzeigen-IDs zum Herunterladen an, z. B. "--ads=1,2,3" - --ads=all|changed| (update) - Gibt an, welche Anzeigen aktualisiert werden sollen (STANDARD: changed) - Mögliche Werte: - * all: Aktualisiert alle Anzeigen - * changed: Aktualisiert nur Anzeigen, die seit der letzten Veröffentlichung geändert wurden - * : Gibt eine oder mehrere Anzeigen-IDs zum Aktualisieren an, z. B. "--ads=1,2,3" - --ads=all| (extend) - Gibt an, welche Anzeigen verlängert werden sollen - Mögliche Werte: - * all: Verlängert alle Anzeigen, die innerhalb von 8 Tagen ablaufen - * : Gibt bestimmte Anzeigen-IDs an, z. B. "--ads=1,2,3" - --force - Alias für '--ads=all' - --keep-old - Verhindert das Löschen alter Anzeigen bei erneuter Veröffentlichung - --config= - Pfad zur YAML- oder JSON-Konfigurationsdatei (ändert den Workspace-Modus nicht implizit) - --workspace-mode=portable|xdg - Überschreibt den Workspace-Modus für diesen Lauf - --logfile= - Pfad zur Protokolldatei (STANDARD: vom aktiven Workspace-Modus abhängig) - --lang=en|de - Anzeigesprache (STANDARD: Systemsprache, wenn unterstützt, sonst Englisch) - -v, --verbose - Aktiviert detaillierte Ausgabe – nur nützlich zur Fehlerbehebung - """.rstrip() - ) - ) - else: - print( - textwrap.dedent( - f"""\ - Usage: {colorama.Fore.LIGHTMAGENTA_EX}{exe} COMMAND [OPTIONS]{colorama.Style.RESET_ALL} - - Commands: - publish - (re-)publishes ads - verify - verifies the configuration files - delete - deletes ads - update - updates published ads - extend - extends ads within the 8-day window before expiry - download - downloads one or multiple ads - update-check - checks for available updates - update-content-hash – recalculates each ad's content_hash based on the current ad_defaults; - use this after changing config.yaml/ad_defaults to avoid every ad being marked "changed" and republished - create-config - creates a new default configuration file if one does not exist - diagnose - diagnoses browser connection issues and shows troubleshooting information - -- - help - displays this help (default command) - version - displays the application version - - Options: - --ads=all|due|new|changed| (publish) - specifies which ads to (re-)publish (DEFAULT: due) - Possible values: - * all: (re-)publish all ads ignoring republication_interval - * due: publish all new ads and republish ads according the republication_interval - * new: only publish new ads (i.e. ads that have no id in the config file) - * changed: only publish ads that have been modified since last publication - * : provide one or several ads by ID to (re-)publish, like e.g. "--ads=1,2,3" ignoring republication_interval - * Combinations: You can combine multiple selectors with commas, e.g. "--ads=changed,due" to publish both changed and due ads - --ads=all|new| (download) - specifies which ads to download (DEFAULT: new) - Possible values: - * all: downloads all ads from your profile - * new: downloads ads from your profile that are not locally saved yet - * : provide one or several ads by ID to download, like e.g. "--ads=1,2,3" - --ads=all|changed| (update) - specifies which ads to update (DEFAULT: changed) - Possible values: - * all: update all ads - * changed: only update ads that have been modified since last publication - * : provide one or several ads by ID to update, like e.g. "--ads=1,2,3" - --ads=all| (extend) - specifies which ads to extend - Possible values: - * all: extend all ads expiring within 8 days - * : specify ad IDs to extend, e.g. "--ads=1,2,3" - --force - alias for '--ads=all' - --keep-old - don't delete old ads on republication - --config= - path to the config YAML or JSON file (does not implicitly change workspace mode) - --workspace-mode=portable|xdg - overrides workspace mode for this run - --logfile= - path to the logfile (DEFAULT: depends on active workspace mode) - --lang=en|de - display language (STANDARD: system language if supported, otherwise English) - -v, --verbose - enables verbose output - only useful when troubleshooting issues - """.rstrip() - ) - ) - def _is_valid_ads_selector(self, valid_keywords:set[str]) -> bool: """Check if the current ads_selector is valid for the given set of keyword selectors. @@ -530,184 +408,6 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 or _download_selection.is_numeric_ids_selector(self.ads_selector) ) - def parse_args(self, args:list[str]) -> None: - try: - options, arguments = getopt.gnu_getopt( - args[1:], - "hv", - ["ads=", "config=", "force", "help", "keep-old", "logfile=", "lang=", "verbose", "workspace-mode="], - ) - except getopt.error as ex: - LOG.error(ex.msg) - LOG.error("Use --help to display available options.") - sys.exit(2) - - for option, value in options: - match option: - case "-h" | "--help": - self.show_help() - sys.exit(0) - case "--config": - self.config_file_path = abspath(value) - self._config_arg = value - case "--logfile": - if value: - self.log_file_path = abspath(value) - else: - self.log_file_path = None - self._logfile_arg = value - self._logfile_explicitly_provided = True - case "--workspace-mode": - mode = value.strip().lower() - if mode not in {"portable", "xdg"}: - LOG.error("Invalid --workspace-mode '%s'. Use 'portable' or 'xdg'.", value) - sys.exit(2) - self._workspace_mode_arg = cast(_xdg_paths.InstallationMode, mode) - case "--ads": - self.ads_selector = value.strip().lower() - self._ads_selector_explicit = True - case "--force": - self.ads_selector = "all" - self._ads_selector_explicit = True - case "--keep-old": - self.keep_old_ads = True - case "--lang": - set_current_locale(Locale.of(value)) - case "-v" | "--verbose": - LOG.setLevel(_loggers.DEBUG) - _loggers.get_logger("nodriver").setLevel(_loggers.INFO) - - match len(arguments): - case 0: - self.command = "help" - case 1: - self.command = arguments[0] - case _: - LOG.error("More than one command given: %s", arguments) - sys.exit(2) - - def configure_file_logging(self) -> None: - if not self.log_file_path: - return - if self.file_log: - return - - if self.workspace and self.workspace.log_file: - _xdg_paths.ensure_directory(self.workspace.log_file.parent, "log directory") - - LOG.info("Logging to [%s]...", self.log_file_path) - self.file_log = _loggers.configure_file_logging(self.log_file_path) - - LOG.info("App version: %s", self.get_version()) - LOG.info("Python version: %s", sys.version) - - def create_default_config(self) -> None: - """ - Create a default config.yaml in the project root if it does not exist. - If it exists, log an error and inform the user. - """ - if os.path.exists(self.config_file_path): - LOG.error("Config file %s already exists. Aborting creation.", self.config_file_path) - return - config_parent = self.workspace.config_file.parent if self.workspace else Path(self.config_file_path).parent - _xdg_paths.ensure_directory(config_parent, "config directory") - default_config = Config.model_construct() - default_config.login.username = "changeme" # noqa: S105 placeholder for default config, not a real username - default_config.login.password = "changeme" # noqa: S105 placeholder for default config, not a real password - _dicts.save_commented_model( - self.config_file_path, - default_config, - header = "# yaml-language-server: $schema=https://raw.githubusercontent.com/Second-Hand-Friends/kleinanzeigen-bot/main/schemas/config.schema.json", - exclude = { - "ad_defaults": {"description"}, - }, - ) - - def load_config(self) -> None: - # write default config.yaml if config file does not exist - if not os.path.exists(self.config_file_path): - self.create_default_config() - - config_yaml = _dicts.load_dict_if_exists(self.config_file_path, _("config")) - if isinstance(config_yaml, dict): - self._resolve_login_credentials(config_yaml) - self.config = Config.model_validate(config_yaml, strict = True, context = self.config_file_path) - - timing_enabled = self.config.diagnostics.timing_collection - if timing_enabled and self.workspace: - timing_dir = self.workspace.diagnostics_dir.parent / "timing" - self._timing_collector = TimingCollector(timing_dir, self.command) - else: - self._timing_collector = None - - # load built-in category mappings - self.categories = _dicts.load_dict_from_module(_resources, "categories.yaml", "") - LOG.debug("Loaded %s categories from categories.yaml", len(self.categories)) - deprecated_categories = _dicts.load_dict_from_module(_resources, "categories_old.yaml", "") - LOG.debug("Loaded %s categories from categories_old.yaml", len(deprecated_categories)) - self.categories.update(deprecated_categories) - custom_count = 0 - if self.config.categories: - custom_count = len(self.config.categories) - self.categories.update(self.config.categories) - LOG.debug("Loaded %s categories from config.yaml (custom)", custom_count) - total_count = len(self.categories) - if total_count == 0: - LOG.warning("No categories loaded - category files may be missing or empty") - LOG.debug("Loaded %s categories in total", total_count) - - # populate browser_config object used by WebScrapingMixin - self.browser_config.arguments = self.config.browser.arguments - self.browser_config.binary_location = self.config.browser.binary_location - self.browser_config.extensions = [abspath(item, relative_to = self.config_file_path) for item in self.config.browser.extensions] - self.browser_config.use_private_window = self.config.browser.use_private_window - if self.config.browser.user_data_dir: - self.browser_config.user_data_dir = abspath(self.config.browser.user_data_dir, relative_to = self.config_file_path) - elif self.workspace: - self.browser_config.user_data_dir = str(self.workspace.browser_profile_dir) - self.browser_config.profile_name = self.config.browser.profile_name - - @staticmethod - def _resolve_login_credentials(config_yaml:dict[str, Any]) -> None: - """Resolve ``login.username`` and ``login.password`` from environment variables. - - Supports two patterns: - - ``${VAR}`` : required — raises if VAR is unset - - ``${VAR:-default}`` : optional — uses *default* when VAR is unset - - Non-placeholder values are left unchanged. - - Note: - Mutates *config_yaml* in place. Caller must ensure *config_yaml* is a - dict before calling — the method is a no-op on non-dict values. - """ - if not isinstance(config_yaml, dict): - return - - login = config_yaml.get("login") - if not isinstance(login, dict): - return - - for field in ("username", "password"): - value = login.get(field) - if not isinstance(value, str): - continue - - m = _LOGIN_ENV_PATTERN.match(value) - if not m: - continue - - var_name = m.group("var") - resolved = os.environ.get(var_name) - if resolved is not None: - login[field] = resolved - elif m.group("default") is not None: - login[field] = m.group("default") - else: - raise ValueError( - _("Environment variable %s is required for login.%s but is not set") % (var_name, field) - ) - def __check_ad_republication(self, ad_cfg:Ad, ad_file_relative:str) -> bool: """ Check if an ad needs to be republished based on republication interval. @@ -3320,40 +3020,8 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904 def main(args:list[str]) -> None: - if "version" not in args: - print( - textwrap.dedent(rf""" - _ _ _ _ _ _ - | | _| | ___(_)_ __ __ _ _ __ _______(_) __ _ ___ _ __ | |__ ___ | |_ - | |/ / |/ _ \ | '_ \ / _` | '_ \|_ / _ \ |/ _` |/ _ \ '_ \ ____| '_ \ / _ \| __| - | <| | __/ | | | | (_| | | | |/ / __/ | (_| | __/ | | |____| |_) | (_) | |_ - |_|\_\_|\___|_|_| |_|\__,_|_| |_/___\___|_|\__, |\___|_| |_| |_.__/ \___/ \__| - |___/ - https://github.com/Second-Hand-Friends/kleinanzeigen-bot - Version: {__version__} - """)[1:], - flush = True, - ) # [1:] removes the first empty blank line - - _loggers.configure_console_logging() - - signal.signal(signal.SIGINT, _error_handlers.on_sigint) # capture CTRL+C - - # sys.excepthook = _error_handlers.on_exception - # -> commented out because it causes PyInstaller to log "[PYI-28040:ERROR] Failed to execute script '__main__' due to unhandled exception!", - # despite the exceptions being properly processed by our custom _error_handlers.on_exception callback. - # We now handle exceptions explicitly using a top-level try/except block. - - atexit.register(_loggers.flush_all_handlers) - - try: - bot = KleinanzeigenBot() - atexit.register(bot.close_browser_session) - nodriver.loop().run_until_complete(bot.run(args)) # type: ignore[attr-defined] - except CaptchaEncountered as ex: - raise ex - except Exception: - _error_handlers.on_exception(*sys.exc_info()) + _cli = importlib.import_module(".cli", __name__) + _cli.main(args) if __name__ == "__main__": diff --git a/src/kleinanzeigen_bot/__main__.py b/src/kleinanzeigen_bot/__main__.py index de67206..c183b56 100644 --- a/src/kleinanzeigen_bot/__main__.py +++ b/src/kleinanzeigen_bot/__main__.py @@ -4,7 +4,7 @@ import sys, time # isort: skip from gettext import gettext as _ -import kleinanzeigen_bot +from kleinanzeigen_bot.cli import main from kleinanzeigen_bot.utils.exceptions import CaptchaEncountered from kleinanzeigen_bot.utils.launch_mode_guard import ensure_not_launched_from_windows_explorer from kleinanzeigen_bot.utils.misc import format_timedelta @@ -19,7 +19,7 @@ ensure_not_launched_from_windows_explorer() # --------------------------------------------------------------------------- # while True: try: - kleinanzeigen_bot.main(sys.argv) # runs & returns when finished + main(sys.argv) # runs & returns when finished sys.exit(0) # not using `break` to prevent process closing issues except CaptchaEncountered as ex: delay = ex.restart_delay diff --git a/src/kleinanzeigen_bot/cli.py b/src/kleinanzeigen_bot/cli.py new file mode 100644 index 0000000..f2cf496 --- /dev/null +++ b/src/kleinanzeigen_bot/cli.py @@ -0,0 +1,271 @@ +# SPDX-FileCopyrightText: © Jens Bergmann and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ +"""CLI bootstrap and argument parsing for kleinanzeigen-bot. + +Handles argument parsing via :func:`parse_args`, signal handling, +daemonizing or foreground execution, log and locale setup, and +dispatches to :class:`KleinanzeigenBot `. + +Primary entry point: :func:`main`. +""" +from __future__ import annotations + +import atexit +import getopt +import os +import signal +import sys +import textwrap +from dataclasses import dataclass +from typing import Final, Sequence + +import colorama +import nodriver + +from kleinanzeigen_bot import KleinanzeigenBot +from kleinanzeigen_bot._version import __version__ +from kleinanzeigen_bot.runtime_config import VALID_COMMANDS +from kleinanzeigen_bot.utils import error_handlers as _error_handlers +from kleinanzeigen_bot.utils import loggers as _loggers +from kleinanzeigen_bot.utils.exceptions import CaptchaEncountered +from kleinanzeigen_bot.utils.files import abspath +from kleinanzeigen_bot.utils.i18n import Locale, get_current_locale, set_current_locale +from kleinanzeigen_bot.utils.misc import is_frozen + +LOG:Final[_loggers.Logger] = _loggers.get_logger(__name__) +LOG.setLevel(_loggers.INFO) + + +@dataclass(slots = True) +class ParsedArgs: + command:str = "help" + ads_selector:str = "due" + ads_selector_explicit:bool = False + keep_old_ads:bool = False + config_arg:str | None = None + config_file_path:str | None = None + logfile_arg:str | None = None + log_file_path:str | None = None + logfile_explicitly_provided:bool = False + workspace_mode:str | None = None + + +def _help_executable() -> str: + if is_frozen(): + return sys.argv[0] + if os.getenv("PDM_PROJECT_ROOT", ""): + return "pdm run app" + return "python -m kleinanzeigen_bot" + + +def _help_text() -> str: + exe = _help_executable() + if get_current_locale().language == "de": + return textwrap.dedent( + f"""\ + Verwendung: {colorama.Fore.LIGHTMAGENTA_EX}{exe} BEFEHL [OPTIONEN]{colorama.Style.RESET_ALL} + + Befehle: + publish - (Wieder-)Veröffentlicht Anzeigen + verify - Überprüft die Konfigurationsdateien + delete - Löscht Anzeigen + update - Aktualisiert bestehende Anzeigen + extend - Verlängert Anzeigen innerhalb des 8-Tage-Zeitfensters + download - Lädt eine oder mehrere Anzeigen herunter + update-check - Prüft auf verfügbare Updates + update-content-hash - Berechnet den content_hash aller Anzeigen anhand der aktuellen ad_defaults neu; + nach Änderungen an den config.yaml/ad_defaults verhindert es, dass alle Anzeigen als + "geändert" gelten und neu veröffentlicht werden. + create-config - Erstellt eine neue Standard-Konfigurationsdatei, falls noch nicht vorhanden + diagnose - Diagnostiziert Browser-Verbindungsprobleme und zeigt Troubleshooting-Informationen + -- + help - Zeigt diese Hilfe an (Standardbefehl) + version - Zeigt die Version der Anwendung an + + Optionen: + --ads=all|due|new|changed| (publish) - Gibt an, welche Anzeigen (erneut) veröffentlicht werden sollen (STANDARD: due) + Mögliche Werte: + * all: Veröffentlicht alle Anzeigen erneut, ignoriert republication_interval + * due: Veröffentlicht alle neuen Anzeigen und erneut entsprechend dem republication_interval + * new: Veröffentlicht nur neue Anzeigen (d.h. Anzeigen ohne ID in der Konfigurationsdatei) + * changed: Veröffentlicht nur Anzeigen, die seit der letzten Veröffentlichung geändert wurden + * : Gibt eine oder mehrere Anzeigen-IDs an, die veröffentlicht werden sollen, z. B. "--ads=1,2,3", ignoriert republication_interval + * Kombinationen: Sie können mehrere Selektoren mit Kommas kombinieren, z. B. "--ads=changed,due" um sowohl geänderte als auch + fällige Anzeigen zu veröffentlichen + --ads=all|new| (download) - Gibt an, welche Anzeigen heruntergeladen werden sollen (STANDARD: new) + Mögliche Werte: + * all: Lädt alle Anzeigen aus Ihrem Profil herunter + * new: Lädt Anzeigen aus Ihrem Profil herunter, die lokal noch nicht gespeichert sind + * : Gibt eine oder mehrere Anzeigen-IDs zum Herunterladen an, z. B. "--ads=1,2,3" + --ads=all|changed| (update) - Gibt an, welche Anzeigen aktualisiert werden sollen (STANDARD: changed) + Mögliche Werte: + * all: Aktualisiert alle Anzeigen + * changed: Aktualisiert nur Anzeigen, die seit der letzten Veröffentlichung geändert wurden + * : Gibt eine oder mehrere Anzeigen-IDs zum Aktualisieren an, z. B. "--ads=1,2,3" + --ads=all| (extend) - Gibt an, welche Anzeigen verlängert werden sollen (STANDARD: all) + Mögliche Werte: + * all: Verlängert alle Anzeigen, die innerhalb von 8 Tagen ablaufen + * : Gibt bestimmte Anzeigen-IDs an, z. B. "--ads=1,2,3" + --force - Alias für '--ads=all' + --keep-old - Verhindert das Löschen alter Anzeigen bei erneuter Veröffentlichung + --config= - Pfad zur YAML- oder JSON-Konfigurationsdatei (ändert den Workspace-Modus nicht implizit) + --workspace-mode=portable|xdg - Überschreibt den Workspace-Modus für diesen Lauf + --logfile= - Pfad zur Protokolldatei (STANDARD: vom aktiven Workspace-Modus abhängig) + --lang=en|de - Anzeigesprache (STANDARD: Systemsprache, wenn unterstützt, sonst Englisch) + -v, --verbose - Aktiviert detaillierte Ausgabe – nur nützlich zur Fehlerbehebung + """.rstrip() + ) + + return textwrap.dedent( + f"""\ + Usage: {colorama.Fore.LIGHTMAGENTA_EX}{exe} COMMAND [OPTIONS]{colorama.Style.RESET_ALL} + + Commands: + publish - (re-)publishes ads + verify - verifies the configuration files + delete - deletes ads + update - updates published ads + extend - extends ads within the 8-day window before expiry + download - downloads one or multiple ads + update-check - checks for available updates + update-content-hash – recalculates each ad's content_hash based on the current ad_defaults; + use this after changing config.yaml/ad_defaults to avoid every ad being marked "changed" and republished + create-config - creates a new default configuration file if one does not exist + diagnose - diagnoses browser connection issues and shows troubleshooting information + -- + help - displays this help (default command) + version - displays the application version + + Options: + --ads=all|due|new|changed| (publish) - specifies which ads to (re-)publish (DEFAULT: due) + Possible values: + * all: (re-)publish all ads ignoring republication_interval + * due: publish all new ads and republish ads according the republication_interval + * new: only publish new ads (i.e. ads that have no id in the config file) + * changed: only publish ads that have been modified since last publication + * : provide one or several ads by ID to (re-)publish, like e.g. "--ads=1,2,3" ignoring republication_interval + * Combinations: You can combine multiple selectors with commas, e.g. "--ads=changed,due" to publish both changed and due ads + --ads=all|new| (download) - specifies which ads to download (DEFAULT: new) + Possible values: + * all: downloads all ads from your profile + * new: downloads ads from your profile that are not locally saved yet + * : provide one or several ads by ID to download, like e.g. "--ads=1,2,3" + --ads=all|changed| (update) - specifies which ads to update (DEFAULT: changed) + Possible values: + * all: update all ads + * changed: only update ads that have been modified since last publication + * : provide one or several ads by ID to update, like e.g. "--ads=1,2,3" + --ads=all| (extend) - specifies which ads to extend (DEFAULT: all) + Possible values: + * all: extend all ads expiring within 8 days + * : specify ad IDs to extend, e.g. "--ads=1,2,3" + --force - alias for '--ads=all' + --keep-old - don't delete old ads on republication + --config= - path to the config YAML or JSON file (does not implicitly change workspace mode) + --workspace-mode=portable|xdg - overrides workspace mode for this run + --logfile= - path to the logfile (DEFAULT: depends on active workspace mode) + --lang=en|de - display language (STANDARD: system language if supported, otherwise English) + -v, --verbose - enables verbose output - only useful when troubleshooting issues + """.rstrip() + ) + + +def show_help() -> None: + print(_help_text()) + + +def parse_args(args:Sequence[str]) -> ParsedArgs: + parsed = ParsedArgs() + help_requested = False + try: + options, arguments = getopt.gnu_getopt( + list(args)[1:], + "hv", + ["ads=", "config=", "force", "help", "keep-old", "logfile=", "lang=", "verbose", "workspace-mode="], + ) + except getopt.error as ex: + LOG.error(ex.msg) + LOG.error("Use --help to display available options.") + sys.exit(2) + + for option, value in options: + match option: + case "-h" | "--help": + help_requested = True + case "--config": + parsed.config_file_path = abspath(value) + parsed.config_arg = value + case "--logfile": + parsed.logfile_arg = value + parsed.logfile_explicitly_provided = True + parsed.log_file_path = abspath(value) if value else None + case "--workspace-mode": + mode = value.strip().lower() + if mode not in {"portable", "xdg"}: + LOG.error("Invalid --workspace-mode '%s'. Use 'portable' or 'xdg'.", value) + sys.exit(2) + parsed.workspace_mode = mode + case "--ads": + parsed.ads_selector = value.strip().lower() + parsed.ads_selector_explicit = True + case "--force": + parsed.ads_selector = "all" + parsed.ads_selector_explicit = True + case "--keep-old": + parsed.keep_old_ads = True + case "--lang": + set_current_locale(Locale.of(value)) + case "-v" | "--verbose": + LOG.setLevel(_loggers.DEBUG) + _loggers.get_logger("kleinanzeigen_bot").setLevel(_loggers.DEBUG) + _loggers.get_logger("kleinanzeigen_bot.runtime_config").setLevel(_loggers.DEBUG) + _loggers.get_logger("nodriver").setLevel(_loggers.INFO) + + if help_requested: + show_help() + sys.exit(0) + + match len(arguments): + case 0: + parsed.command = "help" + case 1: + parsed.command = arguments[0] + if parsed.command not in VALID_COMMANDS: + LOG.error("Unknown command: %s", parsed.command) + sys.exit(2) + case _: + LOG.error("More than one command given: %s", arguments) + sys.exit(2) + + return parsed + + +def main(args:Sequence[str]) -> None: + if "version" not in args: + print( + textwrap.dedent(rf""" + _ _ _ _ _ _ + | | _| | ___(_)_ __ __ _ _ __ _______(_) __ _ ___ _ __ | |__ ___ | |_ + | |/ / |/ _ \ | '_ \ / _` | '_ \|_ / _ \ |/ _` |/ _ \ '_ \ ____| '_ \ / _ \| __| + | <| | __/ | | | | (_| | | | |/ / __/ | (_| | __/ | | |____| |_) | (_) | |_ + |_|\_\_|\___|_|_| |_|\__,_|_| |_/___\___|_|\__, |\___|_| |_| |_.__/ \___/ \__| + |___/ + https://github.com/Second-Hand-Friends/kleinanzeigen-bot + Version: {__version__} + """)[1:], + flush = True, + ) # [1:] removes the first empty blank line + + _loggers.configure_console_logging() + signal.signal(signal.SIGINT, _error_handlers.on_sigint) + atexit.register(_loggers.flush_all_handlers) + + try: + bot = KleinanzeigenBot() + nodriver.loop().run_until_complete(bot.run(list(args))) # type: ignore[attr-defined] + except CaptchaEncountered: + raise + except Exception: + _error_handlers.on_exception(*sys.exc_info()) + sys.exit(1) diff --git a/src/kleinanzeigen_bot/resources/translations.de.yaml b/src/kleinanzeigen_bot/resources/translations.de.yaml index 6b59bfd..f2bbd78 100644 --- a/src/kleinanzeigen_bot/resources/translations.de.yaml +++ b/src/kleinanzeigen_bot/resources/translations.de.yaml @@ -19,19 +19,44 @@ kleinanzeigen_bot/__main__.py: "[INFO] Captcha detected. Sleeping %s before restart...": "[INFO] Captcha erkannt. Warte %s h bis zum Neustart..." ################################################# -kleinanzeigen_bot/__init__.py: +kleinanzeigen_bot/cli.py: +################################################# + parse_args: + "Use --help to display available options.": "Mit --help können die verfügbaren Optionen angezeigt werden." + "Unknown command: %s": "Unbekannter Befehl: %s" + "More than one command given: %s": "Mehr als ein Befehl angegeben: %s" + "Invalid --workspace-mode '%s'. Use 'portable' or 'xdg'.": "Ungültiger --workspace-mode '%s'. Verwenden Sie 'portable' oder 'xdg'." + +################################################# +kleinanzeigen_bot/runtime_config.py: ################################################# - module: - "Direct execution not supported. Use 'pdm run app'": "Direkte Ausführung nicht unterstützt. Bitte 'pdm run app' verwenden" create_default_config: "Config file %s already exists. Aborting creation.": "Konfigurationsdatei %s existiert bereits. Erstellung abgebrochen." - _workspace_or_raise: - "Workspace must be resolved before command execution": "Arbeitsbereich muss vor der Befehlsausführung aufgelöst werden" - configure_file_logging: "Logging to [%s]...": "Protokollierung in [%s]..." "App version: %s": "App Version: %s" "Python version: %s": "Python Version: %s" + load_config: + "config": "Konfiguration" + "Loaded %s categories from categories.yaml": "%s Kategorien aus categories.yaml geladen" + "Loaded %s categories from categories_old.yaml": "%s Kategorien aus categories_old.yaml geladen" + "Loaded %s categories from config.yaml (custom)": "%s Kategorien aus config.yaml geladen (benutzerdefiniert)" + "Loaded %s categories in total": "%s Kategorien insgesamt geladen" + "No categories loaded - category files may be missing or empty": "Keine Kategorien geladen - Kategorie-Dateien fehlen oder sind leer" + _resolve_login_credentials: + "Environment variable %s is required for login.%s but is not set": "Umgebungsvariable %s ist für login.%s erforderlich, ist aber nicht gesetzt" + resolve_workspace: + "Config: %s": "Konfiguration: %s" + "Workspace mode: %s": "Arbeitsmodus: %s" + "Workspace: %s": "Arbeitsverzeichnis: %s" + +################################################# +kleinanzeigen_bot/__init__.py: +################################################# + module: + "Direct execution not supported. Use 'pdm run app'": "Direkte Ausführung nicht unterstützt. Bitte 'pdm run app' verwenden" + _workspace_or_raise: + "Workspace must be resolved before command execution": "Arbeitsbereich muss vor der Befehlsausführung aufgelöst werden" _fetch_published_ads: "Empty JSON response content on page %s": "Leerer JSON-Antwortinhalt auf Seite %s" @@ -68,17 +93,6 @@ kleinanzeigen_bot/__init__.py: "Loaded %s": "%s geladen" "ad": "Anzeige" - load_config: - "config": "Konfiguration" - "Loaded %s categories from categories.yaml": "%s Kategorien aus categories.yaml geladen" - "Loaded %s categories from categories_old.yaml": "%s Kategorien aus categories_old.yaml geladen" - "Loaded %s categories from config.yaml (custom)": "%s Kategorien aus config.yaml geladen (benutzerdefiniert)" - "Loaded %s categories in total": "%s Kategorien insgesamt geladen" - "No categories loaded - category files may be missing or empty": "Keine Kategorien geladen - Kategorie-Dateien fehlen oder sind leer" - - _resolve_login_credentials: - "Environment variable %s is required for login.%s but is not set": "Umgebungsvariable %s ist für login.%s erforderlich, ist aber nicht gesetzt" - check_and_wait_for_captcha: "# Captcha present! Please solve the captcha.": "# Captcha vorhanden! Bitte lösen Sie das Captcha." "Captcha recognized - auto-restart enabled, abort run...": "Captcha erkannt - Auto-Neustart aktiviert, Durchlauf wird beendet..." @@ -167,16 +181,6 @@ kleinanzeigen_bot/__init__.py: find_and_click_extend_button: "Found extend button on page %s": "'Verlängern'-Button auf Seite %s gefunden" - _resolve_workspace: - "Config: %s": "Konfiguration: %s" - "Workspace mode: %s": "Arbeitsmodus: %s" - "Workspace: %s": "Arbeitsverzeichnis: %s" - - parse_args: - "Use --help to display available options.": "Mit --help können die verfügbaren Optionen angezeigt werden." - "More than one command given: %s": "Mehr als ein Befehl angegeben: %s" - "Invalid --workspace-mode '%s'. Use 'portable' or 'xdg'.": "Ungültiger --workspace-mode '%s'. Verwenden Sie 'portable' oder 'xdg'." - publish_ads: "Processing %s/%s: '%s' from [%s]...": "Verarbeite %s/%s: '%s' von [%s]..." "Skipping because ad is reserved": "Überspringen, da Anzeige reserviert ist" diff --git a/src/kleinanzeigen_bot/runtime_config.py b/src/kleinanzeigen_bot/runtime_config.py new file mode 100644 index 0000000..1ae0096 --- /dev/null +++ b/src/kleinanzeigen_bot/runtime_config.py @@ -0,0 +1,235 @@ +# SPDX-FileCopyrightText: © Jens Bergmann and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ +"""Runtime configuration and bootstrap helpers. + +Provides config loading, default generation, workspace resolution, +browser config application, and file-logging setup. Central types: +:class:`RuntimeState` and :func:`load_config`. +""" +from __future__ import annotations + +import os +import re +import sys +from dataclasses import dataclass +from gettext import gettext as _ +from pathlib import Path +from typing import Any, Final + +from kleinanzeigen_bot import resources as _resources +from kleinanzeigen_bot.model.config_model import Config +from kleinanzeigen_bot.utils import dicts as _dicts +from kleinanzeigen_bot.utils import loggers as _loggers +from kleinanzeigen_bot.utils import xdg_paths as _xdg_paths +from kleinanzeigen_bot.utils.files import abspath +from kleinanzeigen_bot.utils.timing_collector import TimingCollector + +LOG:Final[_loggers.Logger] = _loggers.get_logger(__name__) +LOG.setLevel(_loggers.INFO) + +_LOGIN_ENV_PATTERN:Final[re.Pattern[str]] = re.compile(r"^\$\{(?P\w+)(?::-(?P.*))?\}$") + +# Commands that do not need workspace / filesystem state. +WORKSPACE_FREE_COMMANDS:Final[frozenset[str]] = frozenset({"help", "version", "create-config"}) +# All valid CLI commands. Keep in sync with the dispatch in __init__.py. +VALID_COMMANDS:Final[frozenset[str]] = frozenset({ + "help", "version", "create-config", "diagnose", "verify", + "update-check", "update-content-hash", + "publish", "update", "delete", "extend", "download", +}) + + +@dataclass(slots = True) +class RuntimeState: + config:Config + categories:dict[str, str] + timing_collector:TimingCollector | None + + +def create_default_config(config_file_path:str, workspace:_xdg_paths.Workspace | None) -> None: + if os.path.exists(config_file_path): + LOG.error("Config file %s already exists. Aborting creation.", config_file_path) + return + + config_parent = workspace.config_file.parent if workspace else Path(config_file_path).parent + _xdg_paths.ensure_directory(config_parent, "config directory") + default_config = Config.model_construct() + default_config.login.username = "changeme" # noqa: S105 placeholder for default config, not a real username + default_config.login.password = "changeme" # noqa: S105 placeholder for default config, not a real password + _dicts.save_commented_model( + config_file_path, + default_config, + header = "# yaml-language-server: $schema=https://raw.githubusercontent.com/Second-Hand-Friends/kleinanzeigen-bot/main/schemas/config.schema.json", + exclude = { + "ad_defaults": {"description"}, + }, + ) + + +def _resolve_login_credentials(config_yaml:dict[str, Any]) -> None: + if not isinstance(config_yaml, dict): + return + + login = config_yaml.get("login") + if not isinstance(login, dict): + return + + for field in ("username", "password"): + value = login.get(field) + if not isinstance(value, str): + continue + + match = _LOGIN_ENV_PATTERN.match(value) + if not match: + continue + + var_name = match.group("var") + resolved = os.environ.get(var_name) + if resolved is not None: + login[field] = resolved + elif match.group("default") is not None: + login[field] = match.group("default") + else: + raise ValueError(_("Environment variable %s is required for login.%s but is not set") % (var_name, field)) + + +def load_config(config_file_path:str, workspace:_xdg_paths.Workspace | None, command:str) -> RuntimeState: + """Load the runtime config and derived lookup tables. + + Args: + config_file_path: Path to the active config file. + workspace: Resolved workspace, if one exists for this command. + command: Active CLI command, used for timing collection labels. + + Returns: + RuntimeState: Parsed config, merged categories, and optional timing collector. + + Example: + `load_config("config.yaml", workspace, "verify")` returns a RuntimeState whose + `config`, `categories`, and `timing_collector` fields are ready for the command run. + """ + if not os.path.exists(config_file_path): + # Keep bootstrapping self-contained: first run must create a usable config file. + create_default_config(config_file_path, workspace) + + config_yaml = _dicts.load_dict_if_exists(config_file_path, _("config")) + if isinstance(config_yaml, dict): + # Resolve ${ENV} placeholders before schema validation so the model sees final values. + _resolve_login_credentials(config_yaml) + # Validate strictly and keep the file path in context so model errors point at the source file. + config = Config.model_validate(config_yaml, strict = True, context = config_file_path) + + timing_enabled = config.diagnostics.timing_collection + if timing_enabled and workspace: + # Diagnostics live under the workspace diagnostics tree; timing data sits next to it. + timing_dir = workspace.diagnostics_dir.parent / "timing" + timing_collector:TimingCollector | None = TimingCollector(timing_dir, command) + else: + # No workspace or disabled timing collection means we skip the collector entirely. + timing_collector = None + + # Merge order matters: bundled defaults first, deprecated aliases second, user overrides last. + categories:dict[str, str] = _dicts.load_dict_from_module(_resources, "categories.yaml", "") + LOG.debug("Loaded %s categories from categories.yaml", len(categories)) + deprecated_categories = _dicts.load_dict_from_module(_resources, "categories_old.yaml", "") + LOG.debug("Loaded %s categories from categories_old.yaml", len(deprecated_categories)) + categories.update(deprecated_categories) + if config.categories: + categories.update(config.categories) + LOG.debug("Loaded %s categories from config.yaml (custom)", len(config.categories)) + if not categories: + # This should only happen if resources are broken or empty, so surface it loudly. + LOG.warning("No categories loaded - category files may be missing or empty") + LOG.debug("Loaded %s categories in total", len(categories)) + + return RuntimeState(config = config, categories = categories, timing_collector = timing_collector) + + +def apply_browser_config(browser_config:Any, config:Config, workspace:_xdg_paths.Workspace | None, config_file_path:str) -> None: + browser_config.arguments = config.browser.arguments + browser_config.binary_location = config.browser.binary_location + browser_config.extensions = [abspath(item, relative_to = config_file_path) for item in config.browser.extensions] + browser_config.use_private_window = config.browser.use_private_window + if config.browser.user_data_dir: + browser_config.user_data_dir = abspath(config.browser.user_data_dir, relative_to = config_file_path) + elif workspace: + browser_config.user_data_dir = str(workspace.browser_profile_dir) + browser_config.profile_name = config.browser.profile_name + + +def configure_file_logging( + log_file_path:str | None, + workspace:_xdg_paths.Workspace | None, + file_log:_loggers.LogFileHandle | None, + version:str, +) -> _loggers.LogFileHandle | None: + if not log_file_path or file_log: + return file_log + + if workspace and workspace.log_file: + _xdg_paths.ensure_directory(workspace.log_file.parent, "log directory") + + LOG.info("Logging to [%s]...", log_file_path) + file_log = _loggers.configure_file_logging(log_file_path) + LOG.info("App version: %s", version) + LOG.info("Python version: %s", sys.version) + return file_log + + +def resolve_workspace( + *, + command:str, + config_file_path:str, + config_arg:str | None, + logfile_arg:str | None, + workspace_mode:_xdg_paths.InstallationMode | None, + logfile_explicitly_provided:bool, + log_basename:str, +) -> _xdg_paths.Workspace | None: + """Resolve the workspace for a command that needs filesystem state. + + Typical input: `command="verify"`, `config_file_path="config.yaml"`. + Typical output: a workspace rooted under the configured mode, or `None` for help/version. + """ + if command in WORKSPACE_FREE_COMMANDS: + return None + + effective_config_arg = config_arg + effective_workspace_mode = workspace_mode + if not effective_config_arg: + # Programmatic callers sometimes set config_file_path directly, so infer the config arg from it. + default_config = (Path.cwd() / "config.yaml").resolve() + if Path(config_file_path).resolve() != default_config: + effective_config_arg = config_file_path + if effective_workspace_mode is None: + # Preserve the old default: infer portable vs xdg from the config location. + config_path = Path(config_file_path).resolve() + xdg_config_dir = _xdg_paths.get_xdg_base_dir("config").resolve() + effective_workspace_mode = "xdg" if config_path.is_relative_to(xdg_config_dir) else "portable" + + try: + workspace = _xdg_paths.resolve_workspace( + config_arg = effective_config_arg, + logfile_arg = logfile_arg, + workspace_mode = effective_workspace_mode, + logfile_explicitly_provided = logfile_explicitly_provided, + log_basename = log_basename, + ) + except ValueError as exc: + LOG.error(str(exc)) + sys.exit(2) + + _xdg_paths.ensure_directory(workspace.config_file.parent, "config directory") + + LOG.info("Config: %s", workspace.config_file) + LOG.info("Workspace mode: %s", workspace.mode) + LOG.info("Workspace: %s", workspace.config_dir) + if _loggers.is_debug(LOG): + LOG.debug("Log file: %s", workspace.log_file) + LOG.debug("State dir: %s", workspace.state_dir) + LOG.debug("Download dir: %s", workspace.download_dir) + LOG.debug("Browser profile: %s", workspace.browser_profile_dir) + LOG.debug("Diagnostics dir: %s", workspace.diagnostics_dir) + + return workspace diff --git a/tests/smoke/test_smoke_health.py b/tests/smoke/test_smoke_health.py index 982617c..d4175b4 100644 --- a/tests/smoke/test_smoke_health.py +++ b/tests/smoke/test_smoke_health.py @@ -77,7 +77,7 @@ def invoke_cli( os.chdir(os.fspath(cwd)) logging.getLogger().addHandler(log_handler) with contextlib.ExitStack() as stack: - stack.enter_context(patch("kleinanzeigen_bot.atexit.register", capture_register)) + stack.enter_context(patch("atexit.register", capture_register)) stack.enter_context(contextlib.redirect_stdout(stdout)) stack.enter_context(contextlib.redirect_stderr(stderr)) effective_env_overrides = env_overrides if env_overrides is not None else _default_smoke_env(cwd) @@ -102,10 +102,16 @@ def _xdg_env_overrides(base_path:Path) -> dict[str, str]: xdg_config = base_path / "xdg" / "config" xdg_state = base_path / "xdg" / "state" xdg_cache = base_path / "xdg" / "cache" - for path in (home, xdg_config, xdg_state, xdg_cache): + appdata_roaming = base_path / "appdata" / "roaming" + appdata_local = base_path / "appdata" / "local" + for path in (home, xdg_config, xdg_state, xdg_cache, appdata_roaming, appdata_local): path.mkdir(parents = True, exist_ok = True) return { "HOME": os.fspath(home), + "USERPROFILE": os.fspath(home), + "APPDATA": os.fspath(appdata_roaming), + "LOCALAPPDATA": os.fspath(appdata_local), + "WIN_PD_OVERRIDE_LOCAL_APPDATA": os.fspath(appdata_local), "XDG_CONFIG_HOME": os.fspath(xdg_config), "XDG_STATE_HOME": os.fspath(xdg_state), "XDG_CACHE_HOME": os.fspath(xdg_cache), @@ -119,6 +125,13 @@ def _default_smoke_env(cwd:Path | None) -> dict[str, str] | None: return _xdg_env_overrides(cwd) +def _expected_xdg_config_file(env_overrides:Mapping[str, str]) -> Path: + """Derive the config.yaml path used by xdg mode for the active platform.""" + if os.name == "nt": + return Path(env_overrides["LOCALAPPDATA"]) / "kleinanzeigen-bot" / "kleinanzeigen-bot" / "config.yaml" + return Path(env_overrides["XDG_CONFIG_HOME"]) / "kleinanzeigen-bot" / "config.yaml" + + @pytest.fixture(autouse = True) def disable_update_checker(monkeypatch:pytest.MonkeyPatch) -> None: """Prevent smoke tests from hitting GitHub for update checks.""" @@ -177,6 +190,29 @@ def test_cli_subcommands_create_config_creates_file(tmp_path:Path) -> None: assert "config.yaml" in out, f"Expected config.yaml in CLI output.\n{out}" +@pytest.mark.smoke +@pytest.mark.parametrize("workspace_mode", ["portable", "xdg"]) +def test_cli_subcommands_create_config_honors_workspace_mode( + workspace_mode:str, + tmp_path:Path, +) -> None: + """ + Smoke: CLI 'create-config' writes config.yaml into the selected workspace mode. + """ + env_overrides = _default_smoke_env(tmp_path) + if workspace_mode == "portable": + config_file = tmp_path / "config.yaml" + else: + assert env_overrides is not None + config_file = _expected_xdg_config_file(env_overrides) + + result = invoke_cli(["create-config", "--workspace-mode", workspace_mode], cwd = tmp_path, env_overrides = env_overrides) + assert result.returncode == 0 + assert config_file.exists(), f"config.yaml was not created at {config_file}" + out = (result.stdout + "\n" + result.stderr).lower() + assert str(config_file).lower() in out, f"Expected config path in CLI output.\n{out}" + + @pytest.mark.smoke def test_cli_subcommands_create_config_fails_if_exists(tmp_path:Path) -> None: """ diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py new file mode 100644 index 0000000..c9fb8c0 --- /dev/null +++ b/tests/unit/test_cli.py @@ -0,0 +1,194 @@ +# SPDX-FileCopyrightText: © Jens Bergmann and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ +"""Unit tests for CLI parsing, help text rendering, and main runtime flow.""" + +from __future__ import annotations + +import runpy +import sys +from datetime import timedelta +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import pytest + +from kleinanzeigen_bot import cli, runtime_config +from kleinanzeigen_bot.utils import i18n, loggers +from kleinanzeigen_bot.utils.exceptions import CaptchaEncountered + +pytestmark = pytest.mark.unit + +if TYPE_CHECKING: + from pathlib import Path + + +class TestCliParseArgs: + @pytest.mark.parametrize( + ("args", "expected_command", "expected_selector", "expected_keep_old"), + [ + (["script.py", "publish", "--ads=all"], "publish", "all", False), + (["script.py", "--force", "publish"], "publish", "all", False), + (["script.py", "--keep-old", "publish"], "publish", "due", True), + (["script.py", "download", "--ads=123,456"], "download", "123,456", False), + ], + ) + def test_parses_command_and_selector(self, args:list[str], expected_command:str, expected_selector:str, expected_keep_old:bool) -> None: + parsed = cli.parse_args(args) + + assert parsed.command == expected_command + assert parsed.ads_selector == expected_selector + assert parsed.keep_old_ads is expected_keep_old + + def test_parses_config_and_logfile_paths(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + log_path = tmp_path / "bot.log" + + parsed = cli.parse_args(["script.py", "--config", str(config_path), "--logfile", str(log_path), "help"]) + + assert parsed.config_arg == str(config_path) + assert parsed.config_file_path == str(config_path.resolve()) + assert parsed.logfile_explicitly_provided is True + assert parsed.logfile_arg == str(log_path) + assert parsed.log_file_path == str(log_path.resolve()) + assert parsed.command == "help" + + def test_verbose_flag_enables_debug_logging(self, monkeypatch:pytest.MonkeyPatch) -> None: + package_logger = loggers.get_logger("kleinanzeigen_bot") + runtime_logger = loggers.get_logger(runtime_config.__name__) + monkeypatch.setattr(package_logger, "level", loggers.INFO) + monkeypatch.setattr(runtime_logger, "level", loggers.INFO) + + cli.parse_args(["script.py", "-v", "help"]) + + assert loggers.is_debug(package_logger) + assert loggers.is_debug(runtime_logger) + + def test_help_prints_and_exits(self, capsys:pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as exc_info: + cli.parse_args(["script.py", "--help"]) + + assert exc_info.value.code == 0 + stdout = capsys.readouterr().out + assert "Usage:" in stdout + assert "Commands:" in stdout + + def test_help_respects_language_flag_after_help(self, capsys:pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as exc_info: + cli.parse_args(["script.py", "--help", "--lang=de"]) + + assert exc_info.value.code == 0 + stdout = capsys.readouterr().out + assert "Verwendung:" in stdout + + def test_invalid_workspace_mode_exits(self) -> None: + with pytest.raises(SystemExit) as exc_info: + cli.parse_args(["script.py", "--workspace-mode=invalid", "help"]) + + assert exc_info.value.code == 2 + + def test_lang_option_updates_locale(self) -> None: + cli.parse_args(["script.py", "--lang=en", "help"]) + + assert i18n.get_current_locale().language == "en" + + def test_defaults_to_help_without_command(self) -> None: + parsed = cli.parse_args(["script.py"]) + + assert parsed.command == "help" + + def test_invalid_option_exits(self) -> None: + with pytest.raises(SystemExit) as exc_info: + cli.parse_args(["script.py", "--bogus"]) + + assert exc_info.value.code == 2 + + def test_more_than_one_command_exits(self) -> None: + with pytest.raises(SystemExit) as exc_info: + cli.parse_args(["script.py", "verify", "publish"]) + + assert exc_info.value.code == 2 + + +class TestCliHelpText: + def test_show_help_uses_german_text(self, capsys:pytest.CaptureFixture[str], monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.setattr(cli, "get_current_locale", lambda: SimpleNamespace(language = "de")) + + cli.show_help() + + stdout = capsys.readouterr().out + assert "Verwendung:" in stdout + assert "Befehle:" in stdout + + def test_help_executable_uses_frozen_executable(self, monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.setattr(cli, "is_frozen", lambda: True) + monkeypatch.setattr(sys, "argv", ["/usr/local/bin/kleinanzeigen-bot"]) + + assert cli._help_executable() == "/usr/local/bin/kleinanzeigen-bot" # noqa: SLF001 + + def test_help_executable_uses_pdm_when_project_root_is_set(self, monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.setattr(cli, "is_frozen", lambda: False) + monkeypatch.setenv("PDM_PROJECT_ROOT", "/project") + + assert cli._help_executable() == "pdm run app" # noqa: SLF001 + + def test_help_executable_uses_module_invocation_by_default(self, monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.setattr(cli, "is_frozen", lambda: False) + monkeypatch.delenv("PDM_PROJECT_ROOT", raising = False) + + assert cli._help_executable() == "python -m kleinanzeigen_bot" # noqa: SLF001 + + +class TestCliMain: + @staticmethod + def _fake_bot() -> SimpleNamespace: + return SimpleNamespace(close_browser_session = lambda: None, run = lambda args: object()) + + def test_main_forwards_unhandled_exceptions_to_error_handler(self, monkeypatch:pytest.MonkeyPatch) -> None: + handled:dict[str, object] = {} + + class FailingLoop: + def run_until_complete(self, _coro:object) -> None: + raise RuntimeError("boom") + + def handle_exception(exc_type:object, exc:object, _traceback:object) -> None: + handled.update(exc_type = exc_type, exc = exc) + + monkeypatch.setattr("kleinanzeigen_bot.cli.atexit.register", lambda *args, **kwargs: None) + monkeypatch.setattr("kleinanzeigen_bot.cli.nodriver.loop", FailingLoop) + monkeypatch.setattr(cli, "KleinanzeigenBot", self._fake_bot) + monkeypatch.setattr("kleinanzeigen_bot.cli._error_handlers.on_exception", handle_exception) + + with pytest.raises(SystemExit) as exc_info: + cli.main(["script.py", "version"]) + + assert handled["exc_type"] is RuntimeError + assert str(handled["exc"]) == "boom" + assert exc_info.value.code == 1 + + def test_main_reraises_captcha(self, monkeypatch:pytest.MonkeyPatch) -> None: + captcha = CaptchaEncountered(timedelta(seconds = 1)) + + class CaptchaLoop: + def run_until_complete(self, _coro:object) -> None: + raise captcha + + monkeypatch.setattr("kleinanzeigen_bot.cli.atexit.register", lambda *args, **kwargs: None) + monkeypatch.setattr("kleinanzeigen_bot.cli.nodriver.loop", CaptchaLoop) + monkeypatch.setattr(cli, "KleinanzeigenBot", self._fake_bot) + + with pytest.raises(CaptchaEncountered) as exc_info: + cli.main(["script.py", "version"]) + + assert exc_info.value is captcha + + def test_module_entrypoint_calls_cli_main_and_exits(self, monkeypatch:pytest.MonkeyPatch) -> None: + calls:list[list[str]] = [] + monkeypatch.setattr(sys, "argv", ["kleinanzeigen-bot", "version"]) + monkeypatch.setattr(cli, "main", lambda args: calls.append(list(args))) + + with pytest.raises(SystemExit) as exc_info: + runpy.run_module("kleinanzeigen_bot.__main__", run_name = "__main__") + + assert exc_info.value.code == 0 + assert calls == [["kleinanzeigen-bot", "version"]] diff --git a/tests/unit/test_extend_command.py b/tests/unit/test_extend_command.py index 6246ba6..28ae124 100644 --- a/tests/unit/test_extend_command.py +++ b/tests/unit/test_extend_command.py @@ -9,9 +9,9 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from kleinanzeigen_bot import KleinanzeigenBot +from kleinanzeigen_bot import KleinanzeigenBot, runtime_config from kleinanzeigen_bot.model.ad_model import Ad -from kleinanzeigen_bot.utils import dicts, misc +from kleinanzeigen_bot.utils import dicts, misc, xdg_paths from kleinanzeigen_bot.utils.web_scraping_mixin import By, Element @@ -46,7 +46,18 @@ class TestExtendCommand: async def test_run_extend_command_no_ads(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Test running extend command with no ads.""" test_bot.config_file_path = str(tmp_path / "config.yaml") - with patch.object(test_bot, "load_config"), patch.object(test_bot, "load_ads", return_value = []), patch("kleinanzeigen_bot.UpdateChecker"): + workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot") + with ( + patch("kleinanzeigen_bot.runtime_config.resolve_workspace", return_value = workspace), + patch( + "kleinanzeigen_bot.runtime_config.load_config", + return_value = runtime_config.RuntimeState(config = test_bot.config, categories = {}, timing_collector = None), + ), + patch("kleinanzeigen_bot.runtime_config.configure_file_logging", return_value = None), + patch("kleinanzeigen_bot.runtime_config.apply_browser_config"), + patch.object(test_bot, "load_ads", return_value = []), + patch("kleinanzeigen_bot.UpdateChecker"), + ): await test_bot.run(["script.py", "extend"]) assert test_bot.command == "extend" assert test_bot.ads_selector == "all" @@ -55,8 +66,15 @@ class TestExtendCommand: async def test_run_extend_command_with_specific_ids(self, test_bot:KleinanzeigenBot, tmp_path:Path) -> None: """Test running extend command with specific ad IDs.""" test_bot.config_file_path = str(tmp_path / "config.yaml") + workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot") with ( - patch.object(test_bot, "load_config"), + patch("kleinanzeigen_bot.runtime_config.resolve_workspace", return_value = workspace), + patch( + "kleinanzeigen_bot.runtime_config.load_config", + return_value = runtime_config.RuntimeState(config = test_bot.config, categories = {}, timing_collector = None), + ), + patch("kleinanzeigen_bot.runtime_config.configure_file_logging", return_value = None), + patch("kleinanzeigen_bot.runtime_config.apply_browser_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), @@ -203,29 +221,6 @@ class TestExtendAdsMethod: # Verify extend_ad was called mock_extend_ad.assert_called_once() - @pytest.mark.asyncio - async def test_extend_ads_no_eligible_ads(self, test_bot:KleinanzeigenBot, base_ad_config_with_id:dict[str, Any]) -> None: - """Test extend_ads when no ads are eligible for extension.""" - ad_cfg = Ad.model_validate(base_ad_config_with_id) - - # Set end date to 30 days from now (outside window) - future_date = misc.now() + timedelta(days = 30) - end_date_str = future_date.strftime("%d.%m.%Y") - - published_ads_json = {"ads": [{"id": 12345, "title": "Test Ad Title", "state": "active", "endDate": end_date_str}]} - - with ( - patch.object(test_bot, "web_request", new_callable = AsyncMock) as mock_request, - patch.object(test_bot, "web_sleep", new_callable = AsyncMock), - patch.object(test_bot, "extend_ad", new_callable = AsyncMock) as mock_extend_ad, - ): - mock_request.return_value = {"content": json.dumps(published_ads_json)} - - await test_bot.extend_ads([("test.yaml", ad_cfg, base_ad_config_with_id)]) - - # Verify extend_ad was not called - mock_extend_ad.assert_not_called() - @pytest.mark.asyncio async def test_extend_ads_handles_multiple_ads(self, test_bot:KleinanzeigenBot, base_ad_config_with_id:dict[str, Any]) -> None: """Test that extend_ads processes multiple ads correctly.""" @@ -360,34 +355,6 @@ class TestExtendAdMethod: with pytest.raises(Exception, match = "Unexpected error"): await test_bot.extend_ad(str(ad_file), ad_cfg, base_ad_config_with_id) - @pytest.mark.asyncio - async def test_extend_ad_updates_yaml_file(self, test_bot:KleinanzeigenBot, base_ad_config_with_id:dict[str, Any], tmp_path:Path) -> None: - """Test that extend_ad correctly updates the YAML file with new timestamp.""" - ad_cfg = Ad.model_validate(base_ad_config_with_id) - - # Create temporary YAML file - ad_file = tmp_path / "test_ad.yaml" - original_updated_on = base_ad_config_with_id["updated_on"] - dicts.save_dict(str(ad_file), base_ad_config_with_id) - - with ( - patch.object(test_bot, "_navigate_paginated_ad_overview", new_callable = AsyncMock) as mock_paginate, - patch.object(test_bot, "web_click", new_callable = AsyncMock), - patch("kleinanzeigen_bot.utils.misc.now") as mock_now, - ): - # Test mock datetime - timezone not relevant for timestamp formatting test - mock_now.return_value = datetime(2025, 1, 28, 14, 30, 0) # noqa: DTZ001 - - # Pagination succeeds (button found and clicked) - mock_paginate.return_value = True - - await test_bot.extend_ad(str(ad_file), ad_cfg, base_ad_config_with_id) - - # Load the updated file and verify the timestamp changed - updated_config = dicts.load_dict(str(ad_file)) - assert updated_config["updated_on"] != original_updated_on - assert updated_config["updated_on"] == "2025-01-28T14:30:00" - @pytest.mark.asyncio async def test_extend_ad_with_web_mocks(self, test_bot:KleinanzeigenBot, base_ad_config_with_id:dict[str, Any], tmp_path:Path) -> None: """Test extend_ad with web-level mocks to exercise the find_and_click_extend_button callback.""" diff --git a/tests/unit/test_init.py b/tests/unit/test_init.py index 0a02fe2..f207066 100644 --- a/tests/unit/test_init.py +++ b/tests/unit/test_init.py @@ -1,9 +1,9 @@ # 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, gc, io, json, logging, os, tempfile # isort: skip +import asyncio, copy, fnmatch, json, logging, os, tempfile # isort: skip from collections.abc import Callable, Generator -from contextlib import ExitStack, contextmanager, redirect_stdout +from contextlib import ExitStack, contextmanager from datetime import timedelta from pathlib import Path, PureWindowsPath from typing import Any, Awaitable, Iterator, cast @@ -19,6 +19,7 @@ from kleinanzeigen_bot import ( KleinanzeigenBot, LoginDetectionReason, LoginDetectionResult, + runtime_config, ) from kleinanzeigen_bot._version import __version__ from kleinanzeigen_bot.model.ad_model import Ad, AdUpdateStrategy @@ -29,7 +30,7 @@ from kleinanzeigen_bot.model.config_model import ( DiagnosticsConfig, PublishingConfig, ) -from kleinanzeigen_bot.utils import dicts, loggers, misc, xdg_paths +from kleinanzeigen_bot.utils import dicts, misc, xdg_paths from kleinanzeigen_bot.utils.exceptions import CategoryResolutionError, PublishedAdsFetchIncompleteError, PublishSubmissionUncertainError from kleinanzeigen_bot.utils.web_scraping_mixin import By, Element @@ -112,8 +113,15 @@ def mock_config_setup(test_bot:KleinanzeigenBot, tmp_path:Path) -> Generator[Non """Provide a centralized mock configuration setup for tests. This fixture mocks load_config and other essential configuration-related methods.""" test_bot.config_file_path = str(tmp_path / "config.yaml") + workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot") with ( - patch.object(test_bot, "load_config"), + patch("kleinanzeigen_bot.runtime_config.resolve_workspace", return_value = workspace), + patch( + "kleinanzeigen_bot.runtime_config.load_config", + return_value = runtime_config.RuntimeState(config = test_bot.config, categories = {}, timing_collector = None), + ), + patch("kleinanzeigen_bot.runtime_config.apply_browser_config"), + patch("kleinanzeigen_bot.runtime_config.configure_file_logging", return_value = None), 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, @@ -123,144 +131,18 @@ def mock_config_setup(test_bot:KleinanzeigenBot, tmp_path:Path) -> Generator[Non 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 - - def _login_detection_result(is_logged_in:bool, reason:LoginDetectionReason) -> LoginDetectionResult: return LoginDetectionResult(is_logged_in = is_logged_in, reason = reason) class TestKleinanzeigenBotInitialization: - """Tests for KleinanzeigenBot initialization and basic functionality.""" - - def test_constructor_initializes_default_values(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: - """Verify that constructor sets all default values correctly.""" - monkeypatch.chdir(tmp_path) - bot = KleinanzeigenBot() - assert bot.root_url == "https://www.kleinanzeigen.de" - assert bot.command == "help" - assert bot.ads_selector == "due" - assert bot.keep_old_ads is False - assert bot.log_file_path is not None - assert bot.file_log is None - - @pytest.mark.parametrize("command", ["help", "create-config", "version"]) - def test_resolve_workspace_skips_non_workspace_commands(self, test_bot:KleinanzeigenBot, command:str) -> None: - """Workspace resolution should remain None for commands that need no workspace.""" - test_bot.command = command - 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) -> None: - """Workspace resolution errors should terminate with code 2.""" - test_bot.command = "verify" - - with ( - patch("kleinanzeigen_bot.utils.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 - - 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.utils.xdg_paths.resolve_workspace", return_value = workspace), - patch("kleinanzeigen_bot.utils.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.utils.xdg_paths.get_xdg_base_dir", side_effect = lambda category: xdg_dirs[category]), - patch("kleinanzeigen_bot.utils.xdg_paths.resolve_workspace", side_effect = _make_fake_resolve_workspace(captured_mode, workspace)), - patch("kleinanzeigen_bot.utils.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.utils.xdg_paths.get_xdg_base_dir", side_effect = lambda category: xdg_dirs[category]), - patch("kleinanzeigen_bot.utils.xdg_paths.resolve_workspace", side_effect = _make_fake_resolve_workspace(captured_mode, workspace)), - patch("kleinanzeigen_bot.utils.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() - + """Tests for run-path initialization and download plumbing.""" @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]] = [] + workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot") class DummyUpdateChecker: def __init__(self, config:Config, state_file:Path) -> None: @@ -269,18 +151,19 @@ class TestKleinanzeigenBotInitialization: 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("kleinanzeigen_bot.runtime_config.resolve_workspace", return_value = workspace), + patch( + "kleinanzeigen_bot.runtime_config.load_config", + return_value = runtime_config.RuntimeState(config = test_bot.config, categories = {}, timing_collector = None), + ), + patch("kleinanzeigen_bot.runtime_config.configure_file_logging", return_value = None), + patch("kleinanzeigen_bot.runtime_config.apply_browser_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]) @@ -754,289 +637,6 @@ class TestKleinanzeigenBotInitialization: # All non-"active" states should result in active=False extractor_mock.download_ad.assert_awaited_once_with(123, active = False) - def test_create_default_config_preserves_existing_file(self, tmp_path:Path, test_bot:KleinanzeigenBot) -> None: - """Test that create_default_config does not overwrite an existing config file.""" - config_path = tmp_path / "config.yaml" - original_content = "dummy: value" - config_path.write_text(original_content) - test_bot.config_file_path = str(config_path) - test_bot.create_default_config() - assert config_path.read_text() == original_content - - def test_create_default_config_creates_file(self, tmp_path:Path, test_bot:KleinanzeigenBot) -> None: - """Test that create_default_config creates a config file if it does not exist.""" - config_path = tmp_path / "config.yaml" - test_bot.config_file_path = str(config_path) - assert not config_path.exists() - test_bot.create_default_config() - assert config_path.exists() - content = config_path.read_text() - assert "username: changeme" in content - assert "password: changeme" in content - - -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 - - def test_file_log_closed_after_bot_shutdown(self, tmp_path:Path) -> None: - """Ensure the file log handler is properly closed after the bot is deleted.""" - - # Directly instantiate the bot to control its lifecycle within the test - bot = KleinanzeigenBot() - log_path = tmp_path / "test.log" - bot.log_file_path = str(log_path) - - bot.configure_file_logging() - file_log = bot.file_log - assert file_log is not None - assert log_path.exists() - assert not file_log.is_closed() - - # Delete and garbage collect the bot instance to ensure the destructor (__del__) is called - del bot - gc.collect() - - assert file_log.is_closed() - - -class TestKleinanzeigenBotCommandLine: - """Tests for command line argument parsing.""" - - @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_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()) - - def test_parse_args_create_config(self, test_bot:KleinanzeigenBot) -> None: - """Test parsing of create-config command""" - test_bot.parse_args(["app", "create-config"]) - assert test_bot.command == "create-config" - - -class TestKleinanzeigenBotConfiguration: - """Tests for configuration loading and validation.""" - - def test_load_config_handles_missing_file(self, test_bot:KleinanzeigenBot, test_data_dir:str, monkeypatch:pytest.MonkeyPatch) -> None: - """Verify that loading a missing config file creates default config. No info log is expected anymore.""" - monkeypatch.setenv("KLEINANZEIGEN_BOT_USERNAME", "dummy_user") - monkeypatch.setenv("KLEINANZEIGEN_BOT_PASSWORD", "dummy_pass") - 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 TestResolveLoginCredentials: - """Tests for _resolve_login_credentials env var resolution.""" - # ruff: noqa: S105 test strings are not real passwords - - def test_resolves_var_from_environment(self, monkeypatch:pytest.MonkeyPatch) -> None: - """${VAR} is replaced by the env var value.""" - monkeypatch.setenv("TEST_BOT_USER", "resolved_user") - monkeypatch.setenv("TEST_BOT_PASS", "resolved_pass") - config:dict[str, Any] = { - "login": { - "username": "${TEST_BOT_USER}", - "password": "${TEST_BOT_PASS}", - }, - } - KleinanzeigenBot._resolve_login_credentials(config) - assert config["login"]["username"] == "resolved_user" - assert config["login"]["password"] == "resolved_pass" - - def test_uses_fallback_when_var_unset(self, monkeypatch:pytest.MonkeyPatch) -> None: - """${VAR:-default} uses the default when VAR is not set.""" - monkeypatch.delenv("OPTIONAL_USER", raising = False) - monkeypatch.delenv("OPTIONAL_PASS", raising = False) - config:dict[str, Any] = { - "login": { - "username": "${OPTIONAL_USER:-fallback_user}", - "password": "${OPTIONAL_PASS:-fallback_pass}", - }, - } - KleinanzeigenBot._resolve_login_credentials(config) - assert config["login"]["username"] == "fallback_user" - assert config["login"]["password"] == "fallback_pass" - - def test_raises_value_error_when_required_var_unset(self, monkeypatch:pytest.MonkeyPatch) -> None: - """${VAR} raises ValueError when VAR is not set and no default.""" - monkeypatch.delenv("MUST_HAVE_USER", raising = False) - config:dict[str, Any] = { - "login": { - "username": "${MUST_HAVE_USER}", - "password": "${MUST_HAVE_PASS}", - }, - } - with pytest.raises(ValueError, match = "Environment variable MUST_HAVE_USER is required for login"): - KleinanzeigenBot._resolve_login_credentials(config) - - def test_leaves_plain_text_values_unchanged(self) -> None: - """Non-placeholder values are not modified.""" - config:dict[str, Any] = { - "login": { - "username": "real_user", - "password": "real_pass", - }, - } - KleinanzeigenBot._resolve_login_credentials(config) - assert config["login"]["username"] == "real_user" - assert config["login"]["password"] == "real_pass" - - def test_handles_missing_login_key(self) -> None: - """Missing 'login' key does nothing.""" - config:dict[str, Any] = {} - KleinanzeigenBot._resolve_login_credentials(config) - assert "login" not in config - - def test_handles_login_as_non_dict(self) -> None: - """Login being a non-dict (e.g. str) does nothing.""" - config:dict[str, Any] = {"login": "some_string"} - KleinanzeigenBot._resolve_login_credentials(config) - assert config["login"] == "some_string" - - def test_handles_non_string_field_values(self) -> None: - """Non-string values like None or int are left untouched.""" - config:dict[str, Any] = { - "login": { - "username": None, - "password": 12345, - }, - } - KleinanzeigenBot._resolve_login_credentials(config) - assert config["login"]["username"] is None - assert config["login"]["password"] == 12345 - - def test_resolves_only_username_and_password(self, monkeypatch:pytest.MonkeyPatch) -> None: - """Arbitrary login fields (like api_key) are NOT resolved.""" - monkeypatch.setenv("USER", "user1") - monkeypatch.setenv("PASS", "pass1") - monkeypatch.setenv("API_KEY", "secret123") - config:dict[str, Any] = { - "login": { - "username": "${USER}", - "password": "${PASS}", - "api_key": "${API_KEY}", - }, - } - KleinanzeigenBot._resolve_login_credentials(config) - assert config["login"]["username"] == "user1" - assert config["login"]["password"] == "pass1" - assert config["login"]["api_key"] == "${API_KEY}" # unchanged - - def test_load_config_resolves_env_vars_in_login(self, test_bot:KleinanzeigenBot, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: - """load_config resolves ${VAR} placeholders from environment into the Config object.""" - monkeypatch.setenv("BOT_USERNAME", "env_user") - monkeypatch.setenv("BOT_PASSWORD", "env_pass") - - config_path = tmp_path / "config.yaml" - config_path.write_text(""" -login: - username: ${BOT_USERNAME} - password: ${BOT_PASSWORD} -ad_defaults: - contact: - name: Test User - zipcode: "12345" -publishing: - delete_old_ads: BEFORE_PUBLISH - delete_old_ads_by_title: false -""") - test_bot.config_file_path = str(config_path) - test_bot.load_config() - - assert test_bot.config.login.username == "env_user" - assert test_bot.config.login.password == "env_pass" - class TestKleinanzeigenBotAuthentication: """Tests for login and authentication functionality.""" @@ -1886,28 +1486,6 @@ class TestKleinanzeigenBotDiagnostics: 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.""" @@ -2249,14 +1827,6 @@ class TestKleinanzeigenBotBasics: ): await test_bot.publish_ad("ad.yaml", ad_cfg, ad_cfg_orig, [], AdUpdateStrategy.MODIFY) - 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, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: """Test config file path handling.""" monkeypatch.chdir(tmp_path) @@ -2714,108 +2284,6 @@ class TestKleinanzeigenBotContactLocationHardening: combobox_mock.assert_not_called() -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, tmp_path:Path) -> None: - """Test parsing log file path.""" - log_path = tmp_path / "test.log" - test_bot.parse_args(["script.py", f"--logfile={log_path}", "help"]) - assert test_bot.log_file_path == str(log_path.absolute()) - - 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.""" @@ -2945,22 +2413,6 @@ class TestKleinanzeigenBotAdManagement: 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) diff --git a/tests/unit/test_runtime_config.py b/tests/unit/test_runtime_config.py new file mode 100644 index 0000000..4997e48 --- /dev/null +++ b/tests/unit/test_runtime_config.py @@ -0,0 +1,323 @@ +# SPDX-FileCopyrightText: © Jens Bergmann and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ +"""Unit tests for runtime config loading, validation, workspaces, and env overrides.""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import TYPE_CHECKING, Any +from unittest.mock import MagicMock, patch + +import pytest + +from kleinanzeigen_bot import runtime_config +from kleinanzeigen_bot.model.config_model import Config +from kleinanzeigen_bot.utils import dicts, xdg_paths +from kleinanzeigen_bot.utils.timing_collector import TimingCollector + +pytestmark = pytest.mark.unit + +if TYPE_CHECKING: + from types import ModuleType + + +def _write_minimal_config(config_path:Path) -> None: + config_path.write_text( + """ +login: + username: ${BOT_USERNAME} + password: ${BOT_PASSWORD} +ad_defaults: + contact: + name: Test User + zipcode: "12345" +publishing: + delete_old_ads: BEFORE_PUBLISH + delete_old_ads_by_title: false +""".strip(), + encoding = "utf-8", + ) + + +class TestRuntimeConfig: + def test_resolve_workspace_skips_bootstrap_commands(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + + assert ( + runtime_config.resolve_workspace( + command = "help", + config_file_path = str(config_path), + config_arg = None, + logfile_arg = None, + workspace_mode = None, + logfile_explicitly_provided = False, + log_basename = "kleinanzeigen-bot", + ) + is None + ) + + def test_resolve_workspace_honors_explicit_logfile(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + log_path = tmp_path / "custom.log" + + workspace = runtime_config.resolve_workspace( + command = "verify", + config_file_path = str(config_path), + config_arg = str(config_path), + logfile_arg = str(log_path), + workspace_mode = "portable", + logfile_explicitly_provided = True, + log_basename = "kleinanzeigen-bot", + ) + + assert workspace is not None + assert workspace.log_file == log_path.resolve() + + def test_create_default_config_creates_file(self, tmp_path:Path) -> None: + config_path = tmp_path / "nested" / "config.yaml" + + runtime_config.create_default_config(str(config_path), workspace = None) + + assert config_path.exists() + contents = config_path.read_text() + assert "username: changeme" in contents + assert "password: changeme" in contents + + def test_load_config_resolves_login_env_and_browser_defaults(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.setenv("BOT_USERNAME", "env_user") + monkeypatch.setenv("BOT_PASSWORD", "env_pass") + + config_path = tmp_path / "config.yaml" + _write_minimal_config(config_path) + workspace = xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + + state = runtime_config.load_config(str(config_path), workspace, "verify") + + assert state.config.login.username == "env_user" + assert state.config.login.password == "env_" + "pass" + assert state.categories + + def test_load_config_uses_login_env_defaults_and_custom_categories(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.delenv("BOT_USERNAME", raising = False) + + config_path = tmp_path / "config.yaml" + config_path.write_text( + """ +login: + username: ${BOT_USERNAME:-fallback_user} + password: pass +ad_defaults: + contact: + name: Test User + zipcode: "12345" +categories: + Custom > Category: "1/2" +diagnostics: + timing_collection: false +publishing: + delete_old_ads: BEFORE_PUBLISH + delete_old_ads_by_title: false +""".strip(), + encoding = "utf-8", + ) + + state = runtime_config.load_config(str(config_path), workspace = None, command = "verify") + + assert state.config.login.username == "fallback_user" + assert state.categories["Custom > Category"] == "1/2" + assert state.timing_collector is None + + def test_load_config_raises_for_missing_login_env(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.delenv("BOT_USERNAME", raising = False) + monkeypatch.setenv("BOT_PASSWORD", "env_pass") + + config_path = tmp_path / "config.yaml" + _write_minimal_config(config_path) + workspace = xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + + with pytest.raises(ValueError, match = r"Environment variable BOT_USERNAME is required"): + runtime_config.load_config(str(config_path), workspace, "verify") + + def test_load_config_enables_timing_collection(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: + monkeypatch.setenv("BOT_USERNAME", "env_user") + monkeypatch.setenv("BOT_PASSWORD", "env_pass") + + config_path = tmp_path / "config.yaml" + config_path.write_text( + """ +login: + username: ${BOT_USERNAME} + password: ${BOT_PASSWORD} +ad_defaults: + contact: + name: Test User + zipcode: "12345" +diagnostics: + timing_collection: true +publishing: + delete_old_ads: BEFORE_PUBLISH + delete_old_ads_by_title: false +""".strip(), + encoding = "utf-8", + ) + workspace = xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + + state = runtime_config.load_config(str(config_path), workspace, "verify") + + assert isinstance(state.timing_collector, TimingCollector) + assert state.timing_collector.output_dir == workspace.diagnostics_dir.parent / "timing" + assert state.timing_collector.command == "verify" + + def test_load_config_handles_empty_categories( + self, + tmp_path:Path, + monkeypatch:pytest.MonkeyPatch, + ) -> None: + monkeypatch.setenv("BOT_USERNAME", "env_user") + monkeypatch.setenv("BOT_PASSWORD", "env_pass") + + config_path = tmp_path / "config.yaml" + config_path.write_text( + """ +login: + username: ${BOT_USERNAME} + password: ${BOT_PASSWORD} +ad_defaults: + contact: + name: Test User + zipcode: "12345" +publishing: + delete_old_ads: BEFORE_PUBLISH + delete_old_ads_by_title: false +""".strip(), + encoding = "utf-8", + ) + + load_dict_from_module = dicts.load_dict_from_module + + def fake_load_dict_from_module(module:ModuleType, filename:str, content_label:str = "") -> dict[str, Any]: + if filename in {"categories.yaml", "categories_old.yaml"}: + return {} + return load_dict_from_module(module, filename, content_label) + + with ( + patch( + "kleinanzeigen_bot.runtime_config._dicts.load_dict_from_module", + fake_load_dict_from_module, + ), + ): + state = runtime_config.load_config(str(config_path), workspace = None, command = "verify") + + assert state.categories == {} + + def test_apply_browser_config_uses_workspace_profile_when_custom_dir_missing(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + workspace = xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + browser_config = MagicMock() + config = Config.model_validate( + { + "login": {"username": "user", "password": "pass"}, + "ad_defaults": {"contact": {"name": "Test User", "zipcode": "12345"}}, + "publishing": {"delete_old_ads": "BEFORE_PUBLISH", "delete_old_ads_by_title": False}, + } + ) + + runtime_config.apply_browser_config(browser_config, config, workspace, str(config_path)) + + assert browser_config.user_data_dir == str(workspace.browser_profile_dir) + assert browser_config.profile_name == config.browser.profile_name + + def test_apply_browser_config_uses_custom_profile_dir(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + config_path.touch() + workspace = xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + browser_config = MagicMock() + config = Config.model_validate( + { + "login": {"username": "user", "password": "pass"}, + "ad_defaults": {"contact": {"name": "Test User", "zipcode": "12345"}}, + "browser": {"user_data_dir": "profiles/custom"}, + "publishing": {"delete_old_ads": "BEFORE_PUBLISH", "delete_old_ads_by_title": False}, + }, + context = str(config_path), + ) + + runtime_config.apply_browser_config(browser_config, config, workspace, str(config_path)) + + assert browser_config.user_data_dir == str(tmp_path / "profiles" / "custom") + + def test_configure_file_logging_creates_handler(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + workspace = xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + log_path = tmp_path / "bot.log" + + file_log = runtime_config.configure_file_logging(str(log_path), workspace, None, "1.2.3") + try: + assert file_log is not None + assert log_path.exists() + finally: + if file_log is not None: + file_log.close() + + def test_configure_file_logging_skips_empty_path(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + workspace = xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + + assert runtime_config.configure_file_logging(None, workspace, None, "1.2.3") is None + + def test_resolve_workspace_infers_mode_from_config_path(self, tmp_path:Path, monkeypatch:pytest.MonkeyPatch) -> None: + config_path = tmp_path / "nested" / "config.yaml" + captured:dict[str, object] = {} + + def fake_resolve_workspace(**kwargs:object) -> xdg_paths.Workspace: + captured.update(kwargs) + return xdg_paths.Workspace.for_config(config_path, "kleinanzeigen-bot") + + monkeypatch.setattr(xdg_paths, "resolve_workspace", fake_resolve_workspace) + + runtime_config.resolve_workspace( + command = "verify", + config_file_path = str(config_path), + config_arg = None, + logfile_arg = None, + workspace_mode = None, + logfile_explicitly_provided = False, + log_basename = "kleinanzeigen-bot", + ) + + assert captured["config_arg"] == str(config_path) + assert captured["workspace_mode"] == "portable" + + def test_resolve_workspace_exits_on_workspace_error(self, tmp_path:Path) -> None: + config_path = tmp_path / "config.yaml" + + with patch("kleinanzeigen_bot.utils.xdg_paths.resolve_workspace", side_effect = ValueError("boom")), pytest.raises(SystemExit) as exc_info: + runtime_config.resolve_workspace( + command = "verify", + config_file_path = str(config_path), + config_arg = str(config_path), + logfile_arg = None, + workspace_mode = None, + logfile_explicitly_provided = False, + log_basename = "kleinanzeigen-bot", + ) + + assert exc_info.value.code == 2 + + +def _extract_match_cases() -> frozenset[str]: + """Parse case statements from KleinanzeigenBot.run() in __init__.py.""" + init_path = Path(__file__).resolve().parent.parent.parent / "src" / "kleinanzeigen_bot" / "__init__.py" + source = init_path.read_text() + return frozenset(re.findall(r'^\s+case\s+"([^"]+)"', source, flags = re.MULTILINE)) + + +def test_valid_commands_matches_dispatch_cases() -> None: + """VALID_COMMANDS must stay in sync with the match cases in run(). + + Failures mean a command was added/removed from the dispatch without + updating runtime_config.VALID_COMMANDS. + """ + assert _extract_match_cases() == runtime_config.VALID_COMMANDS