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

enh: support Python 3.12 through 3.15 (#1235)

Browse Source 
## ℹ️ Description

- Closes #1226
- Raise the supported Python range from 3.10–3.14 to 3.12–3.15 before
Python 3.10 reaches end of life.
- Modernize the codebase with behavior-preserving Python 3.11 and 3.12
idioms while keeping CLI, configuration, persistence, and browser
behavior stable.

## 📋 Changes Summary

- Set `requires-python` and the PDM lock target to `>=3.12,<3.16`.
- Test Python 3.12 and Python 3.15 prereleases across all supported CI
platforms.
- Build and publish native release artifacts and Docker images with
stable Python 3.12; keep Python 3.15 as compatibility-only coverage
until its final release.
- Update CodeQL, dependency updates, local Act configuration, Ruff,
mypy, BasedPyright, README, CONTRIBUTING, and AGENTS guidance.
- Enable Ruff's `UP` rules and adopt applicable standard-library
imports, `datetime.UTC`, built-in `TimeoutError`, PEP 695 type aliases
and generic syntax.
- Replace deprecated `shutil.rmtree(onerror=...)` usage with Python
3.12's `onexc` callback API and update its tests.
- Regenerate `pdm.lock` for the full supported interpreter range and
integrate the latest dependency updates from `main`.

### ⚙️ Type of Change

- [ ] 🐞 Bug fix (non-breaking change which fixes an issue)
- [ ] ✨ New feature (adds new functionality without breaking existing
usage)
- [x] 💥 Breaking change (changes that might break existing user setups,
scripts, or configurations)

## ✅ Checklist

- [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.

Validation details:

- 1507 tests passed, 4 skipped
- 93.07% total coverage
- Generated schemas, default configuration, and README usage are up to
date
- Source CLI and PyInstaller binary smoke checks passed
- Local PyInstaller build and binary smoke checks passed on Python
3.12.14; Python 3.12/3.15 platform coverage is delegated to the updated
CI matrix

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

* **New Features**
  * Added support for Python 3.12 through 3.15.
  * Improved compatibility with modern Python typing and datetime APIs.

* **Bug Fixes**
  * Improved timeout handling and directory cleanup behavior.
  * Enhanced update-checking and web-scraping reliability.

* **Documentation**
* Updated installation and contribution requirements to Python
3.12–3.15.

* **Tests**
* Expanded coverage for update checks, timeout messages, and edge cases.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Jens
2026-08-14 11:56:12 +02:00
committed by GitHub
parent 30b7bfceeb
commit f4b4e0fdf6
42 changed files with 591 additions and 733 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.actrc
View File

@@ -6,4 +6,4 @@
-W .github/workflows/build.yml
-j build
--matrix os:ubuntu-latest
--matrix PYTHON_VERSION:3.14
--matrix PYTHON_VERSION:3.15

2
.coderabbit.yaml
View File

@@ -131,7 +131,7 @@ reviews:
# Auto review configuration
auto_review:
enabled: true
auto_incremental_review: true
auto_incremental_review: false
drafts: false
labels: ["!wip", "!draft"] # Review all PRs except those with wip or draft labels

49
.github/workflows/build.yml vendored
View File

@@ -78,36 +78,36 @@ jobs:
matrix:
include:
- os: macos-15-intel # X86
PYTHON_VERSION: "3.10"
PUBLISH_RELEASE: false
PYTHON_VERSION: "3.12"
PUBLISH_RELEASE: true
- os: macos-latest # ARM
PYTHON_VERSION: "3.10"
PUBLISH_RELEASE: false
PYTHON_VERSION: "3.12"
PUBLISH_RELEASE: true
- os: ubuntu-latest
PYTHON_VERSION: "3.10"
PUBLISH_RELEASE: false
PYTHON_VERSION: "3.12"
PUBLISH_RELEASE: true
- os: ubuntu-24.04-arm # https://github.com/actions/partner-runner-images#available-images
PYTHON_VERSION: "3.10"
PUBLISH_RELEASE: false
PYTHON_VERSION: "3.12"
PUBLISH_RELEASE: true
- os: windows-latest
PYTHON_VERSION: "3.10"
PUBLISH_RELEASE: false
PYTHON_VERSION: "3.12"
PUBLISH_RELEASE: true
- os: macos-15-intel # X86
PYTHON_VERSION: "3.14"
PUBLISH_RELEASE: true
PYTHON_VERSION: "3.15"
PUBLISH_RELEASE: false
- os: macos-latest # ARM
PYTHON_VERSION: "3.14"
PUBLISH_RELEASE: true
PYTHON_VERSION: "3.15"
PUBLISH_RELEASE: false
- os: ubuntu-latest
PYTHON_VERSION: "3.14"
PUBLISH_RELEASE: true
PYTHON_VERSION: "3.15"
PUBLISH_RELEASE: false
- os: ubuntu-24.04-arm # https://github.com/actions/partner-runner-images#available-images
PYTHON_VERSION: "3.14"
PUBLISH_RELEASE: true
PYTHON_VERSION: "3.15"
PUBLISH_RELEASE: false
- os: windows-latest
PYTHON_VERSION: "3.14"
PUBLISH_RELEASE: true
PYTHON_VERSION: "3.15"
PUBLISH_RELEASE: false
runs-on: ${{ matrix.os }} # https://github.com/actions/runner-images#available-images
timeout-minutes: 20
@@ -153,6 +153,7 @@ jobs:
uses: pdm-project/setup-pdm@973541a5febeafcfdadf8a51211435be6ecfd90f # v4.5
with:
python-version: "${{ matrix.PYTHON_VERSION }}"
allow-python-prereleases: ${{ matrix.PYTHON_VERSION == '3.15' }}
cache: ${{ !startsWith(matrix.os, 'macos') }} # https://github.com/pdm-project/setup-pdm/issues/55
@@ -178,17 +179,19 @@ jobs:
- name: Check generated schemas and default docs config
if: matrix.os == 'ubuntu-latest' && matrix.PYTHON_VERSION == '3.14'
if: matrix.os == 'ubuntu-latest' && matrix.PYTHON_VERSION == '3.15'
run: pdm run python scripts/check_generated_artifacts.py
- name: Check GitHub Actions workflows
if: matrix.os == 'ubuntu-latest' && matrix.PYTHON_VERSION == '3.14'
if: matrix.os == 'ubuntu-latest' && matrix.PYTHON_VERSION == '3.15'
run: pdm run lint:actions
- name: Check with pip-audit
# until https://github.com/astral-sh/ruff/issues/8277
# pip-api imports sre_constants, which was removed in Python 3.15.
if: matrix.PYTHON_VERSION != '3.15'
run:
pdm run pip-audit --progress-spinner off --skip-editable --verbose
@@ -312,7 +315,7 @@ jobs:
- name: Build Docker image
if: startsWith(matrix.os, 'ubuntu')
if: startsWith(matrix.os, 'ubuntu') && matrix.PUBLISH_RELEASE
run: |
set -eux

3
.github/workflows/codeql-analysis.yml vendored
View File

@@ -34,7 +34,7 @@ defaults:
shell: bash
env:
PYTHON_VERSION: "3.14"
PYTHON_VERSION: "3.15"
jobs:
@@ -75,6 +75,7 @@ jobs:
uses: pdm-project/setup-pdm@973541a5febeafcfdadf8a51211435be6ecfd90f # v4.5
with:
python-version: "${{ env.PYTHON_VERSION }}"
allow-python-prereleases: true
cache: true

2
.github/workflows/update-python-deps.yml vendored
View File

@@ -17,7 +17,7 @@ defaults:
shell: bash
env:
PYTHON_VERSION: "3.10"
PYTHON_VERSION: "3.12"
permissions:
contents: write

4
AGENTS.md
View File

@@ -22,7 +22,7 @@ Before making non-trivial changes, review:
- For runtime/user-facing output, follow the translation rules in `CONTRIBUTING.md` and update translations when messages change.
- Keep log message strings in plain English; do **not** wrap `LOG.*`/`logger.*` strings with `_()`, because logging messages are translated by `TranslatingLogger`.
- New Python files need the full SPDX header block from `CONTRIBUTING.md`.
- Use full type hints (Python 3.10+ syntax).
- Use full type hints (Python 3.12+ syntax).
- Catch `TimeoutError` in browser automation paths.
- Never hardcode credentials or secrets.
- Prefer small, simple changes over speculative abstractions.
@@ -95,7 +95,7 @@ CI and workflows are the source of truth for the exact required checks, coverage
## PR Expectations
- PR titles must follow the semantic format enforced by `.github/workflows/validate-pr-title.yml`.
- Branch names should use the same conventional type prefix as the PR title, e.g. `docs/update-agent-playbook`, `fix/browser-timeout`, or `feat/price-logging`.
- Branch names must use the same conventional type prefix as the PR title, e.g. `docs/update-agent-playbook`, `fix/browser-timeout`, or `feat/price-logging`. Never add an agent- or tool-specific prefix such as `agent/`.
- PR descriptions should use `.github/PULL_REQUEST_TEMPLATE.md` and complete its required sections and checklist.
- Do not open a PR with placeholder sections, missing checklist decisions, or a non-semantic title; fix the title/body before publishing.

2
CONTRIBUTING.md
View File

@@ -25,7 +25,7 @@ Please read through this document before submitting any contributions to ensure
### Prerequisites
- Python 3.10 or higher
- Python 3.12 through 3.15
- PDM for dependency management
- Git

2
README.md
View File

@@ -115,7 +115,7 @@ Die Nutzung erfolgt auf eigenes Risiko. Jede rechtswidrige Verwendung ist unters
1. [Chromium](https://www.chromium.org/getting-involved/download-chromium), [Google Chrome](https://www.google.com/chrome/),
or Chromium-based [Microsoft Edge](https://www.microsoft.com/edge) browser
1. [Python](https://www.python.org/) **3.10** or newer
1. [Python](https://www.python.org/) **3.12 through 3.15**
1. [pip](https://pypi.org/project/pip/)
1. [git client](https://git-scm.com/downloads)

2
docker/image/Dockerfile
View File

@@ -48,7 +48,7 @@ EOF
######################
# https://hub.docker.com/_/python/tags?name=3-slim
FROM python:3.14-slim AS build-image
FROM python:3.12-slim AS build-image
ARG DEBIAN_FRONTEND=noninteractive
ARG LC_ALL=C

868
pdm.lock generated
View File

File diff suppressed because it is too large Load Diff

15
pyproject.toml
View File

@@ -31,9 +31,12 @@ classifiers = [ # https://pypi.org/classifiers/
"License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10"
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Programming Language :: Python :: 3.15"
]
requires-python = ">=3.10,<3.15"
requires-python = ">=3.12,<3.16"
dependencies = [
"certifi",
"colorama",
@@ -166,7 +169,7 @@ cache-dir = ".temp/cache_ruff"
include = ["pyproject.toml", "scripts/**/*.py", "src/**/*.py", "tests/**/*.py"]
line-length = 160
indent-width = 4
target-version = "py310"
target-version = "py312"
[tool.ruff.lint]
# https://docs.astral.sh/ruff/rules/
@@ -210,6 +213,7 @@ select = [
"TC", # flake8-type-checking
"TD", # flake8-todo
"TID", # flake8-flake8-tidy-import
"UP", # pyupgrade
"YTT", # flake8-2020
"E", # pycodestyle-errors
@@ -226,6 +230,7 @@ select = [
]
ignore = [
"ANN401", # Dynamically typed expressions (typing.Any) are disallowed
"ASYNC109", # Timeout parameters are passed through to browser helpers; replacing them with asyncio.timeout would change behavior
"COM812", # Trailing comma missing
"D1", # Missing docstring in ...
"D200", # One-line docstring should fit on one line
@@ -311,7 +316,7 @@ max-complexity = 30
# https://mypy.readthedocs.io/en/stable/config_file.html
#mypy_path = "$MYPY_CONFIG_FILE_DIR/tests/stubs"
cache_dir = ".temp/cache_mypy"
python_version = "3.10"
python_version = "3.12"
files = "scripts,src,tests"
strict = true
disallow_untyped_calls = false
@@ -331,7 +336,7 @@ verbosity = 0
# https://docs.basedpyright.com/latest/configuration/config-files/
include = ["scripts", "src", "tests"]
defineConstant = { DEBUG = false }
pythonVersion = "3.10"
pythonVersion = "3.12"
typeCheckingMode = "standard"

22
scripts/post_autopep8.py
View File

@@ -3,9 +3,7 @@
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import ast, logging, re, sys # isort: skip
from pathlib import Path
from typing import Final, List, Protocol, Tuple
from typing_extensions import override
from typing import Final, Protocol, override
# Configure basic logging
logging.basicConfig(level = logging.INFO, format = "%(levelname)s: %(message)s")
@@ -17,7 +15,7 @@ class FormatterRule(Protocol):
A code processor that can modify source lines based on the AST.
"""
def apply(self, tree:ast.AST, lines:List[str], path:Path) -> List[str]:
def apply(self, tree:ast.AST, lines:list[str], path:Path) -> list[str]:
raise NotImplementedError
@@ -39,8 +37,8 @@ class NoSpaceAfterColonInTypeAnnotationRule(FormatterRule):
"""
@override
def apply(self, tree:ast.AST, lines:List[str], path:Path) -> List[str]:
ann_positions:List[Tuple[int, int]] = []
def apply(self, tree:ast.AST, lines:list[str], path:Path) -> list[str]:
ann_positions:list[tuple[int, int]] = []
for node in ast.walk(tree):
if isinstance(node, ast.arg) and node.annotation is not None:
ann_positions.append((node.annotation.lineno - 1, node.annotation.col_offset))
@@ -51,7 +49,7 @@ class NoSpaceAfterColonInTypeAnnotationRule(FormatterRule):
if not ann_positions:
return lines
new_lines:List[str] = []
new_lines:list[str] = []
for idx, line in enumerate(lines):
if line.lstrip().startswith("#"):
new_lines.append(line)
@@ -95,8 +93,8 @@ class EqualSignSpacingInDefaultsAndNamedArgsRule(FormatterRule):
"""
@override
def apply(self, tree:ast.AST, lines:List[str], path:Path) -> List[str]:
equals_positions:List[Tuple[int, int]] = []
def apply(self, tree:ast.AST, lines:list[str], path:Path) -> list[str]:
equals_positions:list[tuple[int, int]] = []
for node in ast.walk(tree):
# --- Defaults in function definitions, async defs & lambdas ---
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)):
@@ -125,7 +123,7 @@ class EqualSignSpacingInDefaultsAndNamedArgsRule(FormatterRule):
if not equals_positions:
return lines
new_lines:List[str] = []
new_lines:list[str] = []
for line_idx, line in enumerate(lines):
if line.lstrip().startswith("#"):
new_lines.append(line)
@@ -174,7 +172,7 @@ class PreferDoubleQuotesRule(FormatterRule):
"""
@override
def apply(self, tree:ast.AST, lines:List[str], path:Path) -> List[str]:
def apply(self, tree:ast.AST, lines:list[str], path:Path) -> list[str]:
new_lines = lines.copy()
# Track how much each line has shifted so far
@@ -256,7 +254,7 @@ class PreferDoubleQuotesRule(FormatterRule):
return new_lines
FORMATTER_RULES:List[FormatterRule] = [
FORMATTER_RULES:list[FormatterRule] = [
NoSpaceAfterColonInTypeAnnotationRule(),
EqualSignSpacingInDefaultsAndNamedArgsRule(),
PreferDoubleQuotesRule(),

2
src/kleinanzeigen_bot/app.py
View File

@@ -68,7 +68,7 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
# capture_login_detection_diagnostics_if_enabled can read/write it
# via getattr/setattr. The per-attempt reset happens in login_flow.login().
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:

5
src/kleinanzeigen_bot/cli.py
View File

@@ -20,7 +20,7 @@ import sys
import textwrap
from dataclasses import dataclass
from pathlib import Path
from typing import Final, Sequence
from typing import TYPE_CHECKING, Final
import colorama
import nodriver
@@ -35,6 +35,9 @@ 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
if TYPE_CHECKING:
from collections.abc import Sequence
LOG:Final[_loggers.Logger] = _loggers.get_logger(__name__)
LOG.setLevel(_loggers.INFO)

5
src/kleinanzeigen_bot/download_selection.py
View File

@@ -4,7 +4,10 @@
from __future__ import annotations
import re
from typing import Any, Final, Mapping, NamedTuple
from typing import TYPE_CHECKING, Any, Final, NamedTuple
if TYPE_CHECKING:
from collections.abc import Mapping
__all__ = [
"NUMERIC_IDS_RE",

7
src/kleinanzeigen_bot/extract.py
View File

@@ -72,8 +72,7 @@ def _is_retryable_rmtree_error(error:BaseException) -> bool:
return error.errno in {errno.EACCES, errno.EPERM, errno.EBUSY}
def _handle_rmtree_onerror(func:Any, path:str, exc_info:tuple[type[BaseException], BaseException, Any]) -> None:
error = exc_info[1]
def _handle_rmtree_onexc(func:Any, path:str, error:BaseException) -> None:
if not _is_retryable_rmtree_error(error):
raise error
@@ -99,7 +98,7 @@ def _remove_tree_with_retries(path:Path) -> None:
last_error:OSError | None = None
for attempt in range(_RMTREE_RETRY_ATTEMPTS):
try:
shutil.rmtree(path, onerror = _handle_rmtree_onerror)
shutil.rmtree(path, onexc = _handle_rmtree_onexc)
return
except FileNotFoundError:
return
@@ -603,7 +602,7 @@ class AdExtractor(WebScrapingMixin):
"""
if reflect.is_integer(id_or_url):
# navigate to search page
await self.web_open("https://www.kleinanzeigen.de/s-suchanfrage.html?keywords={0}".format(id_or_url))
await self.web_open(f"https://www.kleinanzeigen.de/s-suchanfrage.html?keywords={id_or_url}")
else:
await self.web_open(str(id_or_url)) # navigate to URL directly given
await self.web_sleep()

8
src/kleinanzeigen_bot/login_flow.py
View File

@@ -14,11 +14,11 @@ import asyncio
import enum
import sys
import urllib.parse as urllib_parse
from collections.abc import Callable
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from gettext import gettext as _
from pathlib import Path
from typing import Final, Sequence
from typing import Final
from nodriver.core.connection import ProtocolException
@@ -477,7 +477,7 @@ async def wait_for_post_auth0_submit_transition(
login_confirmed = False
try:
login_confirmed = await asyncio.wait_for(is_logged_in(web, username = username), timeout = post_submit_timeout)
except (TimeoutError, asyncio.TimeoutError):
except TimeoutError:
LOG.debug("Post-submit login verification did not complete within %.1fs", post_submit_timeout)
if login_confirmed:
@@ -490,7 +490,7 @@ async def wait_for_post_auth0_submit_transition(
try:
if await asyncio.wait_for(is_logged_in(web, username = username), timeout = quick_dom_timeout):
return
except (TimeoutError, asyncio.TimeoutError):
except TimeoutError:
LOG.debug("Final post-submit login confirmation did not complete within %.1fs", quick_dom_timeout)
classification = await _classify_post_submit_state(web)

7
src/kleinanzeigen_bot/model/ad_model.py
View File

@@ -11,10 +11,9 @@ from dataclasses import dataclass
from datetime import datetime # noqa: TC003 Move import into a type-checking block
from decimal import ROUND_CEILING, ROUND_HALF_UP, Decimal
from gettext import gettext as _
from typing import Annotated, Any, Final, Literal
from typing import Annotated, Any, Final, Literal, Self
from pydantic import AfterValidator, Field, field_validator, model_validator
from typing_extensions import Self
from kleinanzeigen_bot.model.config_model import AdDefaults, AutoPriceReductionConfig # noqa: TC001 Move application import into a type-checking block
from kleinanzeigen_bot.utils import dicts
@@ -446,14 +445,14 @@ class Ad(AdPartial):
price_reduction_count:int = 0
@model_validator(mode = "after")
def _validate_auto_price_config(self) -> "Ad":
def _validate_auto_price_config(self) -> Ad:
# Validate the final Ad object after merging with defaults
# This ensures the merged configuration is valid even if raw YAML had None values
_validate_auto_price_reduction_constraints(self.price, self.auto_price_reduction)
return self
@model_validator(mode = "after")
def _validate_sell_directly(self) -> "Ad":
def _validate_sell_directly(self) -> Ad:
# Direct-buy rules apply only to non-WANTED ads.
# WANTED ads with sell_directly: true are silently accepted
# (publishing skips direct-buy handling for them).

6
src/kleinanzeigen_bot/model/config_model.py
View File

@@ -48,7 +48,7 @@ class AutoPriceReductionConfig(ContextualModel):
)
@model_validator(mode = "after")
def _validate_config(self) -> "AutoPriceReductionConfig":
def _validate_config(self) -> AutoPriceReductionConfig:
if self.enabled:
if self.strategy is None:
raise ValueError(_("strategy must be specified when auto_price_reduction is enabled"))
@@ -197,7 +197,7 @@ class DownloadConfig(ContextualModel):
return trimmed
@model_validator(mode = "after")
def _validate_templates(self) -> "DownloadConfig":
def _validate_templates(self) -> DownloadConfig:
self.folder_name_template = _validate_download_template(
self.folder_name_template,
allowed_fields = _DOWNLOAD_TEMPLATE_ALLOWED_FIELDS,
@@ -487,7 +487,7 @@ class DiagnosticsConfig(ContextualModel):
return data
@model_validator(mode = "after")
def _validate_pause_requires_capture(self) -> "DiagnosticsConfig":
def _validate_pause_requires_capture(self) -> DiagnosticsConfig:
if self.pause_on_login_detection_failure and not self.capture_on.login_detection:
raise ValueError(_("pause_on_login_detection_failure requires capture_on.login_detection to be enabled"))
return self

14
src/kleinanzeigen_bot/model/update_check_state.py
View File

@@ -41,10 +41,10 @@ class UpdateCheckState(ContextualModel):
timestamp = datetime.datetime.fromisoformat(timestamp_str)
if timestamp.tzinfo is None:
# If no timezone info, assume UTC
timestamp = timestamp.replace(tzinfo = datetime.timezone.utc)
elif timestamp.tzinfo != datetime.timezone.utc:
timestamp = timestamp.replace(tzinfo = datetime.UTC)
elif timestamp.tzinfo != datetime.UTC:
# Convert to UTC if in a different timezone
timestamp = timestamp.astimezone(datetime.timezone.utc)
timestamp = timestamp.astimezone(datetime.UTC)
return timestamp
except ValueError as e:
LOG.warning("Invalid timestamp format in state file: %s", e)
@@ -114,8 +114,8 @@ class UpdateCheckState(ContextualModel):
data = self.model_dump()
if data["last_check"]:
# Ensure timestamp is in UTC before saving
if data["last_check"].tzinfo != datetime.timezone.utc:
data["last_check"] = data["last_check"].astimezone(datetime.timezone.utc)
if data["last_check"].tzinfo != datetime.UTC:
data["last_check"] = data["last_check"].astimezone(datetime.UTC)
data["last_check"] = data["last_check"].isoformat()
xdg_paths.ensure_directory(state_file.parent, "update check state directory")
dicts.save_dict(str(state_file), data)
@@ -126,7 +126,7 @@ class UpdateCheckState(ContextualModel):
def update_last_check(self) -> None:
"""Update the last check time to now in UTC."""
self.last_check = datetime.datetime.now(datetime.timezone.utc)
self.last_check = datetime.datetime.now(datetime.UTC)
def _validate_update_interval(self, interval:str) -> tuple[datetime.timedelta, bool, str]:
"""
@@ -189,7 +189,7 @@ class UpdateCheckState(ContextualModel):
LOG.warning("Falling back to default interval: 7d (latest channel). Please fix your config to avoid this warning.")
if not self.last_check:
return True
now = datetime.datetime.now(datetime.timezone.utc)
now = datetime.datetime.now(datetime.UTC)
elapsed = now - self.last_check
# Compare using integer seconds to avoid microsecond-level flakiness
return int(elapsed.total_seconds()) > int(td.total_seconds())

4
src/kleinanzeigen_bot/published_ads.py
View File

@@ -5,14 +5,14 @@
import json
from gettext import gettext as _
from typing import Any, Final, TypeAlias
from typing import Any, Final
from .utils import loggers as _loggers
from .utils import misc as _misc
from .utils.exceptions import KleinanzeigenBotError
from .utils.web_scraping_mixin import WebScrapingMixin
PublishedAd:TypeAlias = dict[str, Any]
type PublishedAd = dict[str, Any]
"""A raw published ad entry from the Kleinanzeigen manage-ads JSON API."""

3
src/kleinanzeigen_bot/publishing_form.py
View File

@@ -6,8 +6,9 @@
import json
import re
from collections.abc import Sequence
from gettext import gettext as _
from typing import Any, Final, Sequence, cast
from typing import Any, Final, cast
from .ad_description import get_ad_description
from .ad_form_helpers import (

2
src/kleinanzeigen_bot/update_checker.py
View File

@@ -28,7 +28,7 @@ logger = logging.getLogger(__name__)
class UpdateChecker:
"""Checks for updates to the bot."""
def __init__(self, config:"Config", state_file:"Path") -> None:
def __init__(self, config:Config, state_file:Path) -> None:
"""Initialize the update checker.
Args:

5
src/kleinanzeigen_bot/utils/color.py
View File

@@ -19,7 +19,10 @@ from __future__ import annotations
import os
import sys
from typing import IO, Mapping
from typing import IO, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Mapping
# ---------------------------------------------------------------------------
# Public API

8
src/kleinanzeigen_bot/utils/dicts.py
View File

@@ -8,7 +8,7 @@ from gettext import gettext as _
from importlib.resources import read_text as get_resource_as_string
from pathlib import Path
from types import ModuleType
from typing import Any, Final, TypeVar, cast, get_origin
from typing import Any, Final, cast, get_origin
from ruamel.yaml import YAML
@@ -16,10 +16,6 @@ from . import files, loggers # pylint: disable=cyclic-import
LOG:Final[loggers.Logger] = loggers.get_logger(__name__)
# https://mypy.readthedocs.io/en/stable/generics.html#generic-functions
K = TypeVar("K")
V = TypeVar("V")
def apply_defaults(
target:dict[Any, Any],
@@ -56,7 +52,7 @@ def apply_defaults(
return target
def defaultdict_to_dict(d:defaultdict[K, V]) -> dict[K, V]:
def defaultdict_to_dict[K, V](d:defaultdict[K, V]) -> dict[K, V]:
"""Recursively convert defaultdict to dict."""
result:dict[K, V] = {}
for key, value in d.items():

11
src/kleinanzeigen_bot/utils/misc.py
View File

@@ -3,16 +3,15 @@
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import asyncio, decimal, re, sys, time # isort: skip
import unicodedata
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from collections.abc import Callable, Mapping
from datetime import UTC, datetime, timedelta
from gettext import gettext as _
from typing import Any, Mapping, TypeVar
from typing import Any, TypeVar
from sanitize_filename import sanitize
from . import i18n
# https://mypy.readthedocs.io/en/stable/generics.html#generic-functions
T = TypeVar("T")
@@ -150,7 +149,7 @@ def get_attr(obj:Mapping[str, Any] | Any, key:str, default:Any | None = None) ->
def now() -> datetime:
return datetime.now(timezone.utc)
return datetime.now(UTC)
def is_frozen() -> bool:
@@ -222,7 +221,7 @@ def parse_datetime(date:datetime | str | None, *, add_timezone_if_missing:bool =
dt = date if isinstance(date, datetime) else datetime.fromisoformat(date)
if dt.tzinfo is None and add_timezone_if_missing:
dt = dt.astimezone() if use_local_timezone else dt.replace(tzinfo = timezone.utc)
dt = dt.astimezone() if use_local_timezone else dt.replace(tzinfo = UTC)
return dt

3
src/kleinanzeigen_bot/utils/pydantics.py
View File

@@ -2,11 +2,10 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
from gettext import gettext as _
from typing import Any, Literal, cast
from typing import Any, Literal, Self, cast
from pydantic import BaseModel, ValidationError
from pydantic_core import InitErrorDetails
from typing_extensions import Self
from kleinanzeigen_bot.utils.i18n import pluralize

78
src/kleinanzeigen_bot/utils/web_scraping_mixin.py
View File

@@ -5,14 +5,9 @@ import asyncio, enum, inspect, json, math, os, platform, secrets, shutil, subpro
from collections.abc import Awaitable, Callable, Coroutine, Iterable, Sequence
from gettext import gettext as _
from pathlib import Path, PureWindowsPath
from typing import Any, Final, Optional, cast
from typing import Any, Final, cast, overload
from urllib.parse import urlparse
try:
from typing import Never # type: ignore[attr-defined,unused-ignore] # mypy
except ImportError:
from typing import NoReturn as Never # Python <3.11
import nodriver, psutil # isort: skip
from nodriver.cdp import browser as cdp_browser, input_ as cdp_input # isort: skip
from typing import TYPE_CHECKING, TypeGuard
@@ -39,7 +34,6 @@ from .misc import T, ensure
if TYPE_CHECKING:
from nodriver.cdp.runtime import RemoteObject
# Crypto-secure RNG used for human-like interaction jitter (typing, timing, viewport).
# Using SystemRandom keeps behavior unpredictable and stays consistent with the module's use of
# `secrets` elsewhere (avoids the weak-PRNG lint that a plain `random` import would trigger).
@@ -323,7 +317,7 @@ class WebScrapingMixin: # noqa: PLR0904
def _get_humanization_config(self) -> HumanizationConfig:
config = getattr(self, "config", None)
if config is not None:
humanization = cast(Optional[HumanizationConfig], getattr(config, "humanization", None))
humanization = cast(HumanizationConfig | None, getattr(config, "humanization", None))
if humanization is not None:
return humanization
@@ -335,7 +329,7 @@ class WebScrapingMixin: # noqa: PLR0904
config = getattr(self, "config", None)
timeouts:TimeoutConfig | None = None
if config is not None:
timeouts = cast(Optional[TimeoutConfig], getattr(config, "timeouts", None))
timeouts = cast(TimeoutConfig | None, getattr(config, "timeouts", None))
if timeouts is not None:
return timeouts
@@ -951,7 +945,7 @@ class WebScrapingMixin: # noqa: PLR0904
timeout = _BROWSER_PROCESS_EXIT_TIMEOUT_SECONDS,
)
return
except (TimeoutError, asyncio.TimeoutError, OSError) as exc:
except (TimeoutError, OSError) as exc:
LOG.debug("Browser process did not exit cleanly: %s", exc)
try:
@@ -965,7 +959,7 @@ class WebScrapingMixin: # noqa: PLR0904
browser_process.wait(),
timeout = _BROWSER_PROCESS_KILL_TIMEOUT_SECONDS,
)
except (TimeoutError, asyncio.TimeoutError, OSError) as exc:
except (TimeoutError, OSError) as exc:
LOG.debug("Browser process could not be reaped after being killed: %s", exc)
def _close_browser_session_nowait(self) -> None:
@@ -1062,9 +1056,31 @@ class WebScrapingMixin: # noqa: PLR0904
raise AssertionError(_("Installed browser could not be detected"))
@overload
async def web_await(
self,
condition:Callable[[], T | Never | Coroutine[Any, Any, T | Never]],
condition:Callable[[], Coroutine[Any, Any, T]],
*,
timeout:int | float | None = None,
timeout_error_message:str = "",
apply_multiplier:bool = True,
) -> T:
pass # pragma: no cover
@overload
async def web_await(
self,
condition:Callable[[], T],
*,
timeout:int | float | None = None,
timeout_error_message:str = "",
apply_multiplier:bool = True,
) -> T:
pass # pragma: no cover
async def web_await(
self,
condition:Callable[[], T | Coroutine[Any, Any, T]],
*,
timeout:int | float | None = None,
timeout_error_message:str = "",
@@ -1197,7 +1213,7 @@ class WebScrapingMixin: # noqa: PLR0904
if _is_remote_object(result):
try:
# Type cast to RemoteObject for type checker
remote_obj:"RemoteObject" = result
remote_obj:RemoteObject = result
# Use the proper RemoteObject API - try to get the value directly first
if hasattr(remote_obj, "value") and remote_obj.value is not None:
@@ -1346,50 +1362,50 @@ class WebScrapingMixin: # noqa: PLR0904
match selector_type:
case By.ID:
escaped_id = selector_value.translate(METACHAR_ESCAPER)
return await self.web_await(
return cast(Element, await self.web_await(
lambda: self.page.query_selector(f"#{escaped_id}", parent),
timeout = timeout,
timeout_error_message = f"No HTML element found with ID '{selector_value}'{timeout_suffix}",
apply_multiplier = False,
)
))
case By.CLASS_NAME:
escaped_classname = selector_value.translate(METACHAR_ESCAPER)
return await self.web_await(
return cast(Element, await self.web_await(
lambda: self.page.query_selector(f".{escaped_classname}", parent),
timeout = timeout,
timeout_error_message = f"No HTML element found with CSS class '{selector_value}'{timeout_suffix}",
apply_multiplier = False,
)
))
case By.TAG_NAME:
return await self.web_await(
return cast(Element, await self.web_await(
lambda: self.page.query_selector(selector_value, parent),
timeout = timeout,
timeout_error_message = f"No HTML element found of tag <{selector_value}>{timeout_suffix}",
apply_multiplier = False,
)
))
case By.CSS_SELECTOR:
return await self.web_await(
return cast(Element, await self.web_await(
lambda: self.page.query_selector(selector_value, parent),
timeout = timeout,
timeout_error_message = f"No HTML element found using CSS selector '{selector_value}'{timeout_suffix}",
apply_multiplier = False,
)
))
case By.TEXT:
ensure(not parent, f"Specifying a parent element currently not supported with selector type: {selector_type}")
return await self.web_await(
return cast(Element, await self.web_await(
lambda: self.page.find_element_by_text(selector_value, best_match = True),
timeout = timeout,
timeout_error_message = f"No HTML element found containing text '{selector_value}'{timeout_suffix}",
apply_multiplier = False,
)
))
case By.XPATH:
ensure(not parent, f"Specifying a parent element currently not supported with selector type: {selector_type}")
return await self.web_await(
return cast(Element, await self.web_await(
lambda: self._xpath_first(selector_value),
timeout = timeout,
timeout_error_message = f"No HTML element found using XPath '{selector_value}'{timeout_suffix}",
apply_multiplier = False,
)
))
raise AssertionError(_("Unsupported selector type: %s") % selector_type)
@@ -1399,26 +1415,26 @@ class WebScrapingMixin: # noqa: PLR0904
match selector_type:
case By.CLASS_NAME:
escaped_classname = selector_value.translate(METACHAR_ESCAPER)
return await self.web_await(
return cast(list[Element], await self.web_await(
lambda: self.page.query_selector_all(f".{escaped_classname}", parent),
timeout = timeout,
timeout_error_message = f"No HTML elements found with CSS class '{selector_value}'{timeout_suffix}",
apply_multiplier = False,
)
))
case By.CSS_SELECTOR:
return await self.web_await(
return cast(list[Element], await self.web_await(
lambda: self.page.query_selector_all(selector_value, parent),
timeout = timeout,
timeout_error_message = f"No HTML elements found using CSS selector '{selector_value}'{timeout_suffix}",
apply_multiplier = False,
)
))
case By.TAG_NAME:
return await self.web_await(
return cast(list[Element], await self.web_await(
lambda: self.page.query_selector_all(selector_value, parent),
timeout = timeout,
timeout_error_message = f"No HTML elements found of tag <{selector_value}>{timeout_suffix}",
apply_multiplier = False,
)
))
case By.TEXT:
ensure(not parent, f"Specifying a parent element currently not supported with selector type: {selector_type}")
return await self.web_await(

3
tests/smoke/test_smoke_health.py
View File

@@ -12,9 +12,10 @@ import json
import logging
import os
import re
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping
from typing import Any
from unittest.mock import patch
import pytest

4
tests/unit/test_ad_status.py
View File

@@ -7,7 +7,7 @@ from __future__ import annotations
import copy
import re
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from typing import Any
from unittest.mock import AsyncMock, patch
@@ -37,7 +37,7 @@ def _strip_ansi(value:str) -> str:
return _ANSI_RE.sub("", value)
_TZ = timezone.utc
_TZ = UTC
def _now() -> datetime:

24
tests/unit/test_extract.py
View File

@@ -453,7 +453,7 @@ class TestAdExtractorNavigation:
"""Test navigation to ad page using an ID."""
ad_id = 12345
page_mock = AsyncMock()
page_mock.url = "https://www.kleinanzeigen.de/s-anzeige/test/{0}".format(ad_id)
page_mock.url = f"https://www.kleinanzeigen.de/s-anzeige/test/{ad_id}"
with (
patch.object(test_extractor, "page", page_mock),
@@ -463,7 +463,7 @@ class TestAdExtractorNavigation:
):
result = await test_extractor.navigate_to_ad_page(ad_id)
assert result is True
mock_web_open.assert_called_with("https://www.kleinanzeigen.de/s-suchanfrage.html?keywords={0}".format(ad_id))
mock_web_open.assert_called_with(f"https://www.kleinanzeigen.de/s-suchanfrage.html?keywords={ad_id}")
mock_web_click.assert_awaited_once_with(By.CLASS_NAME, "mfp-close")
@pytest.mark.asyncio
@@ -2698,7 +2698,7 @@ class TestAdExtractorDownload:
assert calls == 1
assert path.exists()
def test_handle_rmtree_onerror_adds_write_bit_preserving_other_mode_bits_on_windows(self) -> None:
def test_handle_rmtree_onexc_adds_write_bit_preserving_other_mode_bits_on_windows(self) -> None:
path = "C:/Temp/readonly.txt"
retry_func = MagicMock()
stat_result = MagicMock(st_mode = 0o555)
@@ -2708,13 +2708,13 @@ class TestAdExtractorDownload:
patch("kleinanzeigen_bot.extract.os.stat", return_value = stat_result) as mock_stat,
patch("kleinanzeigen_bot.extract.os.chmod") as mock_chmod,
):
extract_module._handle_rmtree_onerror(retry_func, path, (PermissionError, PermissionError("busy"), None))
extract_module._handle_rmtree_onexc(retry_func, path, PermissionError("busy"))
mock_stat.assert_any_call(path)
mock_chmod.assert_called_once_with(path, 0o555 | stat.S_IWRITE)
retry_func.assert_called_once_with(path)
def test_handle_rmtree_onerror_skips_chmod_on_posix(self, tmp_path:Path) -> None:
def test_handle_rmtree_onexc_skips_chmod_on_posix(self, tmp_path:Path) -> None:
path = str(tmp_path / "readonly.txt")
retry_func = MagicMock()
@@ -2723,12 +2723,12 @@ class TestAdExtractorDownload:
patch("kleinanzeigen_bot.extract.os.stat"),
patch("kleinanzeigen_bot.extract.os.chmod") as mock_chmod,
):
extract_module._handle_rmtree_onerror(retry_func, path, (PermissionError, PermissionError("busy"), None))
extract_module._handle_rmtree_onexc(retry_func, path, PermissionError("busy"))
mock_chmod.assert_not_called()
retry_func.assert_called_once_with(path)
def test_handle_rmtree_onerror_ignores_chmod_failures_on_windows(self) -> None:
def test_handle_rmtree_onexc_ignores_chmod_failures_on_windows(self) -> None:
path = "C:/Temp/readonly.txt"
retry_func = MagicMock()
stat_result = MagicMock(st_mode = 0o555)
@@ -2738,11 +2738,11 @@ class TestAdExtractorDownload:
patch("kleinanzeigen_bot.extract.os.stat", return_value = stat_result),
patch("kleinanzeigen_bot.extract.os.chmod", side_effect = OSError("chmod failed")),
):
extract_module._handle_rmtree_onerror(retry_func, path, (PermissionError, PermissionError("busy"), None))
extract_module._handle_rmtree_onexc(retry_func, path, PermissionError("busy"))
retry_func.assert_called_once_with(path)
def test_handle_rmtree_onerror_continues_when_stat_fails_on_windows(self) -> None:
def test_handle_rmtree_onexc_continues_when_stat_fails_on_windows(self) -> None:
path = "C:/Temp/readonly.txt"
retry_func = MagicMock()
@@ -2751,17 +2751,17 @@ class TestAdExtractorDownload:
patch("kleinanzeigen_bot.extract.os.stat", side_effect = OSError("stat failed")),
patch("kleinanzeigen_bot.extract.os.chmod") as mock_chmod,
):
extract_module._handle_rmtree_onerror(retry_func, path, (PermissionError, PermissionError("busy"), None))
extract_module._handle_rmtree_onexc(retry_func, path, PermissionError("busy"))
mock_chmod.assert_not_called()
retry_func.assert_called_once_with(path)
def test_handle_rmtree_onerror_raises_for_non_retryable_error(self, tmp_path:Path) -> None:
def test_handle_rmtree_onexc_raises_for_non_retryable_error(self, tmp_path:Path) -> None:
path = str(tmp_path / "file.txt")
retry_func = MagicMock()
with pytest.raises(OSError, match = "bad"):
extract_module._handle_rmtree_onerror(retry_func, path, (OSError, OSError(errno.EINVAL, "bad"), None))
extract_module._handle_rmtree_onexc(retry_func, path, OSError(errno.EINVAL, "bad"))
retry_func.assert_not_called()

17
tests/unit/test_login_flow.py
View File

@@ -1,7 +1,6 @@
# 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
import inspect
from collections.abc import Callable
from pathlib import Path
@@ -760,7 +759,7 @@ class TestKleinanzeigenBotAuthentication:
"""Sleep fallback should run when bounded login check times out."""
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]) as mock_wait,
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError) as mock_is_logged_in,
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError) as mock_is_logged_in,
patch.object(test_bot, "web_sleep", new_callable = AsyncMock) as mock_sleep,
patch(
"kleinanzeigen_bot.login_flow._classify_post_submit_state",
@@ -1449,7 +1448,7 @@ class TestClassifyPostSubmitState:
"""TimeoutError uses coarse labels and sanitised URL."""
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError),
patch.object(test_bot, "web_sleep", new_callable = AsyncMock),
patch(
"kleinanzeigen_bot.login_flow._classify_post_submit_state",
@@ -1495,7 +1494,7 @@ class TestClassifyPostSubmitState:
"""TimeoutError prefix preserved when URL retrieval fails."""
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError),
patch.object(test_bot, "web_sleep", new_callable = AsyncMock),
patch(
"kleinanzeigen_bot.login_flow._classify_post_submit_state",
@@ -1528,7 +1527,7 @@ class TestClassifyPostSubmitState:
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = _call_predicate),
patch("kleinanzeigen_bot.login_flow.current_page_url", side_effect = RuntimeError("boom")),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError),
patch.object(test_bot, "web_sleep", new_callable = AsyncMock),
patch("kleinanzeigen_bot.login_flow._classify_post_submit_state", new_callable = AsyncMock, return_value = "STILL_ON_PASSWORD_PAGE"),
pytest.raises(TimeoutError, match = "Auth0 post-submit verification remained inconclusive"),
@@ -1556,7 +1555,7 @@ class TestClassifyPostSubmitState:
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError),
patch.object(test_bot, "web_sleep", new_callable = AsyncMock),
patch(
"kleinanzeigen_bot.login_flow._classify_post_submit_state",
@@ -1599,7 +1598,7 @@ class TestClassifyPostSubmitState:
"""Disabled diagnostics config (None) skips capture and raises original TimeoutError."""
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError),
patch.object(test_bot, "web_sleep", new_callable = AsyncMock),
patch(
"kleinanzeigen_bot.login_flow._classify_post_submit_state",
@@ -1630,7 +1629,7 @@ class TestClassifyPostSubmitState:
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError),
patch.object(test_bot, "web_sleep", new_callable = AsyncMock),
patch(
"kleinanzeigen_bot.login_flow._classify_post_submit_state",
@@ -1669,7 +1668,7 @@ class TestClassifyPostSubmitState:
with (
patch.object(test_bot, "web_await", new_callable = AsyncMock, side_effect = [TimeoutError()]),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = asyncio.TimeoutError),
patch("kleinanzeigen_bot.login_flow.is_logged_in", new_callable = AsyncMock, side_effect = TimeoutError),
patch.object(test_bot, "web_sleep", new_callable = AsyncMock),
patch(
"kleinanzeigen_bot.login_flow._classify_post_submit_state",

2
tests/unit/test_net.py
View File

@@ -7,7 +7,7 @@ Covers port availability checking functionality.
"""
import socket
from typing import Generator
from collections.abc import Generator
from unittest.mock import MagicMock, patch
import pytest

22
tests/unit/test_price_reduction.py
View File

@@ -2,7 +2,7 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import logging
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import Any, Protocol, runtime_checkable
@@ -347,7 +347,7 @@ def test_apply_auto_price_reduction_waits_when_reduction_already_applied(
def test_apply_auto_price_reduction_respects_day_delay(
monkeypatch:pytest.MonkeyPatch, caplog:pytest.LogCaptureFixture, apply_auto_price_reduction:_ApplyAutoPriceReduction
) -> None:
reference = datetime(2025, 1, 1, tzinfo = timezone.utc)
reference = datetime(2025, 1, 1, tzinfo = UTC)
ad_cfg = SimpleNamespace(
price = 150,
auto_price_reduction = AutoPriceReductionConfig(
@@ -377,7 +377,7 @@ def test_apply_auto_price_reduction_respects_day_delay(
@pytest.mark.unit
def test_apply_auto_price_reduction_runs_after_delays(monkeypatch:pytest.MonkeyPatch, apply_auto_price_reduction:_ApplyAutoPriceReduction) -> None:
reference = datetime(2025, 1, 1, tzinfo = timezone.utc)
reference = datetime(2025, 1, 1, tzinfo = UTC)
ad_cfg = SimpleNamespace(
price = 120,
auto_price_reduction = AutoPriceReductionConfig(
@@ -607,7 +607,7 @@ def test_apply_modify_mode_applies_reduction_when_on_update_true_and_day_delay_s
delay_reposts must be ignored in MODIFY mode (repost count does not change).
"""
reference = datetime(2025, 1, 1, tzinfo = timezone.utc)
reference = datetime(2025, 1, 1, tzinfo = UTC)
ad_cfg = SimpleNamespace(
price = 200,
# delay_reposts=5 would normally block reduction, but MODIFY mode ignores it
@@ -632,7 +632,7 @@ def test_apply_modify_mode_skips_new_cycle_when_day_delay_not_satisfied(
apply_auto_price_reduction:_ApplyAutoPriceReduction,
) -> None:
"""MODIFY mode with on_update=true does NOT apply a new cycle when day delay is not met."""
reference = datetime(2025, 1, 1, tzinfo = timezone.utc)
reference = datetime(2025, 1, 1, tzinfo = UTC)
ad_cfg = SimpleNamespace(
price = 200,
auto_price_reduction = _price_cfg(on_update = True, amount = 25, delay_days = 3),
@@ -905,7 +905,7 @@ def test_is_price_reduction_due_returns_true_when_eligible_real(
monkeypatch:pytest.MonkeyPatch,
) -> None:
"""is_auto_price_reduction_due returns True when eligible (exercises real evaluate_auto_price_reduction)."""
now_dt = datetime(2024, 6, 1, tzinfo = timezone.utc)
now_dt = datetime(2024, 6, 1, tzinfo = UTC)
monkeypatch.setattr("kleinanzeigen_bot.utils.misc.now", lambda: now_dt)
price_cfg = _price_cfg(on_update = True, delay_days = 0)
@@ -914,7 +914,7 @@ def test_is_price_reduction_due_returns_true_when_eligible_real(
price = 100,
auto_price_reduction = price_cfg,
price_reduction_count = 0,
updated_on = datetime(2024, 1, 1, tzinfo = timezone.utc),
updated_on = datetime(2024, 1, 1, tzinfo = UTC),
repost_count = 5,
)
@@ -926,7 +926,7 @@ def test_is_price_reduction_due_returns_false_when_delay_not_satisfied_real(
monkeypatch:pytest.MonkeyPatch,
) -> None:
"""is_auto_price_reduction_due returns False when day-delay is not satisfied (exercises real evaluate_auto_price_reduction)."""
now_dt = datetime(2024, 1, 1, 12, 0, 0, tzinfo = timezone.utc)
now_dt = datetime(2024, 1, 1, 12, 0, 0, tzinfo = UTC)
monkeypatch.setattr("kleinanzeigen_bot.utils.misc.now", lambda: now_dt)
price_cfg = _price_cfg(on_update = True, delay_days = 1)
@@ -935,7 +935,7 @@ def test_is_price_reduction_due_returns_false_when_delay_not_satisfied_real(
price = 100,
auto_price_reduction = price_cfg,
price_reduction_count = 0,
updated_on = datetime(2024, 1, 1, tzinfo = timezone.utc),
updated_on = datetime(2024, 1, 1, tzinfo = UTC),
repost_count = 5,
)
@@ -947,7 +947,7 @@ def test_is_price_reduction_due_returns_false_when_on_update_disabled_real(
monkeypatch:pytest.MonkeyPatch,
) -> None:
"""is_auto_price_reduction_due returns False when on_update is False (exercises real evaluate_auto_price_reduction)."""
now_dt = datetime(2024, 6, 1, tzinfo = timezone.utc)
now_dt = datetime(2024, 6, 1, tzinfo = UTC)
monkeypatch.setattr("kleinanzeigen_bot.utils.misc.now", lambda: now_dt)
price_cfg = _price_cfg(on_update = False, delay_days = 0)
@@ -956,7 +956,7 @@ def test_is_price_reduction_due_returns_false_when_on_update_disabled_real(
price = 100,
auto_price_reduction = price_cfg,
price_reduction_count = 0,
updated_on = datetime(2024, 1, 1, tzinfo = timezone.utc),
updated_on = datetime(2024, 1, 1, tzinfo = UTC),
repost_count = 5,
)

4
tests/unit/test_publishing_form.py
View File

@@ -6,10 +6,10 @@
import asyncio
import json
import logging
from collections.abc import Callable
from collections.abc import Awaitable, Callable, Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Awaitable, Iterator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest

8
tests/unit/test_publishing_persistence.py
View File

@@ -3,7 +3,7 @@
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
"""Tests for publishing persistence functionality."""
from datetime import datetime, timezone
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
@@ -167,7 +167,7 @@ class TestPersistPublishedAdTimestamps:
ad.id = 12345
ad_cfg_orig = self._make_ad_cfg_orig()
cfg = _make_config()
published_at = datetime(2026, 7, 4, 12, 34, 56, tzinfo = timezone.utc)
published_at = datetime(2026, 7, 4, 12, 34, 56, tzinfo = UTC)
with (
patch("kleinanzeigen_bot.local_path_renaming.rename_referenced_local_image_files_after_id_change",
@@ -198,10 +198,10 @@ class TestPersistPublishedAdTimestamps:
"""Updating an existing ad keeps the original created_on value."""
ad = _make_min_ad()
ad.id = 12345
ad.created_on = datetime(2024, 1, 2, 3, 4, 5, tzinfo = timezone.utc)
ad.created_on = datetime(2024, 1, 2, 3, 4, 5, tzinfo = UTC)
ad_cfg_orig = self._make_ad_cfg_orig(created_on = "2024-01-02T03:04:05")
cfg = _make_config()
updated_at = datetime(2026, 7, 4, 12, 34, 56, tzinfo = timezone.utc)
updated_at = datetime(2026, 7, 4, 12, 34, 56, tzinfo = UTC)
with (
patch("kleinanzeigen_bot.local_path_renaming.rename_referenced_local_image_files_after_id_change",

3
tests/unit/test_pydantics.py
View File

@@ -6,12 +6,11 @@
Covers ContextualValidationError, ContextualModel, and format_validation_error.
"""
from typing import Any, TypedDict, cast
from typing import Any, NotRequired, TypedDict, cast
import pytest
from pydantic import BaseModel, ValidationError
from pydantic_core import ErrorDetails as PydanticErrorDetails
from typing_extensions import NotRequired
from kleinanzeigen_bot.utils.pydantics import (
ContextualModel,

2
tests/unit/test_translations.py
View File

@@ -95,7 +95,7 @@ def _extract_log_messages(file_path:str, exclude_debug:bool = False) -> MessageD
Returns:
Dictionary mapping function names to their messages
"""
with open(file_path, "r", encoding = "utf-8") as file:
with open(file_path, encoding = "utf-8") as file:
tree = ast.parse(file.read(), filename = file_path)
# Add parent references for context tracking

83
tests/unit/test_update_checker.py
View File

@@ -6,7 +6,7 @@ from __future__ import annotations
import json
import logging
from datetime import datetime, timedelta, timezone, tzinfo
from datetime import UTC, datetime, timedelta, timezone, tzinfo
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import MagicMock, patch
@@ -29,13 +29,13 @@ def _freeze_update_state_datetime(monkeypatch:pytest.MonkeyPatch, fixed_now:date
class FixedDateTime(datetime):
@classmethod
def now(cls, tz:tzinfo | None = None) -> "FixedDateTime":
def now(cls, tz:tzinfo | None = None) -> FixedDateTime:
base = fixed_now.replace(tzinfo = None) if tz is None else fixed_now.astimezone(tz)
return cls(base.year, base.month, base.day, base.hour, base.minute, base.second, base.microsecond, tzinfo = base.tzinfo)
@classmethod
def utcnow(cls) -> "FixedDateTime":
base = fixed_now.astimezone(timezone.utc).replace(tzinfo = None)
def utcnow(cls) -> FixedDateTime:
base = fixed_now.astimezone(UTC).replace(tzinfo = None)
return cls(base.year, base.month, base.day, base.hour, base.minute, base.second, base.microsecond)
datetime_module = getattr(update_check_state_module, "datetime")
@@ -72,9 +72,9 @@ class TestUpdateChecker:
with patch("requests.get", return_value = MagicMock(json = lambda: {"sha": "e7a3d46", "commit": {"author": {"date": "2025-05-18T00:00:00Z"}}})):
commit_hash, commit_date = checker._resolve_commitish("latest")
assert commit_hash == "e7a3d46"
assert commit_date == datetime(2025, 5, 18, tzinfo = timezone.utc)
assert commit_date == datetime(2025, 5, 18, tzinfo = UTC)
def test_request_timeout_uses_config(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_request_timeout_uses_config(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Ensure HTTP calls honor the timeout configuration."""
config.timeouts.multiplier = 1.5
checker = UpdateChecker(config, state_file)
@@ -86,7 +86,7 @@ class TestUpdateChecker:
expected_timeout = config.timeouts.effective("update_check")
assert mock_get.call_args.kwargs["timeout"] == expected_timeout
def test_resolve_commitish_no_commit(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_resolve_commitish_no_commit(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Test resolving a commit-ish when the API returns no commit data."""
checker = UpdateChecker(config, state_file)
mocker.patch("requests.get", return_value = mocker.Mock(json = lambda: {"sha": "abc"}))
@@ -134,7 +134,7 @@ class TestUpdateChecker:
checker.check_for_updates() # Should not raise exception
def test_check_for_updates_latest_prerelease_warning(
self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture
self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture
) -> None:
"""Test that the update checker warns when latest points to a prerelease."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
@@ -149,7 +149,7 @@ class TestUpdateChecker:
expected = "Latest release from GitHub is a prerelease, but 'latest' channel expects a stable release."
assert any(expected in r.getMessage() for r in caplog.records)
def test_check_for_updates_ahead(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
def test_check_for_updates_ahead(self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker correctly identifies when the local version is ahead of the latest release."""
caplog.set_level("INFO", logger = "kleinanzeigen_bot.update_checker")
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
@@ -157,7 +157,7 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 16, tzinfo = timezone.utc))],
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = UTC)), ("e7a3d46", datetime(2025, 5, 16, tzinfo = UTC))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": False}))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
@@ -168,7 +168,7 @@ class TestUpdateChecker:
msgs = [r.getMessage() for r in caplog.records if r.levelno >= logging.INFO]
assert any("different commit" in m and "channel 'latest'" in m for m in msgs)
def test_check_for_updates_preview(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
def test_check_for_updates_preview(self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker correctly handles preview releases."""
caplog.set_level("INFO", logger = "kleinanzeigen_bot.update_checker")
config.update_check.channel = "preview"
@@ -177,7 +177,7 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 16, tzinfo = timezone.utc))],
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = UTC)), ("e7a3d46", datetime(2025, 5, 16, tzinfo = UTC))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: [{"tag_name": "preview", "prerelease": True, "draft": False}]))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
@@ -189,7 +189,7 @@ class TestUpdateChecker:
assert any("different commit" in m and "channel 'preview'" in m for m in msgs)
def test_check_for_updates_preview_missing_prerelease(
self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture
self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture
) -> None:
"""Test that the update checker warns when no preview prerelease is available."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
@@ -204,7 +204,7 @@ class TestUpdateChecker:
assert any("No prerelease found for 'preview' channel." in r.getMessage() for r in caplog.records)
def test_check_for_updates_behind(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
def test_check_for_updates_behind(self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker correctly identifies when the local version is behind the latest release."""
caplog.set_level("INFO", logger = "kleinanzeigen_bot.update_checker")
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
@@ -212,7 +212,7 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [("fb00f11", datetime(2025, 5, 16, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 18, tzinfo = timezone.utc))],
side_effect = [("fb00f11", datetime(2025, 5, 16, tzinfo = UTC)), ("e7a3d46", datetime(2025, 5, 18, tzinfo = UTC))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": False}))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
@@ -223,7 +223,7 @@ class TestUpdateChecker:
msgs = [r.getMessage() for r in caplog.records if r.levelno >= logging.INFO]
assert any("new version is available" in m and "channel: latest" in m for m in msgs)
def test_check_for_updates_logs_release_notes(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
def test_check_for_updates_logs_release_notes(self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture) -> None:
"""Test that release notes are logged when present."""
caplog.set_level("INFO", logger = "kleinanzeigen_bot.update_checker")
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
@@ -231,7 +231,7 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [("fb00f11", datetime(2025, 5, 16, tzinfo = timezone.utc)), ("e7a3d46", datetime(2025, 5, 18, tzinfo = timezone.utc))],
side_effect = [("fb00f11", datetime(2025, 5, 16, tzinfo = UTC)), ("e7a3d46", datetime(2025, 5, 18, tzinfo = UTC))],
)
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
mocker.patch.object(
@@ -248,7 +248,7 @@ class TestUpdateChecker:
assert any("Release notes:\nRelease notes here" in r.getMessage() for r in caplog.records)
def test_check_for_updates_same(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
def test_check_for_updates_same(self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker correctly identifies when the local version is the same as the latest release."""
caplog.set_level("INFO", logger = "kleinanzeigen_bot.update_checker")
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
@@ -256,7 +256,7 @@ class TestUpdateChecker:
mocker.patch.object(
UpdateChecker,
"_resolve_commitish",
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc)), ("fb00f11", datetime(2025, 5, 18, tzinfo = timezone.utc))],
side_effect = [("fb00f11", datetime(2025, 5, 18, tzinfo = UTC)), ("fb00f11", datetime(2025, 5, 18, tzinfo = UTC))],
)
mocker.patch.object(requests, "get", return_value = mocker.Mock(json = lambda: {"tag_name": "latest", "prerelease": False}))
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
@@ -267,7 +267,7 @@ class TestUpdateChecker:
msgs = [r.getMessage() for r in caplog.records if r.levelno >= logging.INFO]
assert any("on the latest version" in m and "channel latest" in m for m in msgs)
def test_check_for_updates_unknown_channel(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
def test_check_for_updates_unknown_channel(self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture) -> None:
"""Test that the update checker warns on unknown update channels."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
cast(Any, config.update_check).channel = "unknown"
@@ -320,7 +320,7 @@ class TestUpdateChecker:
def test_update_check_state_interval_units(self, monkeypatch:pytest.MonkeyPatch) -> None:
"""Test that different interval units are handled correctly."""
state = UpdateCheckState()
fixed_now = datetime(2025, 1, 15, 8, 0, tzinfo = timezone.utc)
fixed_now = datetime(2025, 1, 15, 8, 0, tzinfo = UTC)
_freeze_update_state_datetime(monkeypatch, fixed_now)
now = fixed_now
@@ -369,7 +369,7 @@ class TestUpdateChecker:
def test_update_check_state_interval_validation(self, monkeypatch:pytest.MonkeyPatch) -> None:
"""Test that interval validation works correctly."""
state = UpdateCheckState()
fixed_now = datetime(2025, 1, 1, 12, 0, tzinfo = timezone.utc)
fixed_now = datetime(2025, 1, 1, 12, 0, tzinfo = UTC)
_freeze_update_state_datetime(monkeypatch, fixed_now)
@@ -442,7 +442,7 @@ class TestUpdateChecker:
state = UpdateCheckState.load(state_file)
assert state.last_check is None
def test_resolve_commitish_no_author(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_resolve_commitish_no_author(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Test resolving a commit-ish when the API returns no author key."""
checker = UpdateChecker(config, state_file)
mocker.patch("requests.get", return_value = mocker.Mock(json = lambda: {"sha": "abc", "commit": {}}))
@@ -450,7 +450,7 @@ class TestUpdateChecker:
assert commit_hash == "abc"
assert commit_date is None
def test_resolve_commitish_no_date(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_resolve_commitish_no_date(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Test resolving a commit-ish when the API returns no date key."""
checker = UpdateChecker(config, state_file)
mocker.patch("requests.get", return_value = mocker.Mock(json = lambda: {"sha": "abc", "commit": {"author": {}}}))
@@ -458,7 +458,7 @@ class TestUpdateChecker:
assert commit_hash == "abc"
assert commit_date is None
def test_resolve_commitish_list_instead_of_dict(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_resolve_commitish_list_instead_of_dict(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Test resolving a commit-ish when the API returns a list instead of dict."""
checker = UpdateChecker(config, state_file)
mocker.patch("requests.get", return_value = mocker.Mock(json = list))
@@ -466,7 +466,7 @@ class TestUpdateChecker:
assert commit_hash is None
assert commit_date is None
def test_check_for_updates_missing_release_commitish(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_check_for_updates_missing_release_commitish(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Test check_for_updates handles missing release commit-ish."""
checker = UpdateChecker(config, state_file)
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
@@ -475,21 +475,26 @@ class TestUpdateChecker:
mocker.patch("requests.get", return_value = mocker.Mock(json = lambda: {"prerelease": False}))
checker.check_for_updates() # Should not raise
def test_check_for_updates_no_releases_empty(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_check_for_updates_no_releases_empty(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Test check_for_updates handles no releases found (API returns empty list)."""
config.update_check.channel = "preview"
checker = UpdateChecker(config, state_file)
mocker.patch("requests.get", return_value = mocker.Mock(json = list))
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
mock_get = mocker.patch("requests.get", return_value = mocker.Mock(json = list))
checker.check_for_updates() # Should not raise
mock_get.assert_called_once()
def test_check_for_updates_no_commit_hash_extracted(self, config:Config, state_file:Path, mocker:"MockerFixture") -> None:
def test_check_for_updates_no_commit_hash_extracted(self, config:Config, state_file:Path, mocker:MockerFixture) -> None:
"""Test check_for_updates handles no commit hash extracted."""
checker = UpdateChecker(config, state_file)
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025")
mocker.patch.object(UpdateCheckState, "should_check", return_value = True)
mock_get = mocker.patch("requests.get")
checker.check_for_updates() # Should not raise
mock_get.assert_not_called()
def test_check_for_updates_no_commit_dates(self, config:Config, state_file:Path, mocker:"MockerFixture", caplog:pytest.LogCaptureFixture) -> None:
def test_check_for_updates_no_commit_dates(self, config:Config, state_file:Path, mocker:MockerFixture, caplog:pytest.LogCaptureFixture) -> None:
"""Test check_for_updates logs warning if commit dates cannot be determined."""
caplog.set_level("WARNING", logger = "kleinanzeigen_bot.update_checker")
mocker.patch.object(UpdateChecker, "get_local_version", return_value = "2025+fb00f11")
@@ -505,7 +510,7 @@ class TestUpdateChecker:
def test_update_check_state_version_tracking(self, state_file:Path) -> None:
"""Test that version tracking works correctly."""
# Create a state with version 0 (old format)
state_file.write_text(json.dumps({"last_check": datetime.now(timezone.utc).isoformat()}), encoding = "utf-8")
state_file.write_text(json.dumps({"last_check": datetime.now(UTC).isoformat()}), encoding = "utf-8")
# Load the state - should migrate to version 1
state = UpdateCheckState.load(state_file)
@@ -521,7 +526,7 @@ class TestUpdateChecker:
def test_update_check_state_migration(self, state_file:Path) -> None:
"""Test that state migration works correctly."""
# Create a state with version 0 (old format)
old_time = datetime.now(timezone.utc)
old_time = datetime.now(UTC)
state_file.write_text(json.dumps({"last_check": old_time.isoformat()}), encoding = "utf-8")
# Load the state - should migrate to version 1
@@ -533,15 +538,15 @@ class TestUpdateChecker:
state.save(state_file)
# Verify the saved file has the new version
with open(state_file, "r", encoding = "utf-8") as f:
with open(state_file, encoding = "utf-8") as f:
data = json.load(f)
assert data["version"] == 1
assert data["last_check"] == old_time.isoformat()
def test_update_check_state_save_errors(self, state_file:Path, mocker:"MockerFixture") -> None:
def test_update_check_state_save_errors(self, state_file:Path, mocker:MockerFixture) -> None:
"""Test that save errors are handled gracefully."""
state = UpdateCheckState()
state.last_check = datetime.now(timezone.utc)
state.last_check = datetime.now(UTC)
# Test permission error
mocker.patch("kleinanzeigen_bot.utils.dicts.save_dict", side_effect = PermissionError)
@@ -557,7 +562,7 @@ class TestUpdateChecker:
state_file.write_text(json.dumps({"version": 1, "last_check": "2024-03-20T12:00:00"}), encoding = "utf-8")
state = UpdateCheckState.load(state_file)
assert state.last_check is not None
assert state.last_check.tzinfo == timezone.utc
assert state.last_check.tzinfo == UTC
assert state.last_check.hour == 12
# Test loading timestamp with different timezone (should convert to UTC)
@@ -572,14 +577,14 @@ class TestUpdateChecker:
)
state = UpdateCheckState.load(state_file)
assert state.last_check is not None
assert state.last_check.tzinfo == timezone.utc
assert state.last_check.tzinfo == UTC
assert state.last_check.hour == 10 # Converted to UTC
# Test saving timestamp (should always be in UTC)
state = UpdateCheckState()
state.last_check = datetime(2024, 3, 20, 12, 0, tzinfo = timezone(timedelta(hours = 2)))
state.save(state_file)
with open(state_file, "r", encoding = "utf-8") as f:
with open(state_file, encoding = "utf-8") as f:
data = json.load(f)
assert data["last_check"] == "2024-03-20T10:00:00+00:00" # Converted to UTC
@@ -604,7 +609,7 @@ class TestUpdateChecker:
def test_should_check_fallback_to_default_interval(self, caplog:pytest.LogCaptureFixture) -> None:
"""Test that should_check falls back to default interval and logs a warning for invalid/too short/too long/zero intervals and unsupported units."""
state = UpdateCheckState()
now = datetime.now(timezone.utc)
now = datetime.now(UTC)
state.last_check = now - timedelta(days = 2)
# Invalid format (unsupported unit)

4
tests/unit/test_utils_misc.py
View File

@@ -4,7 +4,7 @@
import asyncio
import decimal
import sys
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
import pytest
from sanitize_filename import sanitize
@@ -51,7 +51,7 @@ def test_parse_datetime_none_returns_none() -> None:
def test_parse_datetime_from_datetime() -> None:
dt = datetime(2020, 1, 1, 0, 0, tzinfo = timezone.utc)
dt = datetime(2020, 1, 1, 0, 0, tzinfo = UTC)
assert misc.parse_datetime(dt, add_timezone_if_missing = False) == dt

6
tests/unit/test_version.py
View File

@@ -2,7 +2,7 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
import importlib
from datetime import datetime, timezone
from datetime import UTC, datetime
from unittest.mock import MagicMock, patch
import pytest
@@ -16,7 +16,7 @@ class TestVersion:
monkeypatch.setenv("GIT_COMMIT_HASH", "abc1234")
with patch("version.shutil.which") as which_mock, patch("version.subprocess.run") as run_mock:
assert version.get_version() == f"{datetime.now(timezone.utc).year}+abc1234"
assert version.get_version() == f"{datetime.now(UTC).year}+abc1234"
which_mock.assert_not_called()
run_mock.assert_not_called()
@@ -27,7 +27,7 @@ class TestVersion:
result = MagicMock(stdout = "deadbee\n")
with patch("version.shutil.which", return_value = "/usr/bin/git") as which_mock, patch("version.subprocess.run", return_value = result) as run_mock:
assert version.get_version() == f"{datetime.now(timezone.utc).year}+deadbee"
assert version.get_version() == f"{datetime.now(UTC).year}+deadbee"
which_mock.assert_called_once_with("git")
run_mock.assert_called_once_with(

1
tests/unit/test_web_scraping_mixin.py
View File

@@ -727,6 +727,7 @@ class TestSelectorTimeoutMessages:
@pytest.mark.parametrize(
("selector_type", "selector_value", "expected_message"),
[
(By.CLASS_NAME, "hero", "No HTML element found with CSS class 'hero' within 2.0 seconds."),
(By.TAG_NAME, "section", "No HTML element found of tag <section> within 2.0 seconds."),
(By.CSS_SELECTOR, ".hero", "No HTML element found using CSS selector '.hero' within 2.0 seconds."),
(By.TEXT, "Submit", "No HTML element found containing text 'Submit' within 2.0 seconds."),