From 884d6437f07d36cb004b00bad3a1367f78f8af0b Mon Sep 17 00:00:00 2001 From: Patrik Jansson Date: Sun, 25 Oct 2026 22:44:28 +0100 Subject: [PATCH] chore: enforce type annotations on every function - Pyright `reportMissingParameterType` is now an error; the ~600 missing parameter types are filled in - `tests/test_annotations.py` requires a return type on every function (Pyright has no rule for it) - a few signatures were tightened or made honest (`Optional`) where the new types exposed a mismatch - CLAUDE.md states the rule and how it is enforced Co-Authored-By: Claude Sonnet 5.5 --- CLAUDE.md | 2 +- main.py | 63 +++--- modules/asset_downloader.py | 24 +-- modules/config_manager.py | 59 +++--- modules/daemon_lock.py | 8 +- modules/db_manager.py | 258 +++++++++++++------------ modules/log_files.py | 4 +- modules/mailbox_listener.py | 13 +- modules/queue_reports.py | 31 +-- modules/queue_worker.py | 33 ++-- pyrightconfig.json | 3 +- tests/test_annotations.py | 53 +++++ tests/test_asset_downloader_helpers.py | 3 +- tests/test_autostart.py | 23 +-- tests/test_backup_manager.py | 9 +- tests/test_cli_dispatch.py | 13 +- tests/test_cli_unused_options.py | 5 +- tests/test_config_getters.py | 9 +- tests/test_connection_tests.py | 25 +-- tests/test_daemon_control.py | 16 +- tests/test_daemon_stop.py | 5 +- tests/test_download_release.py | 50 ++--- tests/test_env_manager.py | 3 +- tests/test_folder_check_trigger.py | 13 +- tests/test_folder_naming_and_moves.py | 8 +- tests/test_form_helpers_and_logger.py | 5 +- tests/test_gui_daemon.py | 10 +- tests/test_gui_doctor.py | 7 +- tests/test_gui_logic.py | 13 +- tests/test_gui_state.py | 3 +- tests/test_gui_theme.py | 8 +- tests/test_gui_tray.py | 20 +- tests/test_idle_polling.py | 17 +- tests/test_job_queue.py | 9 +- tests/test_mailbox_imap.py | 21 +- tests/test_mapping_lookups.py | 10 +- tests/test_mapping_manager.py | 7 +- tests/test_mapping_upgrade.py | 3 +- tests/test_new_entry_defaults.py | 3 +- tests/test_pending_jobs_view.py | 5 +- tests/test_polling_efficiency.py | 5 +- tests/test_queue_counts.py | 3 +- tests/test_queue_reports.py | 6 +- tests/test_queue_worker_flow.py | 36 ++-- tests/test_stats.py | 16 +- tests/test_status_tabs.py | 21 +- tests/test_terminal_log.py | 10 +- 47 files changed, 533 insertions(+), 438 deletions(-) create mode 100644 tests/test_annotations.py diff --git a/CLAUDE.md b/CLAUDE.md index 5ac2c35..a00331a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,7 +4,7 @@ ## System Profile - **Project Name:** GHAADD (GitHub Asset/Asset Downloader Daemon) -- **Tech Stack:** Python 3.11+ (Typing annotations required; `typing.NotRequired` in `daemon_control.py` and `daemon_lock.py` needs 3.11, found by the CI on 2026-10-13, and 3.10 reaches EOL in October 2026, so 3.11 is the declared minimum; `vermin` does not catch names imported from `typing`; developed and tested on 3.14; the CI runs 3.11 and 3.14 on Windows, macOS and Linux), SQLite (WAL mode enabled), JSON Configs, Cross-platform Filelocks. +- **Tech Stack:** Python 3.11+ (Typing annotations required, enforced since 2026-10-25: every function, nested ones and tests included, has a type on each parameter (Pyright `reportMissingParameterType` is an error in `pyrightconfig.json`) and a return type (`tests/test_annotations.py` reads the source, because Pyright has no rule for that); `Any` is the honest answer for test fakes and sqlite rows, not for new production code; the rule exists so Pyright has something to check; `typing.NotRequired` in `daemon_control.py` and `daemon_lock.py` needs 3.11, found by the CI on 2026-10-13, and 3.10 reaches EOL in October 2026, so 3.11 is the declared minimum; `vermin` does not catch names imported from `typing`; developed and tested on 3.14; the CI runs 3.11 and 3.14 on Windows, macOS and Linux), SQLite (WAL mode enabled), JSON Configs, Cross-platform Filelocks. - **Dependencies:** `requests`, `python-dotenv`, `imapclient`, `filelock` (pinned lower bounds in `requirements.txt`). - **Version:** see `modules/app_info.py` (the single source, used by `main.py` and the GUI window title; do not copy the number here) - **Architecture:** Monolithic entry points (`main.py`, `main_gui.py`) orchestrating isolated, highly specific functional files within the `/modules` directory. diff --git a/main.py b/main.py index 398d4a2..3e2a9a8 100644 --- a/main.py +++ b/main.py @@ -5,10 +5,12 @@ import sys import time from contextlib import contextmanager, nullcontext +from typing import Any, Dict, Iterator, Optional, TextIO, Tuple, Union from modules.config_manager import ( get_config_fingerprint, get_destination_check_every_n_polls, get_polling_settings, + TerminalLogSettings, get_terminal_log_settings, get_viewer_settings, load_config, @@ -42,7 +44,7 @@ from modules.queue_worker import process_queue_once, run_ingest_and_queue_cycle, run_single_cycle -def _ensure_standard_streams(): +def _ensure_standard_streams() -> None: """Give a console-less run (pythonw.exe, e.g. from Task Scheduler) real stdout/stderr objects. Without a console both are None, which breaks the TeeStream wrapping and any write to them. Output is @@ -65,7 +67,7 @@ def _ensure_standard_streams(): _ORIGINAL_STDERR = sys.stderr -def _configure_output_encoding(): +def _configure_output_encoding() -> None: """Make console output UTF-8 and unable to crash on a character the stream cannot encode. A redirected stream (GUI Start button, service, "> file") falls back to cp1252 on Windows, where the @@ -84,12 +86,12 @@ def _configure_output_encoding(): class TeeStream: """Mirror writes to terminal and a log file stream.""" - def __init__(self, primary_stream, secondary_stream): + def __init__(self, primary_stream: TextIO, secondary_stream: Optional[Union[TextIO, RollingLogFile]]) -> None: """Remember the real stream (primary) and the optional log file stream (secondary).""" self.primary_stream = primary_stream self.secondary_stream = secondary_stream - def _disable_secondary_stream(self, exc): + def _disable_secondary_stream(self, exc: Exception) -> None: """Drop the log file after a write error so logging trouble never stops the daemon. The failure is reported once on the original stderr; the console keeps working. @@ -109,7 +111,7 @@ def _disable_secondary_stream(self, exc): file=_ORIGINAL_STDERR, ) - def write(self, data): + def write(self, data: str) -> int: """Write to the console, and to the log file while one is attached.""" self.primary_stream.write(data) if self.secondary_stream is not None: @@ -120,7 +122,7 @@ def write(self, data): self._disable_secondary_stream(exc) return len(data) - def flush(self): + def flush(self) -> None: """Flush the console and the log file (if any).""" self.primary_stream.flush() if self.secondary_stream is not None: @@ -129,12 +131,12 @@ def flush(self): except (OSError, ValueError) as exc: self._disable_secondary_stream(exc) - def isatty(self): + def isatty(self) -> bool: """Report whether the console is a terminal (the log file does not count).""" return self.primary_stream.isatty() @property - def encoding(self): + def encoding(self) -> str: """Report the console's encoding, UTF-8 when the stream has none.""" return getattr(self.primary_stream, "encoding", "utf-8") @@ -147,18 +149,18 @@ class TerminalLog: daemon runs (each start() opens a new file). """ - def __init__(self, settings, tees): + def __init__(self, settings: TerminalLogSettings, tees: Tuple[TeeStream, TeeStream]) -> None: """Keep the log settings and the two TeeStreams (stdout, stderr) whose file side this class switches.""" self.settings = settings self._tees = tees self._log_file = None @property - def active(self): + def active(self) -> bool: """Return True while a log file is open and attached to both streams.""" return self._log_file is not None and all(tee.secondary_stream is not None for tee in self._tees) - def start(self): + def start(self) -> bool: """Open a new log file and mirror output into it; return True if logging is on.""" if self.active: return True @@ -183,7 +185,7 @@ def start(self): print(f"Logging enabled. Writing terminal output to: {self._log_file.path}") return True - def stop(self): + def stop(self) -> None: """Stop mirroring and close the log file (no-op when logging is off).""" if self._log_file is None: return @@ -196,11 +198,11 @@ def stop(self): except (OSError, ValueError): pass - def prune(self): + def prune(self) -> int: """Apply the retention limit; returns how many old files were removed.""" return self._log_file.prune() if self._log_file is not None else 0 - def apply_override(self, override): + def apply_override(self, override: Optional[bool]) -> None: """Follow a live switch: True/False force logging on/off, None returns to config.json.""" want = self.settings["enabled"] if override is None else override if want and not self.active: @@ -212,7 +214,7 @@ def apply_override(self, override): self.stop() -def setup_terminal_logging(config): +def setup_terminal_logging(config: Optional[Dict[str, Any]]) -> TerminalLog: """Route console output through TeeStream and return the (still closed) TerminalLog. The log file itself starts later, once main() knows it is entering a run mode, so --help and the @@ -223,7 +225,7 @@ def setup_terminal_logging(config): return TerminalLog(get_terminal_log_settings(config), tees) -def run_internal_smoke_tests(): +def run_internal_smoke_tests() -> None: """Run internal smoke tests for baseline release-download behavior.""" print("🚀 Starting Internal Smoke Tests...") @@ -248,7 +250,7 @@ def run_internal_smoke_tests(): @contextmanager -def _viewer_push_scope(pusher): +def _viewer_push_scope(pusher: Optional[ViewerPusher]) -> Iterator[None]: """Stop pushing to the viewer (telling it goodbye) however the polling loop ends.""" try: yield @@ -258,8 +260,13 @@ def _viewer_push_scope(pusher): def run_polling_loop( - interval_seconds, jitter_min_seconds, jitter_max_seconds, terminal_log=None, config_fingerprint=None, idle=False -): + interval_seconds: float, + jitter_min_seconds: int, + jitter_max_seconds: int, + terminal_log: Optional[TerminalLog] = None, + config_fingerprint: Optional[str] = None, + idle: bool = False, +) -> None: """Run processing continuously with randomized jitter between cycles. With `idle` (polling.enabled is false, started with --daemon) nothing is polled on a schedule: the daemon @@ -292,16 +299,16 @@ def run_polling_loop( else None ) - def apply_push_override(override): + def apply_push_override(override: Optional[bool]) -> None: """Follow a live push switch (None = the viewer.enabled setting of config.json).""" if pusher is not None: pusher.apply_override(override, viewer_settings["enabled"]) - def log_is_active(): + def log_is_active() -> bool: """Return whether the terminal log file is open right now (published for the GUI).""" return bool(terminal_log is not None and terminal_log.active) - def apply_log_override(override): + def apply_log_override(override: Optional[bool]) -> None: """Follow a live log switch, then publish whether the file really is open (the GUI shows it).""" if terminal_log is None: # nothing to switch (never registered then, but keeps the types honest) return @@ -309,7 +316,7 @@ def apply_log_override(override): if control_enabled: update_daemon_status(log_active=log_is_active()) - def publish_wait_state(paused, next_poll_at): + def publish_wait_state(paused: bool, next_poll_at: Optional[float]) -> None: """Publish pause state and next poll time to the status file, and announce pause/resume once.""" nonlocal announced_paused if control_enabled: @@ -358,14 +365,14 @@ def run_folder_checks(reason: str) -> None: run_folder_checks("at start") # counts and warnings are there right away, not after N polls run_backup_check() # a daemon that was off for a while catches up at once - def announce_requested_poll(wait_result): + def announce_requested_poll(wait_result: Optional[str]) -> None: """Say which poll was asked for; and note that a pause goes on afterwards.""" what = "Single poll requested (one notification, one queue item)" if wait_result == "single" else "Poll now requested" print(f"⏩ {what}." + (" Polling stays paused afterwards." if watcher.forced_while_paused else "")) if control_enabled: update_daemon_status(last_forced_poll_handled=watcher.last_handled_request) - def wait_for_poll_now(): + def wait_for_poll_now() -> Optional[str]: """Idle mode: wait (no countdown) until Poll now / Single poll or Stop; the request, or None on Stop.""" wait_result = watcher.wait(0, on_change=publish_wait_state, idle=True) if wait_result == "stop": @@ -434,7 +441,7 @@ def wait_for_poll_now(): cycle += 1 -def run_backup_check(): +def run_backup_check() -> None: """Make a backup when the schedule in config.json says one is due (never stops the daemon). The settings are read afresh each time, so a change in the Backup window applies without a restart. @@ -455,7 +462,7 @@ def run_backup_check(): log_warning("BACKUP", result.message) -def run_drain_queue_loop(github_token): +def run_drain_queue_loop(github_token: Optional[str]) -> None: """Process only due queue jobs (no email ingestion) until the queue is fully empty.""" print("Drain-queue mode enabled. Email ingestion is skipped.\n") @@ -484,7 +491,7 @@ def run_drain_queue_loop(github_token): cycle += 1 -def main(): +def main() -> None: """Entry point: parse the CLI, then run one of the modes (one-shot command, single, drain, once, polling). Only mutating run modes take the daemon lock and start the terminal log; dry-run and the read-only diff --git a/modules/asset_downloader.py b/modules/asset_downloader.py index 8d873aa..cbfad94 100644 --- a/modules/asset_downloader.py +++ b/modules/asset_downloader.py @@ -34,7 +34,7 @@ ) from modules.payload_types import DownloadReleaseResult, DownloadResultPayload, ReleaseAssetQueueItem, SkippedItemPayload from requests.exceptions import ChunkedEncodingError, ConnectionError, Timeout -from typing import Literal, Optional, cast +from typing import Any, cast, Dict, Literal, Optional from urllib.parse import quote, urljoin # Load environment variables from .env. @@ -806,7 +806,7 @@ def _github_archive_name(repo_name: str, tag: str, extension: str) -> str: return f"{repo_name}-{version}.{extension}" -def _probe_archive_name(session, repo: str, tag: str, extension: str, fallback: str) -> str: +def _probe_archive_name(session: requests.Session, repo: str, tag: str, extension: str, fallback: str) -> str: """Return the file name GitHub sends for the release page's own archive link, else the fallback.""" url = f"https://github.com/{repo}/archive/refs/tags/{quote(tag, safe='/')}.{extension}" try: @@ -820,7 +820,7 @@ def _probe_archive_name(session, repo: str, tag: str, extension: str, fallback: return name if name and name != "unknown" else fallback -def sanitize_folder_name(text): +def sanitize_folder_name(text: Optional[str]) -> str: """Return a Windows-safe folder name using JS-compatible rules.""" if not text: return "unknown" @@ -829,7 +829,7 @@ def sanitize_folder_name(text): return text.strip() -def get_short_commit_hash(repo, tag, headers): +def get_short_commit_hash(repo: str, tag: str, headers: Dict[str, str]) -> Any | Literal['unknown-commit']: """Return the 7-character commit hash for a specific tag.""" url = f"https://api.github.com/repos/{repo}/commits/{tag}" response = requests.get(url, headers=headers) @@ -838,7 +838,7 @@ def get_short_commit_hash(repo, tag, headers): return "unknown-commit" -def build_folder_name(repo, release_data, headers, raw_name): +def build_folder_name(repo: str, release_data: Dict[str, Any], headers: Dict[str, str], raw_name: str) -> str: """Build a release folder name from a specific display name.""" pub_time_str = release_data.get("published_at", "") if pub_time_str: @@ -858,13 +858,13 @@ def build_folder_name(repo, release_data, headers, raw_name): return f"{formatted_date}, {safe_name}, {safe_tag}, {safe_commit}" -def generate_folder_name(repo, release_data, headers): +def generate_folder_name(repo: str, release_data: Dict[str, Any], headers: Dict[str, str]) -> str: """Generate the default folder name based on the release title.""" release_name = release_data.get("name") or release_data.get("tag_name", "unknown-name") return build_folder_name(repo, release_data, headers, release_name) -def is_state_persistence_enabled(): +def is_state_persistence_enabled() -> bool: """Return whether persistent duplicate-detection state is enabled.""" return not is_state_persistence_disabled() @@ -883,7 +883,7 @@ def build_expected_signature(item: ReleaseAssetQueueItem) -> Optional[str]: return "|".join(signature_parts) -def get_release_data(repo, tag): +def get_release_data(repo: str, tag: str) -> Any | None: """Return release JSON for a repository tag, or None when unavailable.""" url = f"https://api.github.com/repos/{repo}/releases/tags/{tag}" @@ -907,7 +907,7 @@ def get_release_data(repo, tag): return None -def get_release_attestation_url(release_data, headers): +def get_release_attestation_url(release_data: Dict[str, Any], headers: Dict[str, str]) -> str | None: """Return the release attestation download URL from the expanded assets page. GitHub exposes this link in expanded-assets HTML, not in release JSON. @@ -1117,15 +1117,15 @@ def download_all_assets( per_file_retry_delay = 2 normal_downloaded_count = 0 - def refresh_release_state(): + def refresh_release_state() -> None: """Reload the stored state of this release from state.db into the enclosing function's variable.""" nonlocal release_state if state_db is not None: release_state = load_release_state(state_db, release_key) - def should_skip_existing_file(file_path, item_key, expected_signature, remote_size, remote_last_modified, remote_etag): + def should_skip_existing_file(file_path: Optional[str], item_key: str, expected_signature: Optional[str], remote_size: Any, remote_last_modified: Any, remote_etag: Any) -> bool: """Return True when the local file can be treated as up to date.""" - if not os.path.exists(file_path): + if file_path is None or not os.path.exists(file_path): return False state_entry = release_state.get(item_key, {}) diff --git a/modules/config_manager.py b/modules/config_manager.py index 4a65a95..9d8dab4 100644 --- a/modules/config_manager.py +++ b/modules/config_manager.py @@ -9,7 +9,7 @@ import sys import tempfile import time -from typing import Any, Callable, Dict, Optional, TypedDict, TypeVar +from typing import Any, Callable, Dict, List, Optional, Tuple, TypedDict, TypeVar from filelock import FileLock, Timeout @@ -86,7 +86,7 @@ class TerminalLogSettings(TypedDict): keep_files: int -def _as_bool(value, default=False): +def _as_bool(value: Any, default: bool = False) -> bool: """Return a boolean parsed from value, or default when value is unset.""" if isinstance(value, bool): return value @@ -95,7 +95,7 @@ def _as_bool(value, default=False): return str(value).strip().lower() == "true" -def _as_int(value, default): +def _as_int(value: Any, default: int) -> int: """Return value parsed as int, or default when parsing fails.""" if value is None: return default @@ -105,7 +105,7 @@ def _as_int(value, default): return default -def _get_nested(config, *keys): +def _get_nested(config: Dict[str, Any], *keys: str) -> Any: """Return a nested dictionary value by key path, or None when missing.""" current = config for key in keys: @@ -289,7 +289,7 @@ def _apply(config: Dict[str, Any]) -> bool: return update_config(_apply) -def _hand_edited_defaults(): +def _hand_edited_defaults() -> List[Tuple[str, Any]]: """Optional settings that have no field in the Settings window, with the value used when they are missing. The getters fall back to these anyway; writing them into config.json only makes them visible and easy to edit. @@ -334,14 +334,14 @@ def _apply(config: Dict[str, Any]) -> bool: return update_config(_apply) -def get_max_emails_to_process(config=None): +def get_max_emails_to_process(config: Optional[Dict[str, Any]] = None) -> int: """Return the configured maximum number of emails to process.""" config = config if config is not None else load_config() value = _get_nested(config, "processing", "max_emails_to_process") return max(0, _as_int(value, DEFAULT_MAX_EMAILS_TO_PROCESS)) -def get_recheck_intervals_minutes(config=None): +def get_recheck_intervals_minutes(config: Optional[Dict[str, Any]] = None) -> list[int] | list[Any]: """Return queue re-check intervals in minutes, normalized and validated.""" config = config if config is not None else load_config() @@ -352,8 +352,11 @@ def get_recheck_intervals_minutes(config=None): normalized = [] for item in value: - minutes = _as_int(item, None) - if minutes is None or minutes <= 0: + try: + minutes = int(item) + except (TypeError, ValueError): + continue + if minutes <= 0: continue if minutes not in normalized: normalized.append(minutes) @@ -364,14 +367,14 @@ def get_recheck_intervals_minutes(config=None): return normalized -def is_state_persistence_disabled(config=None): +def is_state_persistence_disabled(config: Optional[Dict[str, Any]] = None) -> bool: """Return whether state persistence is disabled in configuration.""" config = config if config is not None else load_config() value = _get_nested(config, "state", "disable_state_persistence") return _as_bool(value, DEFAULT_DISABLE_STATE_PERSISTENCE) -def get_destination_check_every_n_polls(config=None): +def get_destination_check_every_n_polls(config: Optional[Dict[str, Any]] = None) -> int: """Return how often (in poll cycles) to verify mapped destinations still exist. A value of 0 disables the periodic check. @@ -393,7 +396,7 @@ def is_valid_subfolder(value: Any) -> bool: return ".." not in re.split(r"[\\/]+", text) -def get_default_subfolder(config=None): +def get_default_subfolder(config: Optional[Dict[str, Any]] = None) -> str: """Return the subfolder given to newly created mapping entries (paths.default_subfolder). An empty string (also the default) means no subfolder; a missing or unusable value gives the default. @@ -405,7 +408,7 @@ def get_default_subfolder(config=None): return DEFAULT_SUBFOLDER -def get_default_repository_limit(config=None): +def get_default_repository_limit(config: Optional[Dict[str, Any]] = None) -> int: """Return the folder limit given to newly created mapping entries (processing.default_limit). 0 switches the limit check off for new entries; a missing or negative value gives 10. @@ -415,7 +418,7 @@ def get_default_repository_limit(config=None): return value if value >= 0 else DEFAULT_REPOSITORY_LIMIT -def _gui_seconds(config, key, default, bounds): +def _gui_seconds(config: Dict[str, Any], key: str, default: float, bounds: Tuple[float, float]) -> float: """A number of seconds from the "gui" section, kept inside bounds; anything unusable gives the default.""" value = _get_nested(config, "gui", key) if isinstance(value, bool) or value is None: @@ -429,13 +432,13 @@ def _gui_seconds(config, key, default, bounds): return min(max(seconds, bounds[0]), bounds[1]) -def get_gui_refresh_seconds(config=None): +def get_gui_refresh_seconds(config: Optional[Dict[str, Any]] = None) -> float: """Seconds between the GUI's data refreshes (gui.refresh_seconds, default 3, 1 to 60).""" config = config if config is not None else load_config() return _gui_seconds(config, "refresh_seconds", DEFAULT_GUI_REFRESH_SECONDS, GUI_REFRESH_SECONDS_RANGE) -def get_gui_status_message_seconds(config=None): +def get_gui_status_message_seconds(config: Optional[Dict[str, Any]] = None) -> float: """Seconds a GUI status-bar message stays (gui.status_message_seconds, default 6, 2 to 60).""" config = config if config is not None else load_config() return _gui_seconds( @@ -443,38 +446,38 @@ def get_gui_status_message_seconds(config=None): ) -def _gui_flag(config, key, default): +def _gui_flag(config: Dict[str, Any], key: str, default: bool) -> bool: """A true/false setting from the "gui" section; anything that is not a real boolean gives the default.""" value = _get_nested(config, "gui", key) return value if isinstance(value, bool) else default -def get_gui_tray_enabled(config=None): +def get_gui_tray_enabled(config: Optional[Dict[str, Any]] = None) -> bool: """Use a system tray icon (gui.tray, default true; it also needs the optional pystray + Pillow packages).""" config = config if config is not None else load_config() return _gui_flag(config, "tray", True) -def get_gui_minimize_to_tray(config=None): +def get_gui_minimize_to_tray(config: Optional[Dict[str, Any]] = None) -> bool: """Minimizing hides the window in the tray (gui.minimize_to_tray; default true on Windows/macOS, false on Linux where many desktops show no tray icon and the window would be lost).""" config = config if config is not None else load_config() return _gui_flag(config, "minimize_to_tray", sys.platform != "linux") -def get_gui_close_to_tray(config=None): +def get_gui_close_to_tray(config: Optional[Dict[str, Any]] = None) -> bool: """Closing the window hides it in the tray instead of ending the GUI (gui.close_to_tray, default false).""" config = config if config is not None else load_config() return _gui_flag(config, "close_to_tray", False) -def get_gui_notifications_enabled(config=None): +def get_gui_notifications_enabled(config: Optional[Dict[str, Any]] = None) -> bool: """Return gui.notifications (default true): tray notices for new warnings, failed jobs, unmapped repos, a stopped daemon.""" config = config if config is not None else load_config() return _gui_flag(config, "notifications", True) -def get_gui_silenced_warning_types(config=None) -> list[str]: +def get_gui_silenced_warning_types(config: Optional[Dict[str, Any]] = None) -> list[str]: """Warning types that never pop up a notification (gui.silenced_warning_types; default API and LIMIT). The Warnings tab still shows them. A value that is not a list of strings gives the default. @@ -486,19 +489,19 @@ def get_gui_silenced_warning_types(config=None) -> list[str]: return list(DEFAULT_SILENCED_WARNING_TYPES) -def get_gui_dark_mode(config=None): +def get_gui_dark_mode(config: Optional[Dict[str, Any]] = None) -> bool: """Return gui.dark_mode (default false): the GUI's dark color scheme, toggled by the button in the header.""" config = config if config is not None else load_config() return _gui_flag(config, "dark_mode", False) -def get_gui_start_minimized(config=None): +def get_gui_start_minimized(config: Optional[Dict[str, Any]] = None) -> bool: """Return gui.start_minimized (default false): open the GUI minimized, as the minimize button does, not shown.""" config = config if config is not None else load_config() return _gui_flag(config, "start_minimized", False) -def get_gui_start_daemon(config=None): +def get_gui_start_daemon(config: Optional[Dict[str, Any]] = None) -> bool: """Start the polling daemon when the GUI starts and none is running (gui.start_daemon, default false).""" config = config if config is not None else load_config() return _gui_flag(config, "start_daemon", False) @@ -534,7 +537,7 @@ def add_path(candidate: Any) -> None: return unique_dirs -def get_gmail_folder(config=None): +def get_gmail_folder(config: Optional[Dict[str, Any]] = None) -> str: """Return the Gmail folder used for GitHub notification processing.""" config = config if config is not None else load_config() value = _get_nested(config, "mailbox", "folder") @@ -654,12 +657,12 @@ def get_config_fingerprint(config: Optional[Dict[str, Any]] = None) -> str: return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] -def get_download_dir_for_release(repo, release_type=None, config=None): +def get_download_dir_for_release(repo: str, release_type: Optional[str] = None, config: Optional[Dict[str, Any]] = None) -> str: """Return the configured default download directory for a release.""" return get_default_download_dir(config) -def get_polling_settings(config=None): +def get_polling_settings(config: Optional[Dict[str, Any]] = None) -> dict[str, Any]: """Return normalized polling settings with safe defaults.""" config = config if config is not None else load_config() diff --git a/modules/daemon_lock.py b/modules/daemon_lock.py index 8018209..0f50366 100644 --- a/modules/daemon_lock.py +++ b/modules/daemon_lock.py @@ -6,7 +6,7 @@ import tempfile import time from contextlib import contextmanager -from typing import NotRequired, Optional, TypedDict +from typing import Any, Iterator, NotRequired, Optional, TypedDict from filelock import FileLock, Timeout @@ -87,7 +87,7 @@ def _write_daemon_status() -> None: ) -def update_daemon_status(**fields) -> None: +def update_daemon_status(**fields: Any) -> None: """Merge fields (e.g. paused, next_poll_at) into the status sidecar, best-effort. Only the daemon that holds the singleton lock may call this. @@ -191,7 +191,7 @@ def clear_queue_progress() -> None: @contextmanager -def publishing_current_job(repo: str, tag: str): +def publishing_current_job(repo: str, tag: str) -> Iterator[None]: """Publish the job being processed so the GUI can show it as "Running". A no-op unless this process holds the daemon lock (tests, dry-run and @@ -208,7 +208,7 @@ def publishing_current_job(repo: str, tag: str): @contextmanager -def acquire_daemon_lock(): +def acquire_daemon_lock() -> Iterator[None]: """Acquire the singleton daemon lock, exiting the process if already held. Guards only mutating run modes (default single run, --once, --poll, diff --git a/modules/db_manager.py b/modules/db_manager.py index 322695a..c1b0328 100644 --- a/modules/db_manager.py +++ b/modules/db_manager.py @@ -3,18 +3,18 @@ import re import os import sqlite3 -from typing import Iterable, Optional +from typing import Any, Iterable, List, Optional, Sequence STATE_DB_NAME = "state.db" -def get_state_db_path(): +def get_state_db_path() -> str: """Return the SQLite state database path beside the app files.""" app_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) return os.path.join(app_dir, STATE_DB_NAME) -def purge_state_database(): +def purge_state_database() -> bool: """Delete the local state database if it exists.""" state_db_path = get_state_db_path() if not os.path.exists(state_db_path): @@ -24,7 +24,7 @@ def purge_state_database(): return True -def backup_database(destination_path): +def backup_database(destination_path: str) -> None: """Write a consistent copy of state.db to `destination_path` (SQLite online backup, safe while the daemon runs).""" source = open_database() try: @@ -37,7 +37,7 @@ def backup_database(destination_path): source.close() -def open_database(): +def open_database() -> sqlite3.Connection: """Open the SQLite database, enable WAL mode, and ensure the schema exists.""" connection = sqlite3.connect(get_state_db_path()) connection.row_factory = sqlite3.Row @@ -324,7 +324,7 @@ def open_database(): return connection -def set_folder_count(connection, repo, folder_count, counted_at): +def set_folder_count(connection: sqlite3.Connection, repo: str, folder_count: int, counted_at: float) -> None: """Remember how many release folders a repository had when they were last counted.""" connection.execute( """ @@ -337,7 +337,7 @@ def set_folder_count(connection, repo, folder_count, counted_at): connection.commit() -def delete_folder_counts(connection, keep_repos=None, repo=None): +def delete_folder_counts(connection: sqlite3.Connection, keep_repos: Optional[Iterable[str]] = None, repo: Optional[str] = None) -> int: """Delete one repository's count, or every count except those of `keep_repos` (names, any case).""" if repo is not None: cursor = connection.execute("DELETE FROM folder_counts WHERE repo_key = ?", (str(repo).strip().lower(),)) @@ -353,7 +353,7 @@ def delete_folder_counts(connection, keep_repos=None, repo=None): return cursor.rowcount -def get_folder_counts(connection): +def get_folder_counts(connection: sqlite3.Connection) -> dict[Any, dict[str, Any]]: """{lower-cased repo: {"folder_count": int, "counted_at": float}} (read-only).""" rows = connection.execute("SELECT repo_key, folder_count, counted_at FROM folder_counts").fetchall() return {row[0]: {"folder_count": row[1], "counted_at": row[2]} for row in rows} @@ -361,7 +361,7 @@ def get_folder_counts(connection): # Asset state helpers. -def load_release_state(connection, release_key): +def load_release_state(connection: sqlite3.Connection, release_key: str) -> dict[Any, Any]: """Return stored asset state for a release key.""" rows = connection.execute( """ @@ -388,11 +388,13 @@ def load_release_state(connection, release_key): return state -def save_state_entry(connection, release_key, item_key, file_name, rel_file_path, expected_signature, remote_size, remote_last_modified, remote_etag, base_dir): +def save_state_entry(connection: sqlite3.Connection, release_key: str, item_key: str, file_name: str, rel_file_path: Optional[str], expected_signature: Optional[str], remote_size: Any, remote_last_modified: Any, remote_etag: Any, base_dir: str) -> None: """Insert or update one asset-state record using a relative path.""" # Reconstruct the absolute path for local filesystem checks - abs_path = os.path.join(base_dir, rel_file_path) try: + if rel_file_path is None: + raise OSError("no file path") + abs_path = os.path.join(base_dir, rel_file_path) local_size = os.path.getsize(abs_path) local_mtime = os.path.getmtime(abs_path) except OSError: @@ -421,7 +423,7 @@ def save_state_entry(connection, release_key, item_key, file_name, rel_file_path connection.commit() -def prune_release_state(connection, release_key, valid_item_keys, base_dir): +def prune_release_state(connection: sqlite3.Connection, release_key: str, valid_item_keys: Iterable[str], base_dir: str) -> None: """Delete stale asset-state rows using dynamically constructed absolute paths.""" rows = connection.execute("SELECT item_key, file_path FROM asset_state WHERE release_key = ?", (release_key,)).fetchall() @@ -448,7 +450,7 @@ def prune_release_state(connection, release_key, valid_item_keys, base_dir): # Queue state helpers. -def enqueue_job(connection, repo, tag, release_type=None, next_check_time=None, expected_commit=None): +def enqueue_job(connection: sqlite3.Connection, repo: str, tag: str, release_type: Optional[str] = None, next_check_time: Optional[float] = None, expected_commit: Optional[str] = None) -> int: """Insert a new queued job row and return its row id.""" if next_check_time is None: next_check_time = 0.0 @@ -483,7 +485,7 @@ def enqueue_job(connection, repo, tag, release_type=None, next_check_time=None, return job_id -def get_pending_job_for_release(connection, repo, tag, release_type=None, expected_commit=None, exclude_job_id=None): +def get_pending_job_for_release(connection: sqlite3.Connection, repo: str, tag: str, release_type: Optional[str] = None, expected_commit: Optional[str] = None, exclude_job_id: Optional[int] = None) -> Any: """Return one pending job for the same release identity, optionally matching commit.""" conditions = [ "status = 'PENDING'", @@ -526,7 +528,7 @@ def get_pending_job_for_release(connection, repo, tag, release_type=None, expect return connection.execute(query, tuple(params)).fetchone() -def get_pending_jobs_for_release(connection, repo, tag, release_type=None): +def get_pending_jobs_for_release(connection: sqlite3.Connection, repo: str, tag: str, release_type: Optional[str] = None) -> list[Any]: """Return pending jobs for one release identity ordered by creation time.""" return connection.execute( """ @@ -560,13 +562,13 @@ def get_pending_jobs_for_release(connection, repo, tag, release_type=None): def supersede_pending_jobs_for_release( - connection, - repo, - tag, - release_type=None, - replacement_expected_commit=None, - last_result="SUPERSEDED_REPLACED_BY_NEW_NOTIFICATION", -): + connection: sqlite3.Connection, + repo: str, + tag: str, + release_type: Optional[str] = None, + replacement_expected_commit: Optional[str] = None, + last_result: str = "SUPERSEDED_REPLACED_BY_NEW_NOTIFICATION", +) -> int: """Supersede pending jobs for a release identity and return affected row count.""" cursor = connection.execute( """ @@ -597,7 +599,7 @@ def supersede_pending_jobs_for_release( return int(cursor.rowcount or 0) -def get_due_jobs(connection, now_timestamp, limit=None): +def get_due_jobs(connection: sqlite3.Connection, now_timestamp: float, limit: Optional[int] = None) -> list[Any]: """Return jobs that are ready to run, sorted by schedule time then id.""" if limit is not None: return connection.execute( @@ -650,7 +652,7 @@ def get_due_jobs(connection, now_timestamp, limit=None): ).fetchall() -def get_pending_jobs(connection): +def get_pending_jobs(connection: sqlite3.Connection) -> list[Any]: """Return all pending queue jobs ordered by creation time.""" return connection.execute( """ @@ -676,7 +678,7 @@ def get_pending_jobs(connection): ).fetchall() -def get_next_pending_job(connection): +def get_next_pending_job(connection: sqlite3.Connection) -> Any: """Return the next scheduled pending job, or None when queue is empty.""" return connection.execute( """ @@ -706,7 +708,7 @@ def get_next_pending_job(connection): _LIMIT_WARNING_REPO = re.compile(r"Folder limit warning: (\S+) currently has ") -def purge_limit_warnings_for_repo(connection, repo, dry_run=False): +def purge_limit_warnings_for_repo(connection: sqlite3.Connection, repo: str, dry_run: bool = False) -> int: """Delete the folder-limit warnings written for one repository; returns the count. The rows hold the repository only in their text, so they are matched on the message's fixed @@ -722,7 +724,7 @@ def purge_limit_warnings_for_repo(connection, repo, dry_run=False): return cursor.rowcount -def get_repos_with_limit_warnings(connection): +def get_repos_with_limit_warnings(connection: sqlite3.Connection) -> set[Any]: """Lower-cased owner/repo names that have at least one folder-limit warning stored (read-only). The warning rows carry the repository only in their text ("Folder limit warning: owner/repo currently @@ -739,7 +741,7 @@ def get_repos_with_limit_warnings(connection): return found -def get_failed_jobs_since(connection, since_epoch): +def get_failed_jobs_since(connection: sqlite3.Connection, since_epoch: float) -> list[dict[Any, Any]]: """Return jobs that ended FAILED at or after `since_epoch` (read-only), oldest first, for the GUI's notifications. Times have whole-second resolution, so the caller drops the ids it has already reported at the boundary. @@ -756,7 +758,7 @@ def get_failed_jobs_since(connection, since_epoch): return [dict(row) for row in rows] -def get_repo_job_summaries(connection): +def get_repo_job_summaries(connection: sqlite3.Connection) -> dict[Any, Any]: """Return one read-only job_queue summary per repo, keyed by lower-cased owner/repo (GUI Mappings table). latest_* come from the newest non-SUPERSEDED job (None if only superseded ones exist); pending_next_check and @@ -870,7 +872,7 @@ def get_repo_job_summaries(connection): return summaries -def get_jobs_by_ids(connection, job_ids: Iterable[int]): +def get_jobs_by_ids(connection: sqlite3.Connection, job_ids: Iterable[int]) -> list[Any]: """Return queue jobs for the provided IDs.""" normalized_ids = sorted({int(job_id) for job_id in job_ids}) if not normalized_ids: @@ -903,10 +905,10 @@ def get_jobs_by_ids(connection, job_ids: Iterable[int]): def supersede_pending_jobs_by_ids( - connection, + connection: sqlite3.Connection, job_ids: Iterable[int], - last_result="SUPERSEDED_MANUAL_REMOVE", -): + last_result: str = "SUPERSEDED_MANUAL_REMOVE", +) -> list[Any]: """Mark specific pending jobs as SUPERSEDED and return removed rows.""" normalized_ids = sorted({int(job_id) for job_id in job_ids}) if not normalized_ids: @@ -958,7 +960,7 @@ def supersede_pending_jobs_by_ids( return pending_rows -def supersede_duplicate_pending_jobs(connection): +def supersede_duplicate_pending_jobs(connection: sqlite3.Connection) -> int: """Mark duplicate pending jobs as SUPERSEDED, keeping the oldest per release+commit.""" pending_rows = connection.execute( """ @@ -1026,14 +1028,14 @@ def supersede_duplicate_pending_jobs(connection): def mark_job_completed( - connection, - job_id, - attempt_count=None, - downloaded_count=0, - skipped_count=0, - total_items=0, - last_result="SUCCESS", -): + connection: sqlite3.Connection, + job_id: int, + attempt_count: Optional[int] = None, + downloaded_count: int = 0, + skipped_count: int = 0, + total_items: int = 0, + last_result: str = "SUCCESS", +) -> None: """Mark a queued job as completed.""" if attempt_count is None: connection.execute( @@ -1077,15 +1079,15 @@ def mark_job_completed( def mark_job_failed( - connection, - job_id, - attempt_count=None, - expected_commit=None, - downloaded_count=0, - skipped_count=0, - total_items=0, - last_result="FAILED", -): + connection: sqlite3.Connection, + job_id: int, + attempt_count: Optional[int] = None, + expected_commit: Optional[str] = None, + downloaded_count: int = 0, + skipped_count: int = 0, + total_items: int = 0, + last_result: str = "FAILED", +) -> None: """Mark a queued job as failed (no remaining retry intervals).""" if attempt_count is None: connection.execute( @@ -1139,15 +1141,15 @@ def mark_job_failed( def mark_job_superseded( - connection, - job_id, - attempt_count=None, - expected_commit=None, - downloaded_count=0, - skipped_count=0, - total_items=0, - last_result="SUPERSEDED", -): + connection: sqlite3.Connection, + job_id: int, + attempt_count: Optional[int] = None, + expected_commit: Optional[str] = None, + downloaded_count: int = 0, + skipped_count: int = 0, + total_items: int = 0, + last_result: str = "SUPERSEDED", +) -> None: """Mark a queued job as superseded by a newer commit/job.""" if attempt_count is None: connection.execute( @@ -1201,17 +1203,17 @@ def mark_job_superseded( def reschedule_job( - connection, - job_id, - next_check_time, - attempt_count, - expected_commit=None, - downloaded_count=0, - skipped_count=0, - total_items=0, - working_dir=None, - last_result="RETRY", -): + connection: sqlite3.Connection, + job_id: int, + next_check_time: float, + attempt_count: int, + expected_commit: Optional[str] = None, + downloaded_count: int = 0, + skipped_count: int = 0, + total_items: int = 0, + working_dir: Optional[str] = None, + last_result: str = "RETRY", +) -> None: """Update a queued job with a new schedule and attempt counter.""" connection.execute( """ @@ -1245,15 +1247,15 @@ def reschedule_job( def update_job_for_manual_check( - connection, - job_id, - expected_commit=None, - downloaded_count=0, - skipped_count=0, - total_items=0, - working_dir=None, - last_result="MANUAL_CHECK", -): + connection: sqlite3.Connection, + job_id: int, + expected_commit: Optional[str] = None, + downloaded_count: int = 0, + skipped_count: int = 0, + total_items: int = 0, + working_dir: Optional[str] = None, + last_result: str = "MANUAL_CHECK", +) -> None: """Update a pending job's counters and result without changing its retry schedule.""" connection.execute( """ @@ -1282,7 +1284,7 @@ def update_job_for_manual_check( connection.commit() -def save_job_skip_details(connection, job_id, attempt_count, skipped_items): +def save_job_skip_details(connection: sqlite3.Connection, job_id: int, attempt_count: int, skipped_items: Any) -> None: """Persist skipped item details for a queue job attempt.""" if not skipped_items: return @@ -1321,8 +1323,8 @@ def save_job_skip_details(connection, job_id, attempt_count, skipped_items): def get_previous_successful_completed_job( - connection, repo, tag, release_type, exclude_job_id, match_tag=True, finished_before=None -): + connection: sqlite3.Connection, repo: str, tag: str, release_type: Optional[str], exclude_job_id: int, match_tag: bool = True, finished_before: Optional[float] = None +) -> Any: """Return the previous successful completed job for the same repo and release_type. match_tag=True also requires the same tag; False accepts the newest earlier success of the repository. @@ -1371,15 +1373,15 @@ def get_previous_successful_completed_job( # Lifecycle event helpers. def insert_lifecycle_event( - connection, - event_type, - message, - category=None, - repo=None, - tag=None, - commit_hash=None, - destination_path=None, -): + connection: sqlite3.Connection, + event_type: str, + message: str, + category: Optional[str] = None, + repo: Optional[str] = None, + tag: Optional[str] = None, + commit_hash: Optional[str] = None, + destination_path: Optional[str] = None, +) -> int: """Insert one lifecycle event row and return its row id.""" cursor = connection.execute( """ @@ -1397,7 +1399,7 @@ def insert_lifecycle_event( return event_id -def get_lifecycle_events(connection, limit: Optional[int] = 20, event_type=None, repo_filter=None): +def get_lifecycle_events(connection: sqlite3.Connection, limit: Optional[int] = 20, event_type: Optional[str] = None, repo_filter: Optional[str] = None) -> list[Any]: """Return recent lifecycle events, newest first, optionally filtered.""" conditions = [] params: list = [] @@ -1425,7 +1427,7 @@ def get_lifecycle_events(connection, limit: Optional[int] = 20, event_type=None, return connection.execute(query, params).fetchall() -def get_existing_event_ids(connection, ids): +def get_existing_event_ids(connection: sqlite3.Connection, ids: Iterable[int]) -> set[Any]: """The subset of `ids` that still exists in lifecycle_events (read-only primary-key lookup).""" ids = [int(i) for i in ids] found = set() @@ -1438,10 +1440,10 @@ def get_existing_event_ids(connection, ids): return found -def _tab_event_filter(event_types, categories, exclude_categories): +def _tab_event_filter(event_types: Sequence[str], categories: Optional[Sequence[str]], exclude_categories: Sequence[str]) -> tuple[list[str], List[Any]]: """WHERE conditions and params selecting the events of one GUI status tab.""" conditions = [f"event_type IN ({','.join('?' * len(event_types))})"] - params = list(event_types) + params: List[Any] = list(event_types) if categories: conditions.append(f"category IN ({','.join('?' * len(categories))})") params.extend(categories) @@ -1451,7 +1453,7 @@ def _tab_event_filter(event_types, categories, exclude_categories): return conditions, params -def purge_events_for_tab(connection, event_types, categories=None, exclude_categories=(), dry_run=False): +def purge_events_for_tab(connection: sqlite3.Connection, event_types: Sequence[str], categories: Optional[Sequence[str]] = None, exclude_categories: Sequence[str] = (), dry_run: bool = False) -> int: """Delete exactly the events a GUI status tab lists (same filter as get_events_for_tab); returns the count. With dry_run=True nothing is deleted and the number of matching events is returned. @@ -1466,13 +1468,13 @@ def purge_events_for_tab(connection, event_types, categories=None, exclude_categ def get_events_for_tab( - connection, - event_types, - categories=None, - exclude_categories=(), - after_id=None, - limit=500, -): + connection: sqlite3.Connection, + event_types: Sequence[str], + categories: Optional[Sequence[str]] = None, + exclude_categories: Sequence[str] = (), + after_id: Optional[int] = None, + limit: int = 500, +) -> list[dict[Any, Any]]: """Return lifecycle events for one GUI status tab, newest first (read-only). event_types/categories are allow-lists (categories=None means any), exclude_categories drops @@ -1497,14 +1499,14 @@ def get_events_for_tab( return [dict(row) for row in rows] -def get_storage_stats(connection): +def get_storage_stats(connection: sqlite3.Connection) -> dict[str, Any]: """Return read-only size and growth figures for state.db (used by the --perf-report diagnostics).""" - def scalar(sql, params=()): + def scalar(sql: str, params: Sequence[Any] = ()) -> Any: """Run a query and return its single value, or None when there is no row or the value is NULL.""" row = connection.execute(sql, params).fetchone() return None if row is None or row[0] is None else row[0] - def grouped(sql): + def grouped(sql: str) -> dict[str, int]: """Run a two-column query and return it as {first column as text: second column as int}.""" return {str(row[0]): int(row[1]) for row in connection.execute(sql).fetchall()} @@ -1532,13 +1534,13 @@ def grouped(sql): } -def get_job_history(connection): +def get_job_history(connection: sqlite3.Connection) -> list[tuple[Any, Any, Any, float]]: """Every job as (repo, tag, status, created_at), oldest first (read-only; feeds the --stats figures).""" rows = connection.execute("SELECT repo, tag, status, created_at FROM job_queue ORDER BY created_at, id").fetchall() return [(row[0], row[1], row[2], float(row[3])) for row in rows] -def add_job_folder_size(connection, job_id, folder_bytes, folder_files): +def add_job_folder_size(connection: sqlite3.Connection, job_id: int, folder_bytes: int, folder_files: int) -> None: """Add the size of one of a job's folders (finished, or quarantined in Partial) to what the job brought in.""" connection.execute( """ @@ -1551,7 +1553,7 @@ def add_job_folder_size(connection, job_id, folder_bytes, folder_files): connection.commit() -def get_job_folder_sizes(connection): +def get_job_folder_sizes(connection: sqlite3.Connection) -> list[tuple[Any, Any, float, int, int]]: """Every job with a measured folder as (repo, tag, created_at, bytes, files), oldest first (read-only). Includes the sizes archived when their jobs were purged, so purging history does not shrink the data volumes. @@ -1567,7 +1569,7 @@ def get_job_folder_sizes(connection): return [(row[0], row[1], float(row[2]), int(row[3]), int(row[4] or 0)) for row in rows] -def get_cycle_times(connection): +def get_cycle_times(connection: sqlite3.Connection) -> list[float]: """Creation times of the CYCLE_SUMMARY events (one per polling cycle), oldest first (read-only).""" rows = connection.execute( "SELECT created_at FROM lifecycle_events WHERE event_type = 'CYCLE_SUMMARY' ORDER BY created_at" @@ -1575,7 +1577,7 @@ def get_cycle_times(connection): return [float(row[0]) for row in rows] -def get_release_sizes(connection): +def get_release_sizes(connection: sqlite3.Connection) -> dict[str, dict[str, int]]: """{"owner/repo|tag": {"bytes": int, "files": int, "newest": epoch}} from asset_state (read-only). Uses the size of the local file, else the size the server reported (source archives may lack one). @@ -1595,13 +1597,13 @@ def get_release_sizes(connection): } -def get_max_event_id(connection): +def get_max_event_id(connection: sqlite3.Connection) -> int: """Return the newest lifecycle event id (0 when there are no events).""" row = connection.execute("SELECT MAX(id) FROM lifecycle_events").fetchone() return int(row[0] or 0) -def purge_lifecycle_events(connection, event_type=None, repo_filter=None, min_age_days=None, dry_run=False): +def purge_lifecycle_events(connection: sqlite3.Connection, event_type: Optional[str] = None, repo_filter: Optional[str] = None, min_age_days: Optional[int] = None, dry_run: bool = False) -> int: """Delete lifecycle events matching the filters and return the number removed. min_age_days=0 means no age floor; larger values only match events at least that many days old. @@ -1635,7 +1637,7 @@ def purge_lifecycle_events(connection, event_type=None, repo_filter=None, min_ag return cursor.rowcount -def purge_job_queue_rows(connection, status=None, repo_filter=None, min_age_days=None, oldest_count=None, dry_run=False): +def purge_job_queue_rows(connection: sqlite3.Connection, status: Optional[str] = None, repo_filter: Optional[str] = None, min_age_days: Optional[int] = None, oldest_count: Optional[int] = None, dry_run: bool = False) -> int: """Delete terminal (non-PENDING) job_queue rows matching the filters; return the number removed. PENDING jobs are never purged. The delete cascades to job_skip_details. The caller gives min_age_days (0 = all @@ -1687,7 +1689,7 @@ def purge_job_queue_rows(connection, status=None, repo_filter=None, min_age_days return cursor.rowcount -def _archive_job_sizes(connection, where_clause, params): +def _archive_job_sizes(connection: sqlite3.Connection, where_clause: str, params: Sequence[Any]) -> None: """Copy the measured folder sizes of the jobs a purge is about to delete into job_size_archive.""" connection.execute( f""" @@ -1700,7 +1702,7 @@ def _archive_job_sizes(connection, where_clause, params): # Daemon control helpers (single row, id = 1). -def get_daemon_control(connection): +def get_daemon_control(connection: sqlite3.Connection) -> tuple[bool, Any, Optional[bool]]: """Return (paused, poll_now_request, log_override); defaults when the row doesn't exist yet. log_override is None (follow config.json), True (force on) or False (force off). @@ -1714,7 +1716,7 @@ def get_daemon_control(connection): return bool(row["paused"]), row["poll_now_request"], None if log_override is None else bool(log_override) -def set_daemon_log_override(connection, override: Optional[bool]): +def set_daemon_log_override(connection: sqlite3.Connection, override: Optional[bool]) -> None: """Set the terminal-log override (None = follow config.json), leaving the rest untouched.""" connection.execute( """ @@ -1726,7 +1728,7 @@ def set_daemon_log_override(connection, override: Optional[bool]): connection.commit() -def get_daemon_push_override(connection) -> Optional[bool]: +def get_daemon_push_override(connection: sqlite3.Connection) -> Optional[bool]: """Return the viewer-push override: None (follow config.json), True (force on) or False (force off).""" row = connection.execute("SELECT push_override FROM daemon_control WHERE id = 1").fetchone() if row is None or row["push_override"] is None: @@ -1734,7 +1736,7 @@ def get_daemon_push_override(connection) -> Optional[bool]: return bool(row["push_override"]) -def set_daemon_push_override(connection, override: Optional[bool]): +def set_daemon_push_override(connection: sqlite3.Connection, override: Optional[bool]) -> None: """Set the viewer-push override (None = follow config.json), leaving the rest untouched.""" connection.execute( """ @@ -1746,7 +1748,7 @@ def set_daemon_push_override(connection, override: Optional[bool]): connection.commit() -def set_daemon_paused(connection, paused: bool): +def set_daemon_paused(connection: sqlite3.Connection, paused: bool) -> None: """Set the pause flag, leaving any pending poll-now request untouched.""" connection.execute( """ @@ -1758,7 +1760,7 @@ def set_daemon_paused(connection, paused: bool): connection.commit() -def set_daemon_poll_now_request(connection, stamp: float) -> float: +def set_daemon_poll_now_request(connection: sqlite3.Connection, stamp: float) -> float: """Store a poll-now request stamp, guaranteed to differ from the previous one. Returns the stamp actually stored. @@ -1782,13 +1784,13 @@ def set_daemon_poll_now_request(connection, stamp: float) -> float: return stored -def get_daemon_check_folders_request(connection): +def get_daemon_check_folders_request(connection: sqlite3.Connection) -> Any | None: """Return the epoch stamp of the latest check-folders request, or None when never requested.""" row = connection.execute("SELECT check_folders_request FROM daemon_control WHERE id = 1").fetchone() return None if row is None else row["check_folders_request"] -def set_daemon_check_folders_request(connection, stamp: float) -> float: +def set_daemon_check_folders_request(connection: sqlite3.Connection, stamp: float) -> float: """Store a check-folders request stamp, guaranteed to differ from the previous one. Returns the stored stamp.""" connection.execute("INSERT OR IGNORE INTO daemon_control (id) VALUES (1)") connection.execute( @@ -1807,13 +1809,13 @@ def set_daemon_check_folders_request(connection, stamp: float) -> float: return stored -def get_daemon_single_request(connection): +def get_daemon_single_request(connection: sqlite3.Connection) -> Any | None: """Return the epoch stamp of the latest poll-one-item request, or None when never requested.""" row = connection.execute("SELECT single_request FROM daemon_control WHERE id = 1").fetchone() return None if row is None else row["single_request"] -def set_daemon_single_request(connection, stamp: float) -> float: +def set_daemon_single_request(connection: sqlite3.Connection, stamp: float) -> float: """Store a poll-one-item request stamp, guaranteed to differ from the previous one. Returns the stored stamp.""" connection.execute("INSERT OR IGNORE INTO daemon_control (id) VALUES (1)") connection.execute( @@ -1832,13 +1834,13 @@ def set_daemon_single_request(connection, stamp: float) -> float: return stored -def get_daemon_stop_request(connection): +def get_daemon_stop_request(connection: sqlite3.Connection) -> Any | None: """Return the epoch stamp of the latest graceful-stop request, or None when never requested.""" row = connection.execute("SELECT stop_request FROM daemon_control WHERE id = 1").fetchone() return None if row is None else row["stop_request"] -def set_daemon_stop_request(connection, stamp: float) -> float: +def set_daemon_stop_request(connection: sqlite3.Connection, stamp: float) -> float: """Store a stop request stamp, guaranteed to differ from the previous one. Returns the stored stamp.""" connection.execute("INSERT OR IGNORE INTO daemon_control (id) VALUES (1)") connection.execute( diff --git a/modules/log_files.py b/modules/log_files.py index da5a415..23d48cb 100644 --- a/modules/log_files.py +++ b/modules/log_files.py @@ -10,7 +10,7 @@ import os import re from datetime import datetime -from typing import Optional +from typing import Optional, TextIO from modules.dry_run_mode import is_dry_run @@ -88,7 +88,7 @@ def __init__(self, directory: str, max_bytes: int, keep_files: int) -> None: self._rollover_failed = False @staticmethod - def _open(path: str): + def _open(path: str) -> TextIO: """Open a log file for appending, line-buffered, UTF-8.""" return open(path, "a", encoding="utf-8", buffering=1) diff --git a/modules/mailbox_listener.py b/modules/mailbox_listener.py index 5907d9a..c4309b2 100644 --- a/modules/mailbox_listener.py +++ b/modules/mailbox_listener.py @@ -3,6 +3,7 @@ import os import re import time +from typing import Any, Optional, Sequence from modules.config_manager import get_gmail_folder from dotenv import load_dotenv from modules.dry_run_mode import is_dry_run @@ -18,7 +19,7 @@ raise ValueError("GMAIL_USER and GMAIL_APP_PASSWORD environment variables must be set") -def decode_email_subject(subject_raw): +def decode_email_subject(subject_raw: Optional[str]) -> str | Any: """Decode MIME-encoded email subjects, including emojis and special characters.""" if not subject_raw: return "" @@ -38,7 +39,7 @@ def decode_email_subject(subject_raw): return subject -def parse_github_subject(subject): +def parse_github_subject(subject: str) -> tuple[Optional[str], Optional[str], Optional[str], bool]: """Parse a GitHub release subject. Returns: @@ -70,7 +71,7 @@ def parse_github_subject(subject): return None, None, None, False -def get_pending_notifications(limit=None): +def get_pending_notifications(limit: Optional[int] = None) -> list[Any]: """Fetch unread GitHub release notifications from Gmail. Returns a list of dicts with the keys repo, tag, release_type and email_id. @@ -136,7 +137,7 @@ def get_pending_notifications(limit=None): return notifications -def mark_as_read_and_delete(email_ids, max_attempts=2, retry_delay_seconds=5): +def mark_as_read_and_delete(email_ids: Sequence[Any], max_attempts: int = 2, retry_delay_seconds: float = 5) -> bool: """Mark a list of emails as read and move them to Trash. Retries transient IMAP store/move failures once before giving up. @@ -185,7 +186,7 @@ def mark_as_read_and_delete(email_ids, max_attempts=2, retry_delay_seconds=5): -def move_unread_to_trash(email_ids, max_attempts=2, retry_delay_seconds=5): +def move_unread_to_trash(email_ids: Sequence[Any], max_attempts: int = 2, retry_delay_seconds: float = 5) -> bool: """Move unread emails to Trash without changing their seen state. Skipped entirely in dry-run mode; the mailbox is left untouched. @@ -227,7 +228,7 @@ def move_unread_to_trash(email_ids, max_attempts=2, retry_delay_seconds=5): return False -def check_releases(): +def check_releases() -> None: """Run the release check flow via get_pending_notifications.""" notifications = get_pending_notifications() print(f"\nFound {len(notifications)} release notification(s).") diff --git a/modules/queue_reports.py b/modules/queue_reports.py index a00ffc3..5f85782 100644 --- a/modules/queue_reports.py +++ b/modules/queue_reports.py @@ -4,8 +4,9 @@ import json import sys import time +import sqlite3 from datetime import datetime, timedelta -from typing import Optional +from typing import Any, Optional, Sequence from modules.db_manager import get_previous_successful_completed_job, open_database from modules.stats import format_bytes @@ -29,7 +30,7 @@ PURGE_AGE_PREVIEW_THRESHOLDS_DAYS = (7, 30, 90, 180, 365) -def _format_timestamp(unix_timestamp): +def _format_timestamp(unix_timestamp: float) -> str: """Format a Unix timestamp for console output.""" if unix_timestamp is None: return "-" @@ -39,7 +40,7 @@ def _format_timestamp(unix_timestamp): return "-" -def _build_skipped_item_preview(row) -> SkippedItemPreview: +def _build_skipped_item_preview(row: sqlite3.Row) -> SkippedItemPreview: """Build a typed skipped-item preview payload from a database row.""" return { "attempt_count": int(row["attempt_count"]), @@ -52,7 +53,7 @@ def _build_skipped_item_preview(row) -> SkippedItemPreview: def _build_next_pending_job_payload( - row, + row: sqlite3.Row, skip_detail_count: int, skipped_items_preview: list[SkippedItemPreview], ) -> NextPendingJobPayload: @@ -82,7 +83,7 @@ def _build_next_pending_job_payload( def _build_queue_job_payload( - row, + row: sqlite3.Row, skip_detail_count: int, skipped_items_preview: list[SkippedItemPreview], previous_success_tag: Optional[str] = None, @@ -120,12 +121,12 @@ def _build_queue_job_payload( } -def _build_top_failed_repo_payload(row) -> TopFailedRepoPayload: +def _build_top_failed_repo_payload(row: sqlite3.Row) -> TopFailedRepoPayload: """Build a typed top-failed-repo payload.""" return {"repo": row["repo"], "failed_count": int(row["failed_count"])} -def _build_top_successful_repo_payload(row) -> TopSuccessfulRepoPayload: +def _build_top_successful_repo_payload(row: sqlite3.Row) -> TopSuccessfulRepoPayload: """Build a typed top-successful-repo payload.""" terminal_count = int(row["terminal_count"] or 0) success_count = int(row["success_count"] or 0) @@ -145,17 +146,17 @@ def _build_top_successful_repo_payload(row) -> TopSuccessfulRepoPayload: } -def _build_top_skipped_item_payload(row) -> TopSkippedItemPayload: +def _build_top_skipped_item_payload(row: sqlite3.Row) -> TopSkippedItemPayload: """Build a typed top-skipped-item payload.""" return {"item_label": row["item_label"], "skip_count": int(row["skip_count"])} -def _build_skip_reason_payload(row) -> SkipReasonPayload: +def _build_skip_reason_payload(row: sqlite3.Row) -> SkipReasonPayload: """Build a typed skip-reason payload.""" return {"reason": row["reason"], "count": int(row["count"])} -def _build_purgeable_job_age_summary(row, now_timestamp: float) -> Optional[PurgeableJobAgeSummary]: +def _build_purgeable_job_age_summary(row: sqlite3.Row, now_timestamp: float) -> Optional[PurgeableJobAgeSummary]: """Build a typed age summary for a purgeable (non-PENDING) job row.""" if row is None: return None @@ -174,7 +175,7 @@ def _build_purgeable_job_age_summary(row, now_timestamp: float) -> Optional[Purg def _build_queue_report_payload( total_jobs: int, - summary_rows, + summary_rows: Sequence[Any], terminal_jobs: int, success_jobs: int, skip_jobs: int, @@ -182,10 +183,10 @@ def _build_queue_report_payload( retry_pending_jobs: int, supersede_finalized_jobs: int, supersede_incomplete_moved_jobs: int, - failed_repo_rows, - successful_repo_rows, - top_skipped_items, - skip_reason_rows, + failed_repo_rows: Sequence[Any], + successful_repo_rows: Sequence[Any], + top_skipped_items: Sequence[Any], + skip_reason_rows: Sequence[Any], oldest_purgeable_job: Optional[PurgeableJobAgeSummary] = None, newest_purgeable_job: Optional[PurgeableJobAgeSummary] = None, purge_age_preview: Optional[list[PurgeAgePreviewEntry]] = None, diff --git a/modules/queue_worker.py b/modules/queue_worker.py index f014370..1f7571e 100644 --- a/modules/queue_worker.py +++ b/modules/queue_worker.py @@ -1,6 +1,7 @@ """The processing engine: ingests notifications into jobs and runs due jobs (PENDING -> COMPLETED/FAILED/SUPERSEDED).""" import requests +import sqlite3 import sys import time import os @@ -61,7 +62,7 @@ QueueCycleStats, SkippedItemPayload, ) -from typing import Callable, Literal, Optional, Tuple, Union, cast, overload +from typing import Any, Callable, cast, Iterable, Literal, Optional, overload, Tuple, Union GITHUB_API_VERSION = "2022-11-28" @@ -76,7 +77,7 @@ def _finalize_staged_release_folder( tag: Optional[str] = None, commit: Optional[str] = None, write_complete_log: bool = True, - connection=None, + connection: Optional[sqlite3.Connection] = None, job_id: Optional[int] = None, ) -> Optional[str]: """Move a terminal job's staging folder from Processing to complete destination. @@ -104,7 +105,7 @@ def _finalize_staged_release_folder( return done_dir -def _record_folder_size(connection, job_id: Optional[int], folder: str) -> None: +def _record_folder_size(connection: Optional[sqlite3.Connection], job_id: Optional[int], folder: str) -> None: """Measure a job's moved folder (finished or in Partial) and add it to the job; a failure only warns.""" if connection is None or job_id is None or is_dry_run(): return @@ -122,7 +123,7 @@ def _handle_working_dir_relocation( repo: str, tag: str, job_id: int, - connection=None, + connection: Optional[sqlite3.Connection] = None, ) -> None: """Warn about and quarantine a Processing folder abandoned by a mid-flight rename. @@ -168,7 +169,7 @@ def _is_terminal_skip_reason(skip_reason: Optional[str]) -> bool: def _preserve_best_known_counters( - row, + row: Any, downloaded_count: int, skipped_count: int, total_items: int, @@ -232,11 +233,11 @@ def _has_all_release_items_accounted( def _handle_superseded_pending_job_artifacts( - row, + row: sqlite3.Row, repo: str, tag: str, reason_code: str, - connection=None, + connection: Optional[sqlite3.Connection] = None, ) -> str: """Return artifact handling outcome: finalized, quarantined, or none.""" job_id = int(row["id"]) @@ -338,7 +339,7 @@ def _handle_superseded_pending_job_artifacts( def _finalize_terminal_skip_job( - row, + row: sqlite3.Row, repo: str, tag: str, commit: Optional[str], @@ -346,7 +347,7 @@ def _finalize_terminal_skip_job( current_skipped_count: int, current_total_items: int, current_working_dir: Optional[str], - connection=None, + connection: Optional[sqlite3.Connection] = None, ) -> Tuple[int, int, int]: """Finalize a job whose release/tag disappeared, moving any real staged files. @@ -394,7 +395,7 @@ def _finalize_terminal_skip_job( def _warn_if_file_count_changed_from_previous_success( - connection, + connection: sqlite3.Connection, job_id: int, repo: str, tag: str, @@ -506,7 +507,7 @@ def _empty_ingest_cycle_stats() -> IngestCycleStats: def ingest_notifications_once( - connection, + connection: sqlite3.Connection, github_token: Optional[str], notification_limit: Optional[int] = None, should_pause: Optional[Callable[[], bool]] = None, @@ -804,7 +805,7 @@ def ingest_notifications_once( return _empty_ingest_cycle_stats(), None -def process_selected_pending_jobs(connection, github_token: Optional[str], job_ids) -> tuple[int, list[int], list[int]]: +def process_selected_pending_jobs(connection: sqlite3.Connection, github_token: Optional[str], job_ids: Iterable[int]) -> tuple[int, list[int], list[int]]: """Run selected pending jobs immediately without changing their retry schedule.""" normalized_ids = sorted({int(job_id) for job_id in (job_ids or [])}) if not normalized_ids: @@ -1088,7 +1089,7 @@ def _empty_queue_cycle_stats() -> QueueCycleStats: def process_queue_once( - connection, + connection: sqlite3.Connection, github_token: Optional[str], limit: Optional[int] = None, job_filter_ids: Optional[list[int]] = None, @@ -1105,7 +1106,7 @@ def process_queue_once( def _process_queue_once( - connection, + connection: sqlite3.Connection, github_token: Optional[str], limit: Optional[int] = None, job_filter_ids: Optional[list[int]] = None, @@ -1447,7 +1448,7 @@ def _process_queue_once( def run_ingest_and_queue_cycle( - connection, + connection: sqlite3.Connection, github_token: Optional[str], should_pause: Optional[Callable[[], bool]] = None, ) -> None: @@ -1482,7 +1483,7 @@ def run_ingest_and_queue_cycle( ) -def run_single_cycle(connection, github_token: Optional[str]) -> None: +def run_single_cycle(connection: sqlite3.Connection, github_token: Optional[str]) -> None: """Ingest at most one new notification and process at most one queue item. Prefers the job just queued from the notification, else the oldest due job already in the queue. In dry-run diff --git a/pyrightconfig.json b/pyrightconfig.json index 58b413e..591ab20 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -1,5 +1,6 @@ { "include": ["main.py", "main_gui.py", "viewer", "modules", "tests"], "extraPaths": ["viewer"], - "typeCheckingMode": "standard" + "typeCheckingMode": "standard", + "reportMissingParameterType": "error" } diff --git a/tests/test_annotations.py b/tests/test_annotations.py new file mode 100644 index 0000000..5e4aadb --- /dev/null +++ b/tests/test_annotations.py @@ -0,0 +1,53 @@ +"""Every function has typed parameters and a return type (the "Typing annotations required" rule in CLAUDE.md). + +Pyright's reportMissingParameterType (pyrightconfig.json) covers the parameters; Pyright has no rule for a missing +return type, so this test reads the source files itself. Run from the project root: python -m unittest discover -s tests -t . +""" +import ast +import os +import unittest +from typing import Iterator, List + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +SOURCE_FOLDERS = ("modules", "viewer", "tests") +SOURCE_FILES = ("main.py", "main_gui.py") + + +def python_files() -> Iterator[str]: + """Yield the path of every Python file the type check covers.""" + for name in SOURCE_FILES: + yield os.path.join(ROOT, name) + for folder in SOURCE_FOLDERS: + for name in sorted(os.listdir(os.path.join(ROOT, folder))): + if name.endswith(".py"): + yield os.path.join(ROOT, folder, name) + + +def untyped_functions(path: str) -> List[str]: + """Return 'file:line name (what)' for each function in the file without a return type or a parameter type.""" + with open(path, encoding="utf-8") as handle: + tree = ast.parse(handle.read()) + found = [] + where = os.path.relpath(path, ROOT).replace(os.sep, "/") + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if node.returns is None: + found.append(f"{where}:{node.lineno} {node.name} (return type)") + arguments = node.args + every = arguments.posonlyargs + arguments.args + arguments.kwonlyargs + every += [argument for argument in (arguments.vararg, arguments.kwarg) if argument is not None] + for argument in every: + if argument.annotation is None and argument.arg not in ("self", "cls"): + found.append(f"{where}:{argument.lineno} {node.name} (parameter {argument.arg})") + return found + + +class AnnotationTests(unittest.TestCase): + def test_every_function_has_typed_parameters_and_a_return_type(self) -> None: + missing = [item for path in python_files() for item in untyped_functions(path)] + self.assertEqual(missing, [], "add the missing type annotations:\n" + "\n".join(missing)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_asset_downloader_helpers.py b/tests/test_asset_downloader_helpers.py index 8e73874..d63d006 100644 --- a/tests/test_asset_downloader_helpers.py +++ b/tests/test_asset_downloader_helpers.py @@ -4,6 +4,7 @@ Everything runs in temporary folders (see MoveTestCase). Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import os import unittest from contextlib import closing @@ -161,7 +162,7 @@ def test_a_release_type_folder_that_cannot_be_read_is_skipped(self) -> None: real_scandir = os.scandir calls = {"n": 0} - def flaky(path): + def flaky(path: Any) -> Any: calls["n"] += 1 if calls["n"] == 2: # the first call lists the repository folder, the second one the type folder raise OSError("denied") diff --git a/tests/test_autostart.py b/tests/test_autostart.py index ec0fea4..1921995 100644 --- a/tests/test_autostart.py +++ b/tests/test_autostart.py @@ -3,6 +3,7 @@ Everything runs against a temp home folder, a fake systemctl and a fake registry; nothing real changes. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any, NoReturn import os import plistlib import subprocess @@ -61,35 +62,35 @@ class FakeRegistry: def __init__(self) -> None: self.values: dict[str, str] = {} - def _handle(self): + def _handle(self) -> Any: values = self.values class Handle: - def __enter__(self): + def __enter__(self) -> Any: return values - def __exit__(self, *_exc): + def __exit__(self, *_exc: Any) -> bool: return False return Handle() - def CreateKeyEx(self, _root, path, _reserved, _access): + def CreateKeyEx(self, _root: Any, path: Any, _reserved: Any, _access: Any) -> Any: assert path == autostart.WINDOWS_RUN_KEY return self._handle() - def OpenKey(self, _root, path, _reserved, _access): + def OpenKey(self, _root: Any, path: Any, _reserved: Any, _access: Any) -> Any: assert path == autostart.WINDOWS_RUN_KEY return self._handle() - def SetValueEx(self, key, name, _reserved, _kind, value): + def SetValueEx(self, key: Any, name: Any, _reserved: Any, _kind: Any, value: Any) -> None: key[name] = value - def QueryValueEx(self, key, name): + def QueryValueEx(self, key: Any, name: Any) -> Any: if name not in key: raise FileNotFoundError(name) return key[name], self.REG_SZ - def DeleteValue(self, key, name): + def DeleteValue(self, key: Any, name: Any) -> None: if name not in key: raise FileNotFoundError(name) del key[name] @@ -150,7 +151,7 @@ def test_install_reports_a_failed_enable(self) -> None: self.assertIn("enabling it failed", result.message) def test_a_missing_systemctl_is_reported_not_raised(self) -> None: - def missing(_command): + def missing(_command: Any) -> NoReturn: raise FileNotFoundError("systemctl") result = autostart.install_autostart(missing, platform="linux", home=self.home) @@ -205,7 +206,7 @@ class TaskSchedulerTests(unittest.TestCase): def setUp(self) -> None: self.registry, self.schtasks = FakeRegistry(), FakeSchtasks() - def install(self, trigger: str = "logon"): + def install(self, trigger: str = "logon") -> Any: return autostart.install_autostart( self.schtasks, self.registry, "win32", python="C:\\py\\python.exe", mode="task", trigger=trigger ) @@ -306,7 +307,7 @@ def test_install_status_uninstall(self) -> None: self.assertFalse(autostart.autostart_status(self.runner, platform="darwin", home=self.home, uid=501).installed) def test_a_failed_load_is_reported(self) -> None: - def failing(command): + def failing(command: Any) -> Any: return subprocess.CompletedProcess(command, 5 if command[1] == "bootstrap" else 0, "", "Bootstrap failed: 5") result = autostart.install_autostart(failing, platform="darwin", home=self.home, uid=501) diff --git a/tests/test_backup_manager.py b/tests/test_backup_manager.py index f4cddac..e2cbbc3 100644 --- a/tests/test_backup_manager.py +++ b/tests/test_backup_manager.py @@ -2,6 +2,7 @@ Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import io import json import os @@ -16,7 +17,7 @@ from modules import backup_manager, cli_commands, config_manager, db_manager, gui_forms, warning_types -def _settings(directory: str, **overrides) -> config_manager.BackupSettings: +def _settings(directory: str, **overrides: Any) -> config_manager.BackupSettings: settings: config_manager.BackupSettings = { "enabled": True, "every_hours": 24, "keep_files": 3, "directory": directory, "include_env": False, } @@ -50,7 +51,7 @@ def setUp(self) -> None: dry.start() self.addCleanup(dry.stop) - def make(self, when: datetime, **overrides) -> backup_manager.BackupResult: + def make(self, when: datetime, **overrides: Any) -> backup_manager.BackupResult: return backup_manager.create_backup(_settings(self.backups, **overrides), now=when, snapshot=_fake_snapshot) @@ -192,7 +193,7 @@ def test_backup_is_a_known_warning_type(self) -> None: class DaemonHookTests(unittest.TestCase): - def check(self, result): + def check(self, result: Any) -> Any: import main # imported here: it reads .env at import time warnings: list[tuple[str, str]] = [] @@ -219,7 +220,7 @@ def test_failure_is_recorded_as_a_backup_warning(self) -> None: def test_an_exception_never_escapes(self) -> None: import main - def boom(): + def boom() -> None: raise RuntimeError("x") with mock.patch.object(main, "run_scheduled_backup", boom), mock.patch.object(main, "is_dry_run", lambda: False), \ diff --git a/tests/test_cli_dispatch.py b/tests/test_cli_dispatch.py index cead21a..ffd5554 100644 --- a/tests/test_cli_dispatch.py +++ b/tests/test_cli_dispatch.py @@ -4,6 +4,7 @@ (daemon control, database, autostart, shortcuts, doctor, ...) is replaced by a recorder, so nothing real is touched. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import contextlib import io import json @@ -22,13 +23,13 @@ def setUp(self) -> None: self.daemon_running = True self.smoke_ran = False - def patch(self, name: str, value) -> None: + def patch(self, name: str, value: Any) -> None: patcher = mock.patch.object(cli_commands, name, value) patcher.start() self.addCleanup(patcher.stop) - def recorder(self, label: str, result=None): - def record(*args, **kwargs): + def recorder(self, label: str, result: Any = None) -> Any: + def record(*args: Any, **kwargs: Any) -> Any: self.calls.append((label, args, kwargs)) return result return record @@ -119,7 +120,7 @@ def test_it_starts_one_when_none_runs(self) -> None: def test_a_launch_failure_is_reported(self) -> None: self.daemon_running = False - def broken(): + def broken() -> None: raise OSError("no python") with mock.patch("modules.daemon_launcher.start_daemon", broken): @@ -223,7 +224,7 @@ def test_new_viewer_token_prints_a_usable_token_and_needs_no_daemon(self) -> Non self.assertNotEqual(token, self.run_cli("--new-viewer-token")[1].strip()) def test_a_locked_database_is_reported_instead_of_crashing(self) -> None: - def locked(*args): + def locked(*args: Any) -> None: raise sqlite3.OperationalError("database is locked") self.patch("request_poll_now", locked) @@ -373,7 +374,7 @@ def test_move_complete_to_destination(self) -> None: _, out, _ = self.run_cli("--move-complete-to-destination", "--json") self.assertEqual(json.loads(out), {"moved": 2}) - def events(self): + def events(self) -> Any: return [ {"created_at_readable": "2026-10-08 10:00", "category": "COMPLETE", "event_type": "X", "message": "done"}, {"created_at_readable": "2026-10-08 11:00", "category": None, "event_type": "WARNING", "message": "hmm"}, diff --git a/tests/test_cli_unused_options.py b/tests/test_cli_unused_options.py index ceaccc0..6ca3f11 100644 --- a/tests/test_cli_unused_options.py +++ b/tests/test_cli_unused_options.py @@ -2,6 +2,7 @@ Only argument parsing is exercised. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any, Iterator import contextlib import io import unittest @@ -9,7 +10,7 @@ from modules import cli_commands -def parse(*args: str): +def parse(*args: str) -> Any: return cli_commands.parse_cli_args(list(args), "test") @@ -22,7 +23,7 @@ def rejected(*args: str) -> str: @contextlib.contextmanager -def self_assert_exit(): +def self_assert_exit() -> Iterator[None]: try: yield except SystemExit as exc: diff --git a/tests/test_config_getters.py b/tests/test_config_getters.py index 32e66ae..a634ae9 100644 --- a/tests/test_config_getters.py +++ b/tests/test_config_getters.py @@ -4,6 +4,7 @@ No real config.json is read: every getter takes the config as an argument here. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import os import unittest @@ -83,7 +84,7 @@ def test_values_and_negative_clamping(self) -> None: class FolderSettingTests(unittest.TestCase): - def folders(self, **names): + def folders(self, **names: Any) -> Any: return cm.get_folder_settings({"folders": names}) def test_defaults(self) -> None: @@ -153,10 +154,10 @@ def test_values_and_clamping(self) -> None: class GuiTimingTests(unittest.TestCase): - def refresh(self, value): + def refresh(self, value: Any) -> Any: return cm.get_gui_refresh_seconds({"gui": {"refresh_seconds": value}}) - def message(self, value): + def message(self, value: Any) -> Any: return cm.get_gui_status_message_seconds({"gui": {"status_message_seconds": value}}) def test_defaults(self) -> None: @@ -187,7 +188,7 @@ class FingerprintTests(unittest.TestCase): "terminal_log": {"enabled": False}, } - def fingerprint(self, **sections) -> str: + def fingerprint(self, **sections: Any) -> str: config = {**self.BASE} for key, value in sections.items(): config[key] = {**config.get(key, {}), **value} if isinstance(value, dict) else value diff --git a/tests/test_connection_tests.py b/tests/test_connection_tests.py index 426e107..df7dd60 100644 --- a/tests/test_connection_tests.py +++ b/tests/test_connection_tests.py @@ -2,6 +2,7 @@ No network is used. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any, Self import socket import unittest from types import SimpleNamespace @@ -21,36 +22,36 @@ class FakeServer: - def __init__(self, login_error=None, folders=FOLDERS, unread=(1, 2, 3), select_error=None): + def __init__(self, login_error: Any = None, folders: Any = FOLDERS, unread: Any = (1, 2, 3), select_error: Any = None) -> None: self.login_error, self.folders, self.unread, self.select_error = login_error, folders, unread, select_error self.calls = [] - def __enter__(self): + def __enter__(self) -> Self: return self - def __exit__(self, *exc): + def __exit__(self, *exc: Any) -> bool: return False - def login(self, user, password): + def login(self, user: Any, password: Any) -> None: self.calls.append(("login", user, password)) if self.login_error: raise self.login_error - def list_folders(self): + def list_folders(self) -> Any: return self.folders - def select_folder(self, folder, readonly=False): + def select_folder(self, folder: Any, readonly: bool = False) -> None: self.calls.append(("select", folder, readonly)) if self.select_error: raise self.select_error - def search(self, criteria): + def search(self, criteria: Any) -> Any: self.calls.append(("search", criteria)) return list(self.unread) -def factory(server=None, error=None, hosts=None): - def make(host): +def factory(server: Any = None, error: Any = None, hosts: Any = None) -> Any: + def make(host: Any) -> Any: if hosts is not None: hosts.append(host) if error: @@ -119,12 +120,12 @@ def test_an_empty_address_or_password_is_not_sent_anywhere(self) -> None: self.assertFalse(connection_tests.check_gmail("me@gmail.com", "", "", make).ok) -def response(status, payload=None): +def response(status: Any, payload: Any = None) -> Any: return SimpleNamespace(status_code=status, json=lambda: payload) -def getter(result=None, error=None, seen=None): - def get(url, headers=None, timeout=None): +def getter(result: Any = None, error: Any = None, seen: Any = None) -> Any: + def get(url: Any, headers: Any = None, timeout: Any = None) -> Any: if seen is not None: seen.update(headers or {}) if error: diff --git a/tests/test_daemon_control.py b/tests/test_daemon_control.py index cd63021..c678a64 100644 --- a/tests/test_daemon_control.py +++ b/tests/test_daemon_control.py @@ -8,7 +8,7 @@ import sqlite3 import tempfile import unittest -from typing import Optional +from typing import Any, NoReturn, Optional from unittest import mock from modules import daemon_control, db_manager, queue_worker @@ -58,7 +58,7 @@ def setUp(self) -> None: self.connection = db_manager.open_database() self.addCleanup(self.connection.close) - def read(self): + def read(self) -> Any: return daemon_control.read_control_state(self.connection) def test_stop_request_round_trip_keeps_other_fields(self) -> None: @@ -206,7 +206,7 @@ def test_watcher_keeps_last_state_when_database_errors(self) -> None: calls = {"n": 0} real_read = daemon_control.read_control_state - def flaky(connection): + def flaky(connection: Any) -> Any: calls["n"] += 1 if calls["n"] > 1: raise sqlite3.OperationalError("database is locked") @@ -219,7 +219,7 @@ def flaky(connection): class ControlWatcherTests(unittest.TestCase): - def make_watcher(self, state_for_time, fake=None): + def make_watcher(self, state_for_time: Any, fake: Any = None) -> Any: fake = fake or FakeTime() watcher = ControlWatcher( read_state=lambda: state_for_time(fake.now), @@ -307,7 +307,7 @@ def test_idle_wait_honours_poll_now_while_paused(self) -> None: self.assertTrue(watcher.forced_while_paused) def test_request_during_pause_polls_at_once_and_the_pause_goes_on(self) -> None: - def state(t): + def state(t: Any) -> Any: return _state(paused=2 <= t, request=5.0 if t >= 3 else None, stop=9.0 if t >= 100 else None) watcher, fake = self.make_watcher(state) @@ -380,7 +380,7 @@ def test_checkpoint_interrupts_work_when_stop_requested(self) -> None: self.assertTrue(watcher.stop_requested) def test_disabled_watcher_ignores_control_file(self) -> None: - def explode(): + def explode() -> NoReturn: raise AssertionError("control file must not be read") fake = FakeTime() @@ -391,7 +391,7 @@ def explode(): class CheckFoldersRequestTests(unittest.TestCase): """A check-folders request runs the callback once and does not disturb the countdown.""" - def make(self, state_for_time, on_check): + def make(self, state_for_time: Any, on_check: Any) -> Any: fake = FakeTime() watcher = ControlWatcher( read_state=lambda: state_for_time(fake.now), @@ -441,7 +441,7 @@ def test_round_trip_through_the_database(self) -> None: class LogOverrideWatcherTests(unittest.TestCase): """The watcher reports live log switches to its callback exactly once per change.""" - def make(self, override_for_time): + def make(self, override_for_time: Any) -> Any: fake = FakeTime() seen = [] watcher = ControlWatcher( diff --git a/tests/test_daemon_stop.py b/tests/test_daemon_stop.py index 50d1231..853c53f 100644 --- a/tests/test_daemon_stop.py +++ b/tests/test_daemon_stop.py @@ -3,6 +3,7 @@ Nothing here touches the real mailbox, daemon lock or state.db. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import contextlib import io import json @@ -58,7 +59,7 @@ def test_stop_during_the_countdown_exits_the_loop(self) -> None: def test_stop_during_a_cycle_exits_without_waiting(self) -> None: calls = [] - def cycle(connection, token, should_pause=None): + def cycle(connection: Any, token: Any, should_pause: Any = None) -> None: calls.append(1) daemon_control.request_stop() assert should_pause is not None @@ -76,7 +77,7 @@ def test_stop_request_from_an_earlier_run_does_not_stop_a_new_daemon(self) -> No daemon_control.request_stop() # left behind in state.db by a previous run cycles = [] - def cycle(connection, token, should_pause=None): + def cycle(connection: Any, token: Any, should_pause: Any = None) -> None: cycles.append(1) if len(cycles) == 2: raise KeyboardInterrupt # end the test loop after the second cycle diff --git a/tests/test_download_release.py b/tests/test_download_release.py index 2382eda..6d5ad8b 100644 --- a/tests/test_download_release.py +++ b/tests/test_download_release.py @@ -15,7 +15,7 @@ import types import unittest from email.utils import formatdate -from typing import Any, cast +from typing import Any, cast, Iterator, NoReturn, Self from unittest import mock import requests @@ -29,7 +29,7 @@ class FakeResponse: - def __init__(self, status_code=200, body=b"", headers=None, json_data=None, text=None, fail_after=None): + def __init__(self, status_code: int = 200, body: Any = b"", headers: Any = None, json_data: Any = None, text: Any = None, fail_after: Any = None) -> None: self.status_code = status_code self.content = body self.headers = headers or {} @@ -37,31 +37,31 @@ def __init__(self, status_code=200, body=b"", headers=None, json_data=None, text self.text = text if text is not None else body.decode("utf-8", errors="replace") self.fail_after = fail_after # raise this error while streaming - def json(self): + def json(self) -> Any: return self._json - def raise_for_status(self): + def raise_for_status(self) -> None: if self.status_code >= 400: raise requests.HTTPError(f"{self.status_code} error") - def iter_content(self, chunk_size=8192): + def iter_content(self, chunk_size: int = 8192) -> Iterator[bytes]: if self.fail_after is not None: yield self.content[:2] raise self.fail_after for start in range(0, len(self.content), chunk_size): yield self.content[start:start + chunk_size] - def __enter__(self): + def __enter__(self) -> Self: return self - def __exit__(self, *exc): + def __exit__(self, *exc: Any) -> bool: return False class FakeGitHub: """Routes for one release. `files` maps a download URL to its bytes.""" - def __init__(self): + def __init__(self) -> None: self.release = { "tag_name": TAG, "name": "First release", @@ -92,7 +92,7 @@ def __init__(self): self.requests: list[tuple[str, str]] = [] # --- API (requests.get) ------------------------------------------------------------- - def api_get(self, url, headers=None, timeout=None, **kwargs): + def api_get(self, url: Any, headers: Any = None, timeout: Any = None, **kwargs: Any) -> Any: self.requests.append(("GET", url)) if "/releases/tags/" in url: if self.release_status == 200: @@ -108,26 +108,26 @@ def api_get(self, url, headers=None, timeout=None, **kwargs): return FakeResponse(404) # --- Session ------------------------------------------------------------------------ - def make_session(self): + def make_session(self) -> Any: github = self class Session: - def __init__(self): + def __init__(self) -> None: self.headers = {} - def __enter__(self): + def __enter__(self) -> Self: return self - def __exit__(self, *exc): + def __exit__(self, *exc: Any) -> bool: return False - def head(self, url, **kwargs): + def head(self, url: Any, **kwargs: Any) -> Any: github.requests.append(("HEAD", url)) if url not in github.files: return FakeResponse(404) return FakeResponse(200, headers=github.headers_for(url)) - def get(self, url, **kwargs): + def get(self, url: Any, **kwargs: Any) -> Any: github.requests.append(("GET", url)) errors = github.get_errors.get(url) if errors: @@ -138,7 +138,7 @@ def get(self, url, **kwargs): return Session() - def headers_for(self, url): + def headers_for(self, url: Any) -> Any: headers = { "Content-Length": str(len(self.files[url])), "Last-Modified": formatdate(LAST_MODIFIED_EPOCH, usegmt=True), @@ -147,7 +147,7 @@ def headers_for(self, url): headers.update(self.file_headers.get(url, {})) return {key: value for key, value in headers.items() if value is not None} # None = leave the header out - def module(self): + def module(self) -> Any: return types.SimpleNamespace( get=self.api_get, Session=self.make_session, @@ -239,7 +239,7 @@ def test_an_api_error_gives_none_and_is_reported(self) -> None: def test_the_token_is_sent_when_there_is_one(self) -> None: seen = {} - def capture(url, headers=None, **kwargs): + def capture(url: Any, headers: Any = None, **kwargs: Any) -> Any: seen.update(headers or {}) return FakeResponse(200, json_data={}) @@ -499,13 +499,13 @@ def test_a_failed_head_request_does_not_stop_the_download(self) -> None: # The HEAD is only an optimisation: without it the file is simply downloaded. session_factory = self.github.make_session - def session_with_failing_head(): + def session_with_failing_head() -> Any: session = session_factory() - def head(url, **kwargs): + def head(url: Any, **kwargs: Any) -> NoReturn: raise requests.ConnectionError("no head") - session.head = head + session.head = head # pyright: ignore[reportAttributeAccessIssue] # the test swaps in a fake return session module = self.github.module() @@ -519,13 +519,13 @@ def test_a_current_file_is_still_skipped_when_only_the_get_response_can_tell(sel self.run_download() session_factory = self.github.make_session - def session_with_failing_head(): + def session_with_failing_head() -> Any: session = session_factory() - def head(url, **kwargs): + def head(url: Any, **kwargs: Any) -> NoReturn: raise requests.ConnectionError("no head") - session.head = head + session.head = head # pyright: ignore[reportAttributeAccessIssue] # the test swaps in a fake return session module = self.github.module() @@ -604,7 +604,7 @@ def test_the_token_is_sent_with_the_api_calls_of_a_download(self) -> None: seen = [] original = self.github.api_get - def spy(url, headers=None, **kwargs): + def spy(url: Any, headers: Any = None, **kwargs: Any) -> Any: seen.append((headers or {}).get("Authorization")) return original(url, headers=headers, **kwargs) diff --git a/tests/test_env_manager.py b/tests/test_env_manager.py index a8fcb13..248c8ab 100644 --- a/tests/test_env_manager.py +++ b/tests/test_env_manager.py @@ -2,6 +2,7 @@ Everything happens in a temporary folder. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import os import stat import tempfile @@ -154,7 +155,7 @@ def test_a_briefly_locked_file_is_retried_on_windows(self) -> None: real_replace = os.replace calls = {"n": 0} - def flaky(source, target): + def flaky(source: Any, target: Any) -> Any: calls["n"] += 1 if calls["n"] < 3: raise PermissionError("in use") diff --git a/tests/test_folder_check_trigger.py b/tests/test_folder_check_trigger.py index d64c396..a73a960 100644 --- a/tests/test_folder_check_trigger.py +++ b/tests/test_folder_check_trigger.py @@ -3,6 +3,7 @@ Nothing here touches the real mailbox, daemon lock, mapping or state.db. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import contextlib import io import os @@ -39,7 +40,7 @@ def setUp(self) -> None: patcher.start() self.addCleanup(patcher.stop) - def run_loop(self, every: int, cycle) -> str: + def run_loop(self, every: int, cycle: Any) -> str: output = io.StringIO() with mock.patch.object(main, "get_destination_check_every_n_polls", lambda: every), \ mock.patch.object(main, "run_ingest_and_queue_cycle", cycle), \ @@ -48,10 +49,10 @@ def run_loop(self, every: int, cycle) -> str: return output.getvalue() @staticmethod - def stop_after(cycles: int): + def stop_after(cycles: int) -> Any: seen = [] - def cycle(connection, token, should_pause=None): + def cycle(connection: Any, token: Any, should_pause: Any = None) -> None: seen.append(1) if len(seen) == cycles: daemon_control.request_stop() @@ -62,7 +63,7 @@ def test_the_check_runs_at_start_before_the_first_poll(self) -> None: order = [] self.destinations.side_effect = lambda: order.append("check") or 0 - def cycle(connection, token, should_pause=None): + def cycle(connection: Any, token: Any, should_pause: Any = None) -> None: order.append("poll") daemon_control.request_stop() @@ -74,7 +75,7 @@ def cycle(connection, token, should_pause=None): def test_it_repeats_every_n_polls(self) -> None: polls = [] - def cycle(connection, token, should_pause=None): + def cycle(connection: Any, token: Any, should_pause: Any = None) -> None: polls.append(1) if len(polls) == 5: raise KeyboardInterrupt # ends the loop; the check after poll 5 never runs @@ -98,7 +99,7 @@ def test_a_request_during_the_countdown_runs_the_check_and_keeps_waiting(self) - # request would then count as already seen. timers: list[threading.Timer] = [] - def cycle(connection, token, should_pause=None): + def cycle(connection: Any, token: Any, should_pause: Any = None) -> None: if timers: return timers.extend(( diff --git a/tests/test_folder_naming_and_moves.py b/tests/test_folder_naming_and_moves.py index 9a0c436..d6b16b0 100644 --- a/tests/test_folder_naming_and_moves.py +++ b/tests/test_folder_naming_and_moves.py @@ -11,7 +11,7 @@ import os import tempfile import unittest -from typing import Optional, cast +from typing import Any, cast, Optional from unittest import mock from modules import asset_downloader, config_manager, db_manager, mapping_manager @@ -157,7 +157,7 @@ def setUp(self) -> None: self.addCleanup(patcher.stop) self.dry_run = False - def write_mapping(self, **entry_fields) -> None: + def write_mapping(self, **entry_fields: Any) -> None: entries = [] if entry_fields is not None and entry_fields != {"none": True}: entries.append({"repository": REPO, **entry_fields}) @@ -173,11 +173,11 @@ def make_release(self, *parts: str, files: dict | None = None, root: str | None handle.write(text) return path - def run_quietly(self, function, *args): + def run_quietly(self, function: Any, *args: Any) -> Any: with contextlib.redirect_stdout(io.StringIO()): return function(*args) - def moved(self, function, *args) -> str: + def moved(self, function: Any, *args: Any) -> str: """Run a move that is expected to succeed and return where the folder went (None would fail the test here).""" target = self.run_quietly(function, *args) assert target is not None, "the folder was not moved" diff --git a/tests/test_form_helpers_and_logger.py b/tests/test_form_helpers_and_logger.py index a79705b..a0bc624 100644 --- a/tests/test_form_helpers_and_logger.py +++ b/tests/test_form_helpers_and_logger.py @@ -4,6 +4,7 @@ Everything runs against temporary files and a temporary state.db. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import contextlib import io import json @@ -212,11 +213,11 @@ def setUp(self) -> None: patcher.start() self.addCleanup(patcher.stop) - def write(self, path: str, payload) -> None: + def write(self, path: str, payload: Any) -> None: with open(path, "w", encoding="utf-8") as handle: handle.write(payload if isinstance(payload, str) else json.dumps(payload)) - def run_doctor(self, platform="win32", env=None): + def run_doctor(self, platform: str = "win32", env: Any = None) -> Any: environment = {k: v for k, v in (env if env is not None else self.env).items()} with mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(doctor_checks.sys, "platform", platform), \ contextlib.redirect_stdout(io.StringIO()): diff --git a/tests/test_gui_daemon.py b/tests/test_gui_daemon.py index ebd1b1f..8f14644 100644 --- a/tests/test_gui_daemon.py +++ b/tests/test_gui_daemon.py @@ -16,7 +16,7 @@ NOW = 1_000_000.0 -def running(**overrides) -> DaemonSnapshot: +def running(**overrides: Any) -> DaemonSnapshot: fields: dict[str, Any] = dict(running=True, pid=42, paused=False, next_poll_at=NOW + 125, current_repo=None, log_on=False) fields.update(overrides) return DaemonSnapshot(**fields) @@ -132,7 +132,7 @@ def test_stop_pending_ends_when_the_daemon_is_gone(self) -> None: class ConfigFingerprintTests(unittest.TestCase): BASE = {"polling": {"enabled": True, "interval_seconds": 300}, "processing": {"recheck_intervals_minutes": [5, 15]}} - def fingerprint(self, config) -> str: + def fingerprint(self, config: Any) -> str: return config_manager.get_config_fingerprint(config) def test_same_settings_give_the_same_fingerprint(self) -> None: @@ -175,7 +175,7 @@ def setUp(self) -> None: patcher.start() self.addCleanup(patcher.stop) - def patches(self, status, control=None, config_log=True): + def patches(self, status: Any, control: Any = None, config_log: bool = True) -> Any: control_patch = ( mock.patch.object(gui_daemon.daemon_control, "get_control_state", return_value=control) if not isinstance(control, Exception) @@ -189,7 +189,7 @@ def patches(self, status, control=None, config_log=True): ), ) - def snapshot(self, status, control=None, config_log=True) -> DaemonSnapshot: + def snapshot(self, status: Any, control: Any = None, config_log: bool = True) -> DaemonSnapshot: first, second, third = self.patches(status, control, config_log) with first, second, third: return gui_daemon.read_snapshot() @@ -283,7 +283,7 @@ def test_pythonw_is_swapped_for_a_console_python(self) -> None: self.assertEqual(os.path.basename(swapped), "python.exe") self.assertEqual(daemon_launcher.get_python_executable("/usr/bin/python3"), "/usr/bin/python3") - def start_with_fake_popen(self, stderr_path): + def start_with_fake_popen(self, stderr_path: Any) -> Any: with mock.patch.object(daemon_launcher, "get_stderr_path", return_value=stderr_path): with mock.patch.object(daemon_launcher.subprocess, "Popen") as popen: popen.return_value.pid = 4321 diff --git a/tests/test_gui_doctor.py b/tests/test_gui_doctor.py index 19a46b5..871125f 100644 --- a/tests/test_gui_doctor.py +++ b/tests/test_gui_doctor.py @@ -3,6 +3,7 @@ No window is opened and nothing real is read: paths and the environment are passed in. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import os import tempfile import unittest @@ -25,7 +26,7 @@ def touch(self, *paths: str) -> None: with open(path, "w", encoding="utf-8") as handle: handle.write("{}") - def reasons(self, environment=FULL_ENV) -> list[str]: + def reasons(self, environment: Any = FULL_ENV) -> list[str]: return gui_doctor.first_run_reasons(environment, self.config, self.mapping) def test_a_complete_setup_is_not_a_first_run(self) -> None: @@ -60,7 +61,7 @@ def test_the_real_environment_is_used_when_none_is_given(self) -> None: class ReportTextTests(unittest.TestCase): - def report(self, errors=(), warnings=()): + def report(self, errors: Any = (), warnings: Any = ()) -> Any: return {"ok": not errors, "platform": "win32", "errors": list(errors), "warnings": list(warnings), "checks": ["c"]} def test_summary_lines(self) -> None: @@ -79,7 +80,7 @@ def test_the_button_is_highlighted_for_a_first_run_or_for_errors_only(self) -> N class CenterTests(unittest.TestCase): SCREEN = (0, 0, 1920, 1080) - def center(self, parent, size, screen=SCREEN): + def center(self, parent: Any, size: Any, screen: Any = SCREEN) -> Any: return gui_state.center_over(*parent, *size, *screen) def test_a_dialog_is_centred_over_its_parent(self) -> None: diff --git a/tests/test_gui_logic.py b/tests/test_gui_logic.py index bf3605b..6a4ae5e 100644 --- a/tests/test_gui_logic.py +++ b/tests/test_gui_logic.py @@ -2,6 +2,7 @@ Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import json import os import tempfile @@ -21,11 +22,11 @@ def setUp(self) -> None: self.addCleanup(patcher.stop) self.addCleanup(dry_run_mode.set_dry_run, False) - def write(self, payload) -> None: + def write(self, payload: Any) -> None: with open(self.config_path, "w", encoding="utf-8") as handle: json.dump(payload, handle) - def read(self): + def read(self) -> Any: with open(self.config_path, encoding="utf-8") as handle: return json.load(handle) @@ -95,7 +96,7 @@ def test_dry_run_writes_nothing(self) -> None: class RepoFormTests(unittest.TestCase): - def valid_form(self, **overrides): + def valid_form(self, **overrides: Any) -> Any: form = { "destination": tempfile.gettempdir(), "folder": " App ", @@ -176,7 +177,7 @@ def test_new_repo_needs_owner_slash_repo_and_destination(self) -> None: class SettingsFormTests(unittest.TestCase): - def valid_form(self, **overrides): + def valid_form(self, **overrides: Any) -> Any: form = { "recheck": "5, 15, 30", "max_emails": "0", @@ -217,7 +218,7 @@ def test_invalid_values_are_reported(self) -> None: class OverviewTests(unittest.TestCase): NOW = 1_800_000_000.0 - def summary(self, **overrides): + def summary(self, **overrides: Any) -> Any: base = { "last_activity": self.NOW - 100, "latest_tag": "v1", @@ -302,7 +303,7 @@ def setUp(self) -> None: self.connection = db_manager.open_database() self.addCleanup(self.connection.close) - def add_job(self, repo, tag, status, next_check, attempts=0, updated=100.0, downloaded=0, skipped=0, total=0): + def add_job(self, repo: Any, tag: Any, status: Any, next_check: Any, attempts: int = 0, updated: float = 100.0, downloaded: int = 0, skipped: int = 0, total: int = 0) -> None: self.connection.execute( "INSERT INTO job_queue (repo, tag, status, attempt_count, next_check_time, downloaded_count," " skipped_count, total_items, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", diff --git a/tests/test_gui_state.py b/tests/test_gui_state.py index 238eea4..80c2bdb 100644 --- a/tests/test_gui_state.py +++ b/tests/test_gui_state.py @@ -2,6 +2,7 @@ Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import json import os import tempfile @@ -76,7 +77,7 @@ def test_keeps_maximized_flag(self) -> None: class PlacePopupTests(unittest.TestCase): SCREEN = (0, 0, 1920, 1080) - def place(self, x, y, width=400, height=60): + def place(self, x: Any, y: Any, width: int = 400, height: int = 60) -> Any: return gui_state.place_popup(x, y, width, height, *self.SCREEN) def test_normal_case_is_below_right_of_the_pointer(self) -> None: diff --git a/tests/test_gui_theme.py b/tests/test_gui_theme.py index d3d81ff..a740faa 100644 --- a/tests/test_gui_theme.py +++ b/tests/test_gui_theme.py @@ -4,19 +4,19 @@ class GuiThemeTests(unittest.TestCase): - def test_palettes_have_the_same_keys(self): + def test_palettes_have_the_same_keys(self) -> None: self.assertEqual(set(gui_theme.LIGHT), set(gui_theme.DARK)) - def test_color_swaps_map_the_other_palette_to_this_one(self): + def test_color_swaps_map_the_other_palette_to_this_one(self) -> None: swaps = gui_theme.color_swaps(True) self.assertEqual(swaps[gui_theme.LIGHT["stripe"].lower()], gui_theme.DARK["stripe"]) self.assertEqual(gui_theme.color_swaps(False)[gui_theme.DARK["muted"].lower()], gui_theme.LIGHT["muted"]) - def test_button_names_the_mode_it_switches_to(self): + def test_button_names_the_mode_it_switches_to(self) -> None: self.assertIn("Light", gui_theme.button_text(True)) self.assertIn("Dark", gui_theme.button_text(False)) - def test_dark_mode_setting_defaults_to_off_and_ignores_non_booleans(self): + def test_dark_mode_setting_defaults_to_off_and_ignores_non_booleans(self) -> None: self.assertFalse(config_manager.get_gui_dark_mode({})) self.assertTrue(config_manager.get_gui_dark_mode({"gui": {"dark_mode": True}})) self.assertFalse(config_manager.get_gui_dark_mode({"gui": {"dark_mode": "yes"}})) diff --git a/tests/test_gui_tray.py b/tests/test_gui_tray.py index a366e06..ae83563 100644 --- a/tests/test_gui_tray.py +++ b/tests/test_gui_tray.py @@ -8,7 +8,7 @@ import tempfile import unittest from types import SimpleNamespace -from typing import Optional +from typing import Any, Optional from unittest import mock from modules import config_manager, db_manager, gui_data, gui_tray @@ -137,27 +137,27 @@ def __init__(self, fail: bool = False) -> None: self.writes = 0 self.fail = fail - def CreateKeyEx(self, root, path, _reserved, _access): + def CreateKeyEx(self, root: Any, path: Any, _reserved: Any, _access: Any) -> Any: if self.fail: raise OSError("denied") values = self.keys.setdefault(path, {}) registry = self class Handle: - def __enter__(self): + def __enter__(self) -> Any: return (path, values) - def __exit__(self, *_exc): + def __exit__(self, *_exc: Any) -> bool: return False return Handle() - def QueryValueEx(self, key, name): + def QueryValueEx(self, key: Any, name: Any) -> Any: if name not in key[1]: raise OSError("missing") return key[1][name], self.REG_SZ - def SetValueEx(self, key, name, _reserved, _kind, value): + def SetValueEx(self, key: Any, name: Any, _reserved: Any, _kind: Any, value: Any) -> None: key[1][name] = value self.writes += 1 @@ -242,10 +242,10 @@ def test_only_failed_jobs_after_the_cursor_are_returned(self) -> None: def test_the_feed_reports_a_failure_once(self) -> None: class Store: - def load(self): + def load(self) -> Any: return {} - def save(self, seen): + def save(self, seen: Any) -> None: pass feed = gui_data.StatusFeed(store=Store()) @@ -275,7 +275,7 @@ def test_success_and_failure_are_reported_without_raising(self) -> None: notice = gui_tray.Notice("t", "m") seen = [] - def runner(command): + def runner(command: Any) -> Any: seen.append(command) return SimpleNamespace(returncode=0) @@ -283,7 +283,7 @@ def runner(command): self.assertIn("-EncodedCommand", seen[0]) self.assertFalse(gui_tray.send_windows_toast("GHAADD.GUI", notice, lambda command: SimpleNamespace(returncode=1))) - def broken(command): + def broken(command: Any) -> None: raise OSError("no powershell") self.assertFalse(gui_tray.send_windows_toast("GHAADD.GUI", notice, broken)) diff --git a/tests/test_idle_polling.py b/tests/test_idle_polling.py index 8030b62..8f29889 100644 --- a/tests/test_idle_polling.py +++ b/tests/test_idle_polling.py @@ -3,6 +3,7 @@ Nothing real is touched: the database, the watcher and the cycle are replaced. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any, Iterator import contextlib import unittest from unittest import mock @@ -14,7 +15,7 @@ class FakeWatcher: """Answers each wait() from a list and records how it was called.""" - def __init__(self, results, **_kwargs) -> None: + def __init__(self, results: Any, **_kwargs: Any) -> None: self.results = list(results) self.calls: list[dict] = [] self.stop_requested = False @@ -28,7 +29,7 @@ def begin_cycle(self) -> None: def checkpoint(self) -> bool: return False - def wait(self, seconds, on_change=None, idle=False): + def wait(self, seconds: Any, on_change: Any = None, idle: bool = False) -> Any: self.calls.append({"seconds": seconds, "idle": idle}) result = self.results.pop(0) if isinstance(result, BaseException): # a result can also be something that goes wrong in the wait @@ -41,13 +42,13 @@ class FakeViewerPusher: instances: list["FakeViewerPusher"] = [] - def __init__(self, settings, **_kwargs) -> None: + def __init__(self, settings: Any, **_kwargs: Any) -> None: self.settings = settings self.switches: list[tuple] = [] self.stopped = 0 FakeViewerPusher.instances.append(self) - def apply_override(self, override, configured) -> bool: + def apply_override(self, override: Any, configured: Any) -> bool: self.switches.append((override, configured)) return bool(configured if override is None else override) @@ -56,17 +57,17 @@ def stop(self) -> None: class IdleLoopTests(unittest.TestCase): - def run_loop(self, results, idle, viewer_enabled=False, dry_run=False): + def run_loop(self, results: Any, idle: Any, viewer_enabled: bool = False, dry_run: bool = False) -> Any: watcher = FakeWatcher(results) cycles = [] FakeViewerPusher.instances = [] self.watcher_kwargs: dict = {} @contextlib.contextmanager - def fake_database(): + def fake_database() -> Iterator[Any]: yield object() - def fake_watcher(**kwargs): + def fake_watcher(**kwargs: Any) -> Any: self.watcher_kwargs = kwargs return watcher @@ -179,7 +180,7 @@ def test_the_gui_action_writes_the_request(self) -> None: class GuiViewTests(unittest.TestCase): - def view(self, **fields): + def view(self, **fields: Any) -> Any: return gui_daemon.build_view(gui_daemon.DaemonSnapshot(running=True, pid=7, **fields), now=1000.0) def test_an_idle_daemon_says_so_and_can_be_polled(self) -> None: diff --git a/tests/test_job_queue.py b/tests/test_job_queue.py index 5a0e2cd..edae239 100644 --- a/tests/test_job_queue.py +++ b/tests/test_job_queue.py @@ -5,6 +5,7 @@ against a throw-away database; the real state.db is never opened. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import os import tempfile import unittest @@ -25,13 +26,13 @@ def setUp(self) -> None: self.connection = db_manager.open_database() self.addCleanup(self.connection.close) - def job(self, repo="o/app", tag="v1", release_type: str | None = "Release", due=0.0, commit=None) -> int: + def job(self, repo: str = "o/app", tag: str = "v1", release_type: str | None = "Release", due: float = 0.0, commit: Any = None) -> int: return db_manager.enqueue_job(self.connection, repo, tag, release_type, due, commit) - def row(self, job_id: int): + def row(self, job_id: int) -> Any: return db_manager.get_jobs_by_ids(self.connection, [job_id])[0] - def ids(self, rows) -> list[int]: + def ids(self, rows: Any) -> list[int]: return [r["id"] for r in rows] @@ -313,7 +314,7 @@ def test_the_oldest_n_are_removed(self) -> None: class LifecycleEventTests(QueueTestCase): - def add(self, event_type="WARNING", repo="o/app", message="m", age_days=0.0) -> int: + def add(self, event_type: str = "WARNING", repo: str = "o/app", message: str = "m", age_days: float = 0.0) -> int: event_id = db_manager.insert_lifecycle_event(self.connection, event_type, message, repo=repo) self.connection.execute( "UPDATE lifecycle_events SET created_at = CAST(strftime('%s', 'now') AS REAL) - ? WHERE id = ?", diff --git a/tests/test_mailbox_imap.py b/tests/test_mailbox_imap.py index 77d2a80..c76611f 100644 --- a/tests/test_mailbox_imap.py +++ b/tests/test_mailbox_imap.py @@ -4,6 +4,7 @@ select (read-only when only reading), search, fetch, flags and the move to Trash. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any, Self import contextlib import io import os @@ -30,39 +31,39 @@ def __init__(self) -> None: class FakeIMAPClient: server = FakeServer() - def __init__(self, host, use_uid=True): + def __init__(self, host: Any, use_uid: bool = True) -> None: type(self).server.calls.append(("connect", host, use_uid)) type(self).server.connections += 1 if type(self).server.connections <= type(self).server.fail_times: raise OSError("connection refused") - def __enter__(self): + def __enter__(self) -> Self: return self - def __exit__(self, *exc): + def __exit__(self, *exc: Any) -> bool: return False - def login(self, user, password): + def login(self, user: Any, password: Any) -> None: self.server.calls.append(("login", user, password)) - def select_folder(self, folder, readonly=False): + def select_folder(self, folder: Any, readonly: bool = False) -> None: self.server.calls.append(("select", folder, readonly)) - def search(self, criteria): + def search(self, criteria: Any) -> Any: self.server.calls.append(("search", criteria)) return list(self.server.subjects) - def fetch(self, ids, items): + def fetch(self, ids: Any, items: Any) -> Any: self.server.calls.append(("fetch", list(ids), items)) return { msg_id: {b"ENVELOPE": SimpleNamespace(subject=self.server.subjects[msg_id])} for msg_id in ids } - def set_flags(self, ids, flags): + def set_flags(self, ids: Any, flags: Any) -> None: self.server.calls.append(("flags", list(ids), flags)) - def move(self, ids, folder): + def move(self, ids: Any, folder: Any) -> None: self.server.calls.append(("move", list(ids), folder)) @@ -80,7 +81,7 @@ def setUp(self) -> None: patcher.start() self.addCleanup(patcher.stop) - def quietly(self, function, *args, **kwargs): + def quietly(self, function: Any, *args: Any, **kwargs: Any) -> Any: buffer = io.StringIO() try: with contextlib.redirect_stdout(buffer): diff --git a/tests/test_mapping_lookups.py b/tests/test_mapping_lookups.py index 9c48b34..f2be205 100644 --- a/tests/test_mapping_lookups.py +++ b/tests/test_mapping_lookups.py @@ -32,11 +32,11 @@ def setUp(self) -> None: self.addCleanup(patcher.stop) mapping_manager._BACKED_UP_INVALID_MAPPING_SIGNATURES.clear() - def write(self, *entries) -> None: + def write(self, *entries: Any) -> None: with open(self.path, "w", encoding="utf-8") as handle: json.dump({"repositories": list(entries)}, handle) - def quietly(self, function, *args): + def quietly(self, function: Any, *args: Any) -> Any: with contextlib.redirect_stdout(io.StringIO()): return function(*args) @@ -196,10 +196,10 @@ def test_each_missing_destination_is_warned_about_once(self) -> None: class ValidationTests(LookupTestCase): - def check(self, *entries): + def check(self, *entries: Any) -> Any: return mapping_manager.validate_mapping_payload({"repositories": list(entries)}) - def errors(self, *entries) -> str: + def errors(self, *entries: Any) -> str: return " | ".join(self.check(*entries)["errors"]) def test_a_good_entry_has_no_errors(self) -> None: @@ -257,7 +257,7 @@ def test_two_repositories_sharing_one_target_folder_are_flagged(self) -> None: self.assertIn("o/a, o/b", warning[0]) self.assertNotIn("o/c", warning[0]) - def shared(self, *marks): + def shared(self, *marks: Any) -> Any: """Entries sharing one folder; each mark is True / False / None (= key absent).""" shared = {"destination": os.path.join(self.root, "d"), "folder": "Same", "subfolder": "@GitHub"} entries = [] diff --git a/tests/test_mapping_manager.py b/tests/test_mapping_manager.py index c5b22ed..eccf0bb 100644 --- a/tests/test_mapping_manager.py +++ b/tests/test_mapping_manager.py @@ -2,6 +2,7 @@ Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import json import multiprocessing import os @@ -16,7 +17,7 @@ def _increment_worker(mapping_path: str, iterations: int) -> None: """Child process: bump a shared counter `iterations` times via update_mapping.""" mapping_manager._mapping_file_path = lambda: mapping_path - def _bump(payload): + def _bump(payload: Any) -> None: payload["repositories"][0]["counter"] = payload["repositories"][0].get("counter", 0) + 1 for _ in range(iterations): @@ -53,7 +54,7 @@ def read_mapping(self) -> list[dict]: class UpdateMappingTests(MappingWriteTestCase): def test_creates_file_sorted_with_collapsed_arrays(self) -> None: - def _add(payload): + def _add(payload: Any) -> None: payload["repositories"].append({"repository": "zed/zebra", "recheck_intervals": [5, 15]}) payload["repositories"].append({"repository": "amy/apple"}) @@ -85,7 +86,7 @@ def test_leaves_no_temp_files_behind(self) -> None: def test_mutator_error_keeps_old_file_and_releases_lock(self) -> None: self.write_mapping([{"repository": "a/a"}]) - def _fail(payload): + def _fail(payload: Any) -> None: payload["repositories"].append({"repository": "b/b"}) raise RuntimeError("boom") diff --git a/tests/test_mapping_upgrade.py b/tests/test_mapping_upgrade.py index af2e4d9..737f095 100644 --- a/tests/test_mapping_upgrade.py +++ b/tests/test_mapping_upgrade.py @@ -4,6 +4,7 @@ Nothing here touches the real mapping.json, state.db or daemon lock. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import json import os import tempfile @@ -144,7 +145,7 @@ def test_an_existing_backup_is_never_overwritten(self) -> None: self.assertEqual(handle.read(), "older backup") self.assertEqual(len(self.backups()), 2) - def daemon(self, running: bool, mapping_format) -> mock._patch: + def daemon(self, running: bool, mapping_format: Any) -> mock._patch: return mock.patch( "modules.daemon_lock.get_daemon_status", return_value={"running": running, "mapping_format": mapping_format} ) diff --git a/tests/test_new_entry_defaults.py b/tests/test_new_entry_defaults.py index 2b6820f..cc56272 100644 --- a/tests/test_new_entry_defaults.py +++ b/tests/test_new_entry_defaults.py @@ -4,6 +4,7 @@ Nothing here touches the real config.json, mapping.json or state.db. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import unittest from unittest import mock @@ -48,7 +49,7 @@ def test_a_new_entry_without_settings_keeps_the_old_defaults(self) -> None: class SettingsFormTests(unittest.TestCase): - def form(self, **overrides): + def form(self, **overrides: Any) -> Any: form = { "recheck": "5", "max_emails": "0", "dest_check": "10", "default_limit": "10", "default_subfolder": "@GitHub", "interval": "300", "jitter_min": "5", "jitter_max": "30", diff --git a/tests/test_pending_jobs_view.py b/tests/test_pending_jobs_view.py index 289af2c..df5d102 100644 --- a/tests/test_pending_jobs_view.py +++ b/tests/test_pending_jobs_view.py @@ -2,6 +2,7 @@ A temporary state.db is used. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import contextlib import os import tempfile @@ -39,7 +40,7 @@ def test_one_job_or_none_needs_no_note(self) -> None: class RowTests(unittest.TestCase): - def summary(self, **fields) -> dict: + def summary(self, **fields: Any) -> dict: base = { "last_activity": NOW, "latest_tag": "build-ccc", "latest_release_type": "Release", "latest_status": "PENDING", "latest_updated_at": NOW, "latest_downloaded": 3, "latest_skipped": 0, "latest_total": 3, @@ -48,7 +49,7 @@ def summary(self, **fields) -> dict: base.update(fields) return base - def rows(self, summary: dict): + def rows(self, summary: dict) -> Any: return repo_overview.build_rows([{"repository": "o/r"}], {"o/r": summary}, NOW, lambda entry: 5) def test_several_pending_jobs_show_in_the_recheck_cell_and_the_note(self) -> None: diff --git a/tests/test_polling_efficiency.py b/tests/test_polling_efficiency.py index 2201036..41fe504 100644 --- a/tests/test_polling_efficiency.py +++ b/tests/test_polling_efficiency.py @@ -2,6 +2,7 @@ Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import json import os import tempfile @@ -38,7 +39,7 @@ def compute(self) -> str: with open(self.path, encoding="utf-8") as handle: return handle.read() - def cache(self, **kwargs) -> StatCache: + def cache(self, **kwargs: Any) -> StatCache: return StatCache(lambda: [self.path], self.compute, clock=self.clock, **kwargs) def test_reuses_the_value_while_the_file_is_unchanged(self) -> None: @@ -261,7 +262,7 @@ def setUp(self) -> None: self.daemon_connection = db_manager.open_database() self.addCleanup(self.daemon_connection.close) - def table(self): + def table(self) -> Any: with mock.patch.object(gui_data.daemon_lock, "get_daemon_status", return_value={"current_job": None}): return gui_data.load_repo_table() diff --git a/tests/test_queue_counts.py b/tests/test_queue_counts.py index 445d504..70cb687 100644 --- a/tests/test_queue_counts.py +++ b/tests/test_queue_counts.py @@ -3,6 +3,7 @@ A temporary state.db is used and the re-check intervals are fixed. Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import contextlib import os import tempfile @@ -57,7 +58,7 @@ def setUp(self) -> None: patcher.start() self.addCleanup(patcher.stop) - def add_job(self, next_check: float, attempt: int = 0, age_minutes: float = 0.0, status: str = "PENDING", repo="o/r") -> None: + def add_job(self, next_check: float, attempt: int = 0, age_minutes: float = 0.0, status: str = "PENDING", repo: str = "o/r") -> None: with contextlib.closing(db_manager.open_database()) as connection: job_id = db_manager.enqueue_job(connection, repo, f"v{next_check}{attempt}{age_minutes}", "Release", next_check_time=next_check) connection.execute( diff --git a/tests/test_queue_reports.py b/tests/test_queue_reports.py index f5a9023..4bd0957 100644 --- a/tests/test_queue_reports.py +++ b/tests/test_queue_reports.py @@ -116,13 +116,13 @@ def seed(self) -> None: db_manager.enqueue_job(c, "o/app", "v4", "Release", 0, "eee") db_manager.enqueue_job(c, "o/app", "v5", "Release", time.time() + 99999, "fff") - def status(self, **kwargs) -> dict: + def status(self, **kwargs: Any) -> dict: buffer = io.StringIO() with contextlib.redirect_stdout(buffer): queue_reports.print_queue_status(as_json=True, **kwargs) return json.loads(buffer.getvalue()) - def text(self, **kwargs) -> str: + def text(self, **kwargs: Any) -> str: buffer = io.StringIO() with contextlib.redirect_stdout(buffer): queue_reports.print_queue_status(**kwargs) @@ -173,7 +173,7 @@ def seed_three_releases(self) -> tuple[int, int, int]: def jobs_by_id(self) -> dict: return {j["id"]: j for j in self.status()["recent_jobs"]} - def use_sanity_mode(self, mode) -> None: + def use_sanity_mode(self, mode: Any) -> None: entry = {"repository": "o/app"} if mode is not None: entry["sanity_check"] = mode diff --git a/tests/test_queue_worker_flow.py b/tests/test_queue_worker_flow.py index 1149abb..e7bbbc8 100644 --- a/tests/test_queue_worker_flow.py +++ b/tests/test_queue_worker_flow.py @@ -15,7 +15,7 @@ import time import unittest from contextlib import closing -from typing import Optional +from typing import Any, Optional from unittest import mock # queue_worker -> mailbox_listener refuses to import without credentials; they are never used to connect. @@ -27,7 +27,7 @@ REPO = "o/app" -def download_result(status="SUCCESS", downloaded=3, skipped=1, total=4, working_dir=None, skip_reason=None, skipped_items=None): +def download_result(status: str = "SUCCESS", downloaded: int = 3, skipped: int = 1, total: int = 4, working_dir: Any = None, skip_reason: Any = None, skipped_items: Any = None) -> Any: return { "status": status, "downloaded_count": downloaded, @@ -55,27 +55,27 @@ def setUp(self) -> None: "download", "complete", "partial", "warnings", "completed_logs", "partial_logs", "deleted", "trashed", "fetch")} self.write_mapping() - def commit_hash(repo, tag, token, include_reason=False): + def commit_hash(repo: Any, tag: Any, token: Any, include_reason: bool = False) -> Any: if self.commit is None: return (None, "http_404") if include_reason else None return (self.commit, None) if include_reason else self.commit - def download(repo, tag, release_type, include_stats=False): + def download(repo: Any, tag: Any, release_type: Any, include_stats: bool = False) -> Any: self.calls["download"].append((repo, tag, release_type)) result = self.download_results.pop(0) if self.download_results else download_result() if isinstance(result, Exception): raise result return result - def move_complete(working_dir, repo=None): + def move_complete(working_dir: Any, repo: Any = None) -> Any: self.calls["complete"].append(working_dir) return os.path.join(self.root, "done", os.path.basename(working_dir)) - def move_partial(working_dir): + def move_partial(working_dir: Any) -> Any: self.calls["partial"].append(working_dir) return os.path.join(self.root, "partial", os.path.basename(working_dir)) - def fetch(limit=None): + def fetch(limit: Any = None) -> Any: self.calls["fetch"].append(limit) return list(self.notifications) @@ -106,24 +106,24 @@ def fetch(limit=None): self.addCleanup(self.connection.close) # ----- helpers - def write_mapping(self, *entries) -> None: + def write_mapping(self, *entries: Any) -> None: with open(self.mapping_path, "w", encoding="utf-8") as handle: json.dump({"repositories": list(entries)}, handle) - def quietly(self, function, *args, **kwargs): + def quietly(self, function: Any, *args: Any, **kwargs: Any) -> Any: with contextlib.redirect_stdout(io.StringIO()): return function(*args, **kwargs) - def job(self, tag="v1", release_type="Release", commit="aaa1111", due=0.0, repo=REPO) -> int: + def job(self, tag: str = "v1", release_type: str = "Release", commit: str = "aaa1111", due: float = 0.0, repo: Any = REPO) -> int: return db_manager.enqueue_job(self.connection, repo, tag, release_type, due, commit) - def row(self, job_id: int): + def row(self, job_id: int) -> Any: return db_manager.get_jobs_by_ids(self.connection, [job_id])[0] - def at_attempt(self, job_id: int, attempt: int, downloaded=0, skipped=0, total=0, working_dir=None, due=0.0) -> None: + def at_attempt(self, job_id: int, attempt: int, downloaded: int = 0, skipped: int = 0, total: int = 0, working_dir: Any = None, due: float = 0.0) -> None: db_manager.reschedule_job(self.connection, job_id, due, attempt, "aaa1111", downloaded, skipped, total, working_dir) - def staged_dir(self, name="rel1") -> str: + def staged_dir(self, name: str = "rel1") -> str: path = os.path.join(self.root, "Processing", "app (o)", "Release", name) os.makedirs(path, exist_ok=True) return path @@ -134,14 +134,14 @@ def warning_kinds(self) -> list[str]: def pending_ids(self) -> list[int]: return [r["id"] for r in db_manager.get_pending_jobs(self.connection)] - def process(self, **kwargs): + def process(self, **kwargs: Any) -> Any: return self.quietly(queue_worker.process_queue_once, self.connection, "token", **kwargs) - def ingest(self, **kwargs): + def ingest(self, **kwargs: Any) -> Any: return self.quietly(queue_worker.ingest_notifications_once, self.connection, "token", **kwargs) -def notification(repo: Optional[str] = REPO, tag: Optional[str] = "v1", release_type="Release", email_id=1) -> dict: +def notification(repo: Optional[str] = REPO, tag: Optional[str] = "v1", release_type: str = "Release", email_id: int = 1) -> dict: return {"repo": repo, "tag": tag, "release_type": release_type, "email_id": email_id} @@ -257,7 +257,7 @@ def test_one_failing_notification_does_not_stop_the_others(self) -> None: self.notifications = [notification(tag="bad", email_id=1), notification(tag="good", email_id=2)] real = queue_worker.enqueue_job - def flaky(connection, repo, tag, **kwargs): + def flaky(connection: Any, repo: Any, tag: Any, **kwargs: Any) -> Any: if tag == "bad": raise RuntimeError("database is locked") return real(connection, repo, tag, **kwargs) @@ -537,7 +537,7 @@ def test_a_pause_during_the_ingest_skips_the_queue_part(self) -> None: class CounterProtectionTests(unittest.TestCase): """A weaker attempt must never overwrite what an earlier one already achieved.""" - def row(self, downloaded=3, skipped=1, total=4, working_dir: Optional[str] = "/work/a"): + def row(self, downloaded: int = 3, skipped: int = 1, total: int = 4, working_dir: Optional[str] = "/work/a") -> Any: return {"downloaded_count": downloaded, "skipped_count": skipped, "total_items": total, "working_dir": working_dir} def test_a_first_attempt_has_nothing_to_protect(self) -> None: diff --git a/tests/test_stats.py b/tests/test_stats.py index 4264a26..f8e4960 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -22,7 +22,7 @@ GB = 1024 ** 3 -def jobs_fixture(): +def jobs_fixture() -> Any: """(repo, tag, status, created_at): oldest first, like db_manager.get_job_history.""" return [ ("o/old", "v1", "COMPLETED", NOW - 400 * DAY), @@ -90,13 +90,13 @@ def test_a_folder_count_at_the_limit_is_not_over_it(self) -> None: class BuildTests(unittest.TestCase): - def build(self, **overrides): + def build(self, **overrides: Any) -> Any: arguments: dict[str, Any] = dict(entries=ENTRIES, jobs=jobs_fixture(), cycles=[NOW - 5 * HOUR, NOW - 2 * DAY, NOW - 100 * DAY], sizes=SIZES, folder_counts={}, now=NOW) arguments.update(overrides) return stats.build(**arguments) - def period(self, report, key): + def period(self, report: Any, key: Any) -> Any: return next(row for row in report["periods"] if row["key"] == key) def test_jobs_per_period(self) -> None: @@ -268,7 +268,7 @@ def test_presets_count_today_as_the_first_day(self) -> None: class BuildRangeTests(unittest.TestCase): - def build(self, window, **overrides): + def build(self, window: Any, **overrides: Any) -> Any: arguments: dict[str, Any] = dict(entries=ENTRIES, jobs=jobs_fixture(), cycles=[], sizes=SIZES, folder_counts={}, now=NOW, top=None, date_range=window) arguments.update(overrides) @@ -552,21 +552,21 @@ def test_json_is_accepted_with_stats_and_rejected_without_a_command(self) -> Non class BucketDailyTests(unittest.TestCase): - def series(self, count): + def series(self, count: Any) -> Any: start = date(2025, 1, 1) return [{"date": str(start + timedelta(days=i)), "jobs": 1} for i in range(count)] - def test_short_series_stays_daily(self): + def test_short_series_stays_daily(self) -> None: unit, buckets = stats.bucket_daily(self.series(30), 100) self.assertEqual((unit, len(buckets)), ("day", 30)) - def test_long_series_becomes_weeks_that_keep_the_total(self): + def test_long_series_becomes_weeks_that_keep_the_total(self) -> None: unit, buckets = stats.bucket_daily(self.series(400), 100) self.assertEqual(unit, "week") self.assertEqual(sum(b["jobs"] for b in buckets), 400) self.assertEqual(buckets[0]["days"], 5) # 2025-01-01 is a Wednesday: the first week is cut off - def test_very_long_series_becomes_months(self): + def test_very_long_series_becomes_months(self) -> None: unit, buckets = stats.bucket_daily(self.series(1500), 100) self.assertEqual(unit, "month") self.assertEqual(sum(b["days"] for b in buckets), 1500) diff --git a/tests/test_status_tabs.py b/tests/test_status_tabs.py index 48fafa4..5903492 100644 --- a/tests/test_status_tabs.py +++ b/tests/test_status_tabs.py @@ -2,6 +2,7 @@ Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any import json import os import tempfile @@ -15,14 +16,14 @@ class MemoryStore: - def __init__(self, initial=None) -> None: + def __init__(self, initial: Any = None) -> None: self.data = dict(initial or {}) self.saves = 0 - def load(self): + def load(self) -> Any: return dict(self.data) - def save(self, seen) -> None: + def save(self, seen: Any) -> None: self.data = dict(seen) self.saves += 1 @@ -34,14 +35,14 @@ def __init__(self) -> None: self.events: list[dict] = [] self.fetches = 0 - def add(self, event_type, category=None, repo=None, tag=None, message="m", created_at=1_000_000.0, - destination_path=None) -> int: + def add(self, event_type: Any, category: Any = None, repo: Any = None, tag: Any = None, message: str = "m", created_at: float = 1_000_000.0, + destination_path: Any = None) -> int: event_id = len(self.events) + 1 self.events.append(dict(id=event_id, event_type=event_type, category=category, repo=repo, tag=tag, message=message, created_at=created_at, destination_path=destination_path)) return event_id - def fetch(self, tab: TabDef, after_id, limit): + def fetch(self, tab: TabDef, after_id: Any, limit: Any) -> Any: self.fetches += 1 matching = [ e for e in self.events @@ -61,7 +62,7 @@ def setUp(self) -> None: self.events = FakeEvents() self.store = MemoryStore() - def model(self, **kwargs) -> StatusTabsModel: + def model(self, **kwargs: Any) -> StatusTabsModel: return StatusTabsModel(status_tabs.TAB_DEFS, self.events.fetch, self.events.max_id, self.store, **kwargs) def test_titles_show_unread_counts_only_when_there_are_some(self) -> None: @@ -334,7 +335,7 @@ def setUp(self) -> None: self.connection = db_manager.open_database() self.addCleanup(self.connection.close) - def add(self, event_type, category=None, message="m") -> None: + def add(self, event_type: Any, category: Any = None, message: str = "m") -> None: db_manager.insert_lifecycle_event(self.connection, event_type, message, category=category) def test_filters_orders_and_limits(self) -> None: @@ -377,7 +378,7 @@ def test_purge_removes_exactly_what_the_tab_lists(self) -> None: self.assertEqual({(e["event_type"], e["category"]) for e in left}, {("WARNING", "LIMIT"), ("COMPLETED_MOVE", None)}) def test_purge_for_one_repository_matches_its_own_warnings_only(self) -> None: - def limit(repo): + def limit(repo: Any) -> None: self.add("WARNING", "LIMIT", f"Folder limit warning: {repo} currently has 30 folder(s) in 'x' (limit=25).") limit("o/my_app") limit("O/MY_APP") @@ -421,7 +422,7 @@ def setUp(self) -> None: self.daemon = db_manager.open_database() # stays open, like the daemon's connection self.addCleanup(self.daemon.close) - def event(self, event_type, category=None, message="m", repo=None) -> None: + def event(self, event_type: Any, category: Any = None, message: str = "m", repo: Any = None) -> None: db_manager.insert_lifecycle_event(self.daemon, event_type, message, category=category, repo=repo) def test_events_flow_from_state_db_to_tab_titles(self) -> None: diff --git a/tests/test_terminal_log.py b/tests/test_terminal_log.py index dea1492..3ac2531 100644 --- a/tests/test_terminal_log.py +++ b/tests/test_terminal_log.py @@ -2,6 +2,7 @@ Run from the project root: python -m unittest discover -s tests -t . """ +from typing import Any, cast import contextlib import io import os @@ -10,6 +11,7 @@ from unittest import mock import main +from modules.config_manager import TerminalLogSettings from modules.log_files import list_log_files @@ -25,7 +27,7 @@ def setUp(self) -> None: redirect.__enter__() self.addCleanup(redirect.__exit__, None, None, None) - def make(self, enabled=False, **overrides): + def make(self, enabled: bool = False, **overrides: Any) -> Any: settings = { "enabled": enabled, "directory": self.directory, @@ -33,14 +35,14 @@ def make(self, enabled=False, **overrides): "keep_files": 30, **overrides, } - log = main.TerminalLog(settings, self.tees) + log = main.TerminalLog(cast(TerminalLogSettings, settings), self.tees) self.addCleanup(log.stop) return log - def logs(self): + def logs(self) -> Any: return list_log_files(self.directory) - def read(self, path): + def read(self, path: Any) -> Any: with open(path, encoding="utf-8") as handle: return handle.read()