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

refactor: extract CLI bootstrap (#1089)

Browse Source 
## ℹ️ Description
This pull request extracts CLI parsing and runtime bootstrap into
dedicated modules while keeping the existing command behavior intact.

- Link to the related issue(s): Issue #
- Describe the motivation and context for this change.

## 📋 Changes Summary

- Moved CLI parsing/help logic into `src/kleinanzeigen_bot/cli.py`.
- Moved workspace resolution, config loading, browser config
application, and file logging setup into
`src/kleinanzeigen_bot/runtime_config.py`.
- Kept `python -m kleinanzeigen_bot` behavior unchanged via
`src/kleinanzeigen_bot/__main__.py`.
- Preserved frozen build startup by importing `runtime_config`
statically.
- Added focused tests for CLI parsing and runtime bootstrap behavior.
- Updated contributor docs and translation entries for the moved
user-facing messages.

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

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

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


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

* **Documentation**
  * Clarified bootstrap docs to state the app’s CLI-driven entry point.

* **New Features**
* Localized CLI with improved argument parsing, help text and banner
control.
* Runtime bootstrapping: default-config creation, env-var credential
placeholders, category loading, workspace-aware browser profile
handling, and optional file logging.

* **Refactor**
* Centralized startup into the CLI/runtime layer; package entrypoint now
delegates to the CLI.

* **Tests**
* Added/updated unit and smoke tests covering CLI, runtime config,
workspace behavior and create-config.

* **Chores**
* Updated project script wrappers to print tool headers before running
linters.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Jens
2026-06-07 07:05:53 +02:00
committed by GitHub
parent b98ae58f77
commit faedfad23c
12 changed files with 1220 additions and 1070 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CONTRIBUTING.md
View File

@@ -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()
```

10
pyproject.toml
View File

@@ -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

486
src/kleinanzeigen_bot/__init__.py
View File

@@ -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<var>\w+)(?::-(?P<default>.*))?\}$")
_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|<id(s)> (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
* <id(s)>: 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|<id(s)> (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
* <id(s)>: Gibt eine oder mehrere Anzeigen-IDs zum Herunterladen an, z. B. "--ads=1,2,3"
--ads=all|changed|<id(s)> (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
* <id(s)>: Gibt eine oder mehrere Anzeigen-IDs zum Aktualisieren an, z. B. "--ads=1,2,3"
--ads=all|<id(s)> (extend) - Gibt an, welche Anzeigen verlängert werden sollen
Mögliche Werte:
* all: Verlängert alle Anzeigen, die innerhalb von 8 Tagen ablaufen
* <id(s)>: 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=<PATH> - 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=<PATH> - 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|<id(s)> (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
* <id(s)>: 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|<id(s)> (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
* <id(s)>: provide one or several ads by ID to download, like e.g. "--ads=1,2,3"
--ads=all|changed|<id(s)> (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
* <id(s)>: provide one or several ads by ID to update, like e.g. "--ads=1,2,3"
--ads=all|<id(s)> (extend) - specifies which ads to extend
Possible values:
* all: extend all ads expiring within 8 days
* <id(s)>: 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> - 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> - 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__":

4
src/kleinanzeigen_bot/__main__.py
View File

@@ -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

271
src/kleinanzeigen_bot/cli.py Normal file
View File

@@ -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 <kleinanzeigen_bot.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|<id(s)> (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
* <id(s)>: 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|<id(s)> (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
* <id(s)>: Gibt eine oder mehrere Anzeigen-IDs zum Herunterladen an, z. B. "--ads=1,2,3"
--ads=all|changed|<id(s)> (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
* <id(s)>: Gibt eine oder mehrere Anzeigen-IDs zum Aktualisieren an, z. B. "--ads=1,2,3"
--ads=all|<id(s)> (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
* <id(s)>: 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=<PATH> - 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=<PATH> - 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|<id(s)> (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
* <id(s)>: 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|<id(s)> (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
* <id(s)>: provide one or several ads by ID to download, like e.g. "--ads=1,2,3"
--ads=all|changed|<id(s)> (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
* <id(s)>: provide one or several ads by ID to update, like e.g. "--ads=1,2,3"
--ads=all|<id(s)> (extend) - specifies which ads to extend (DEFAULT: all)
Possible values:
* all: extend all ads expiring within 8 days
* <id(s)>: 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> - 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> - 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)

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

@@ -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"

235
src/kleinanzeigen_bot/runtime_config.py Normal file
View File

@@ -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<var>\w+)(?::-(?P<default>.*))?\}$")
# 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

40
tests/smoke/test_smoke_health.py
View File

@@ -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:
"""

194
tests/unit/test_cli.py Normal file
View File

@@ -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"]]

77
tests/unit/test_extend_command.py
View File

@@ -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."""

590
tests/unit/test_init.py
View File

@@ -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)

323
tests/unit/test_runtime_config.py Normal file
View File

@@ -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