diff --git a/automation_file/__init__.py b/automation_file/__init__.py index 7dc4507..83e2860 100644 --- a/automation_file/__init__.py +++ b/automation_file/__init__.py @@ -236,9 +236,7 @@ from automation_file.utils.rotate import RotateException, rotate_backups if TYPE_CHECKING: - from automation_file.ui.launcher import ( - launch_ui as launch_ui, # pylint: disable=useless-import-alias - ) + from automation_file.ui.launcher import launch_ui # Shared callback executor + package loader wired to the shared registry. callback_executor: CallbackExecutor = CallbackExecutor(executor.registry) diff --git a/automation_file/__main__.py b/automation_file/__main__.py index 12c00d0..51d6c8f 100644 --- a/automation_file/__main__.py +++ b/automation_file/__main__.py @@ -133,15 +133,7 @@ def _sleep_forever() -> None: time.sleep(3600) -def _build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(prog="automation_file") - parser.add_argument("-e", "--execute_file", help="path to an action JSON file") - parser.add_argument("-d", "--execute_dir", help="directory containing action JSON files") - parser.add_argument("-c", "--create_project", help="scaffold a project at this path") - parser.add_argument("--execute_str", help="JSON action list as a string") - - subparsers = parser.add_subparsers(dest="command") - +def _add_zip_commands(subparsers: argparse._SubParsersAction) -> None: zip_parser = subparsers.add_parser("zip", help="zip a file or directory") zip_parser.add_argument("source") zip_parser.add_argument("target") @@ -159,6 +151,8 @@ def _build_parser() -> argparse.ArgumentParser: unzip_parser.add_argument("--password", default=None) unzip_parser.set_defaults(handler=_cmd_unzip) + +def _add_file_commands(subparsers: argparse._SubParsersAction) -> None: download_parser = subparsers.add_parser("download", help="SSRF-validated HTTP download") download_parser.add_argument("url") download_parser.add_argument("output") @@ -169,6 +163,8 @@ def _build_parser() -> argparse.ArgumentParser: touch_parser.add_argument("--content", default="") touch_parser.set_defaults(handler=_cmd_create_file) + +def _add_server_commands(subparsers: argparse._SubParsersAction) -> None: server_parser = subparsers.add_parser("server", help="run the TCP action server") server_parser.add_argument("--host", default="localhost") server_parser.add_argument("--port", type=int, default=9943) @@ -183,6 +179,8 @@ def _build_parser() -> argparse.ArgumentParser: http_parser.add_argument("--shared-secret", default=None) http_parser.set_defaults(handler=_cmd_http_server) + +def _add_integration_commands(subparsers: argparse._SubParsersAction) -> None: ui_parser = subparsers.add_parser("ui", help="launch the PySide6 GUI") ui_parser.set_defaults(handler=_cmd_ui) @@ -206,6 +204,19 @@ def _build_parser() -> argparse.ArgumentParser: drive_parser.add_argument("--name", default=None) drive_parser.set_defaults(handler=_cmd_drive_upload) + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="automation_file") + parser.add_argument("-e", "--execute_file", help="path to an action JSON file") + parser.add_argument("-d", "--execute_dir", help="directory containing action JSON files") + parser.add_argument("-c", "--create_project", help="scaffold a project at this path") + parser.add_argument("--execute_str", help="JSON action list as a string") + + subparsers = parser.add_subparsers(dest="command") + _add_zip_commands(subparsers) + _add_file_commands(subparsers) + _add_server_commands(subparsers) + _add_integration_commands(subparsers) return parser diff --git a/automation_file/core/config.py b/automation_file/core/config.py index 6b2b1f6..c482a71 100644 --- a/automation_file/core/config.py +++ b/automation_file/core/config.py @@ -53,7 +53,6 @@ from automation_file.notify.manager import NotificationManager from automation_file.notify.sinks import ( EmailSink, - NotificationException, NotificationSink, SlackSink, WebhookSink, @@ -156,8 +155,6 @@ def _build_sink(entry: dict[str, Any]) -> NotificationSink: ) try: return builder(entry) - except NotificationException: - raise except (TypeError, ValueError) as err: raise ConfigException( f"invalid config for sink {entry.get('name') or sink_type!r}: {err}" diff --git a/automation_file/core/dag_executor.py b/automation_file/core/dag_executor.py index 7bea7d3..2f84a55 100644 --- a/automation_file/core/dag_executor.py +++ b/automation_file/core/dag_executor.py @@ -31,6 +31,75 @@ __all__ = ["execute_action_dag"] +class _DagRun: + """Mutable scheduling state shared by the submit / completion helpers.""" + + def __init__( + self, + nodes: list[Mapping[str, Any]], + pool: ThreadPoolExecutor, + fail_fast: bool, + ) -> None: + self.graph, self.indegree = _build_graph(nodes) + self.node_map = {_require_id(node): node for node in nodes} + self.results: dict[str, Any] = {} + self.lock = threading.Lock() + self.ready: deque[str] = deque( + node_id for node_id, count in self.indegree.items() if count == 0 + ) + self.in_flight: dict[Future[Any], str] = {} + self.pool = pool + self.fail_fast = fail_fast + + def _mark_skipped(self, dependent: str, reason_id: str) -> None: + with self.lock: + if dependent in self.results: + return + self.results[dependent] = f"skipped: dep {reason_id!r} failed" + for grandchild in self.graph.get(dependent, ()): + self.indegree[grandchild] -= 1 + self._mark_skipped(grandchild, dependent) + + def _skip_dependents(self, node_id: str) -> None: + for dependent in self.graph.get(node_id, ()): + self.indegree[dependent] -= 1 + self._mark_skipped(dependent, node_id) + + def submit(self, node_id: str) -> None: + action = self.node_map[node_id].get("action") + if not isinstance(action, list): + err = DagException(f"node {node_id!r} missing action list") + with self.lock: + self.results[node_id] = repr(err) + if self.fail_fast: + self._skip_dependents(node_id) + return + future = self.pool.submit(_run_action, action) + self.in_flight[future] = node_id + + def _complete(self, node_id: str, value: Any, failed: bool) -> None: + with self.lock: + self.results[node_id] = value + for dependent in self.graph.get(node_id, ()): + self.indegree[dependent] -= 1 + if failed and self.fail_fast: + self._mark_skipped(dependent, node_id) + elif self.indegree[dependent] == 0 and dependent not in self.results: + self.ready.append(dependent) + + def drain_completed(self) -> None: + done, _ = wait(list(self.in_flight), return_when=FIRST_COMPLETED) + for future in done: + node_id = self.in_flight.pop(future) + try: + value: Any = future.result() + failed = False + except Exception as err: # pylint: disable=broad-except + value = repr(err) + failed = True + self._complete(node_id, value, failed) + + def execute_action_dag( nodes: list[Mapping[str, Any]], max_workers: int = 4, @@ -46,54 +115,15 @@ def execute_action_dag( Raises :class:`DagException` for static errors detected before any action runs: duplicate ids, unknown dependencies, or cycles. """ - graph, indegree = _build_graph(nodes) - node_map = {_require_id(node): node for node in nodes} - results: dict[str, Any] = {} - lock = threading.Lock() - - ready: deque[str] = deque(node_id for node_id, count in indegree.items() if count == 0) - with ThreadPoolExecutor(max_workers=max_workers) as pool: - in_flight: dict[Future[Any], str] = {} - - def submit(node_id: str) -> None: - action = node_map[node_id].get("action") - if not isinstance(action, list): - err = DagException(f"node {node_id!r} missing action list") - with lock: - results[node_id] = repr(err) - if fail_fast: - for dependent in graph.get(node_id, ()): - indegree[dependent] -= 1 - _mark_skipped(dependent, node_id, graph, indegree, results, lock) - return - future = pool.submit(_run_action, action) - in_flight[future] = node_id - - while ready or in_flight: - while ready: - submit(ready.popleft()) - if not in_flight: + state = _DagRun(nodes, pool, fail_fast) + while state.ready or state.in_flight: + while state.ready: + state.submit(state.ready.popleft()) + if not state.in_flight: break - done, _ = wait(list(in_flight), return_when=FIRST_COMPLETED) - for future in done: - node_id = in_flight.pop(future) - failed = False - try: - value: Any = future.result() - except Exception as err: # pylint: disable=broad-except - value = repr(err) - failed = True - with lock: - results[node_id] = value - for dependent in graph.get(node_id, ()): - indegree[dependent] -= 1 - if failed and fail_fast: - _mark_skipped(dependent, node_id, graph, indegree, results, lock) - elif indegree[dependent] == 0 and dependent not in results: - ready.append(dependent) - - return results + state.drain_completed() + return state.results def _run_action(action: list) -> Any: @@ -156,20 +186,3 @@ def _detect_cycle( queue.append(dependent) if visited != len(ids): raise DagException("cycle detected in DAG") - - -def _mark_skipped( - dependent: str, - reason_id: str, - graph: dict[str, list[str]], - indegree: dict[str, int], - results: dict[str, Any], - lock: threading.Lock, -) -> None: - with lock: - if dependent in results: - return - results[dependent] = f"skipped: dep {reason_id!r} failed" - for grandchild in graph.get(dependent, ()): - indegree[grandchild] -= 1 - _mark_skipped(grandchild, dependent, graph, indegree, results, lock) diff --git a/automation_file/core/fim.py b/automation_file/core/fim.py index 12051ef..6ab6541 100644 --- a/automation_file/core/fim.py +++ b/automation_file/core/fim.py @@ -23,7 +23,7 @@ from pathlib import Path from typing import Any -from automation_file.core.manifest import ManifestException, verify_manifest +from automation_file.core.manifest import verify_manifest from automation_file.exceptions import FileAutomationException from automation_file.logging_config import file_automation_logger from automation_file.notify import NotificationManager, notification_manager @@ -86,7 +86,7 @@ def check_once(self) -> dict[str, Any]: """Run one verification pass and return the summary.""" try: summary = verify_manifest(self._root, self._manifest_path) - except (ManifestException, FileAutomationException) as err: + except FileAutomationException as err: file_automation_logger.error("integrity_monitor: verify failed: %r", err) summary = { "matched": [], diff --git a/automation_file/core/metrics.py b/automation_file/core/metrics.py index f9d7c7f..ee8fcc4 100644 --- a/automation_file/core/metrics.py +++ b/automation_file/core/metrics.py @@ -56,7 +56,7 @@ def record_action(action: str, duration_seconds: float, ok: bool) -> None: try: ACTION_COUNT.labels(action=action, status=status).inc() ACTION_DURATION.labels(action=action).observe(max(0.0, float(duration_seconds))) - except Exception as err: # pragma: no cover - defensive + except Exception as err: # pylint: disable=broad-except # pragma: no cover - defensive file_automation_logger.error("metrics.record_action failed: %r", err) diff --git a/automation_file/core/package_loader.py b/automation_file/core/package_loader.py index 46d4564..6f4a23e 100644 --- a/automation_file/core/package_loader.py +++ b/automation_file/core/package_loader.py @@ -32,11 +32,11 @@ def load(self, package: str) -> ModuleType | None: file_automation_logger.error("PackageLoader: cannot find %s", package) return None try: - # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import # `package` is a trusted caller-supplied name (see PackageLoader docstring and # the CLAUDE.md security note on plugin loading); it is not untrusted input. - module = import_module(spec.name) - except (ImportError, ModuleNotFoundError) as error: + name = spec.name + module = import_module(name) # nosemgrep + except ImportError as error: file_automation_logger.error("PackageLoader import error: %r", error) return None self._cache[package] = module diff --git a/automation_file/core/plugins.py b/automation_file/core/plugins.py index 297d2d2..c614489 100644 --- a/automation_file/core/plugins.py +++ b/automation_file/core/plugins.py @@ -83,4 +83,4 @@ def _iter_entry_points() -> list[EntryPoint]: except TypeError: # importlib.metadata before 3.10 used a different API; the project # targets 3.10+, so this branch exists only as defensive padding. - return list(entry_points().get(ENTRY_POINT_GROUP, [])) + return list(entry_points().get(ENTRY_POINT_GROUP, [])) # pylint: disable=no-member diff --git a/automation_file/core/substitution.py b/automation_file/core/substitution.py index 67578c3..44d43ef 100644 --- a/automation_file/core/substitution.py +++ b/automation_file/core/substitution.py @@ -23,7 +23,7 @@ from automation_file.exceptions import FileAutomationException -_PATTERN = re.compile(r"\$\{([a-zA-Z_][a-zA-Z0-9_]*)(?::([^}]*))?\}") +_PATTERN = re.compile(r"\$\{([a-zA-Z_]\w*)(?::([^}]*))?\}", re.ASCII) class SubstitutionException(FileAutomationException): diff --git a/automation_file/local/dir_ops.py b/automation_file/local/dir_ops.py index 7648106..31e462b 100644 --- a/automation_file/local/dir_ops.py +++ b/automation_file/local/dir_ops.py @@ -18,7 +18,7 @@ def copy_dir(dir_path: str, target_dir_path: str) -> bool: shutil.copytree(source, Path(target_dir_path), dirs_exist_ok=True) file_automation_logger.info("copy_dir: %s -> %s", source, target_dir_path) return True - except (OSError, shutil.Error) as error: + except OSError as error: file_automation_logger.error("copy_dir failed: %r", error) return False @@ -32,7 +32,7 @@ def remove_dir_tree(dir_path: str) -> bool: shutil.rmtree(path) file_automation_logger.info("remove_dir_tree: %s", path) return True - except (OSError, shutil.Error) as error: + except OSError as error: file_automation_logger.error("remove_dir_tree failed: %r", error) return False diff --git a/automation_file/local/file_ops.py b/automation_file/local/file_ops.py index 9e3287f..bae3d46 100644 --- a/automation_file/local/file_ops.py +++ b/automation_file/local/file_ops.py @@ -22,7 +22,7 @@ def copy_file(file_path: str, target_path: str, copy_metadata: bool = True) -> b shutil.copy(source, target_path) file_automation_logger.info("copy_file: %s -> %s", source, target_path) return True - except (OSError, shutil.Error) as error: + except OSError as error: file_automation_logger.error("copy_file failed: %r", error) return False @@ -65,7 +65,7 @@ def copy_all_file_to_dir(dir_path: str, target_dir_path: str) -> bool: shutil.move(str(source), str(destination)) file_automation_logger.info("copy_all_file_to_dir: %s -> %s", source, destination) return True - except (OSError, shutil.Error) as error: + except OSError as error: file_automation_logger.error("copy_all_file_to_dir failed: %r", error) return False diff --git a/automation_file/local/json_edit.py b/automation_file/local/json_edit.py index 5b4d101..a7a5b33 100644 --- a/automation_file/local/json_edit.py +++ b/automation_file/local/json_edit.py @@ -114,59 +114,76 @@ def _child(container: Any, segment: str) -> Any: raise TypeError(f"cannot index {type(container).__name__} by {segment!r}") +def _is_int_segment(segment: str) -> bool: + return segment.lstrip("-").isdigit() + + +def _descend_for_set(container: Any, segment: str) -> Any: + if isinstance(container, MutableMapping): + if segment not in container or not isinstance( + container[segment], (MutableMapping, MutableSequence) + ): + container[segment] = {} + return container[segment] + if isinstance(container, MutableSequence) and _is_int_segment(segment): + return container[int(segment)] + raise JsonEditException(f"cannot traverse into {segment!r}") + + +def _assign_into_sequence(container: MutableSequence[Any], last: str, value: Any) -> None: + idx = int(last) + if -len(container) <= idx < len(container): + container[idx] = value + return + if idx == len(container): + container.append(value) + return + raise JsonEditException(f"list index out of range: {idx}") + + def _set_in(data: Any, segments: list[str], value: Any) -> None: container = data for segment in segments[:-1]: - if isinstance(container, MutableMapping): - if segment not in container or not isinstance( - container[segment], (MutableMapping, MutableSequence) - ): - container[segment] = {} - container = container[segment] - elif isinstance(container, MutableSequence) and segment.lstrip("-").isdigit(): - container = container[int(segment)] - else: - raise JsonEditException(f"cannot traverse into {segment!r}") + container = _descend_for_set(container, segment) last = segments[-1] if isinstance(container, MutableMapping): container[last] = value return - if isinstance(container, MutableSequence) and last.lstrip("-").isdigit(): - idx = int(last) - if -len(container) <= idx < len(container): - container[idx] = value - return - if idx == len(container): - container.append(value) - return - raise JsonEditException(f"list index out of range: {idx}") + if isinstance(container, MutableSequence) and _is_int_segment(last): + _assign_into_sequence(container, last, value) + return raise JsonEditException(f"cannot set into {type(container).__name__}") -def _delete_in(data: Any, segments: list[str]) -> bool: - container = data - for segment in segments[:-1]: - if isinstance(container, MutableMapping): - if segment not in container: - return False - container = container[segment] - elif isinstance(container, MutableSequence) and segment.lstrip("-").isdigit(): - idx = int(segment) - if not -len(container) <= idx < len(container): - return False - container = container[idx] - else: - return False - last = segments[-1] +def _descend_for_delete(container: Any, segment: str) -> Any: + if isinstance(container, MutableMapping): + return container.get(segment, _MISSING) + if isinstance(container, MutableSequence) and _is_int_segment(segment): + idx = int(segment) + if -len(container) <= idx < len(container): + return container[idx] + return _MISSING + + +def _remove_last(container: Any, last: str) -> bool: if isinstance(container, MutableMapping): if last not in container: return False del container[last] return True - if isinstance(container, MutableSequence) and last.lstrip("-").isdigit(): + if isinstance(container, MutableSequence) and _is_int_segment(last): idx = int(last) if not -len(container) <= idx < len(container): return False del container[idx] return True return False + + +def _delete_in(data: Any, segments: list[str]) -> bool: + container = data + for segment in segments[:-1]: + container = _descend_for_delete(container, segment) + if container is _MISSING: + return False + return _remove_last(container, segments[-1]) diff --git a/automation_file/local/shell_ops.py b/automation_file/local/shell_ops.py index e7e2a2c..46700c6 100644 --- a/automation_file/local/shell_ops.py +++ b/automation_file/local/shell_ops.py @@ -11,7 +11,7 @@ from __future__ import annotations -import subprocess +import subprocess # nosec B404 — used only via _run_subprocess with argv list + shell=False from collections.abc import Mapping, Sequence from pathlib import Path @@ -25,27 +25,7 @@ class ShellException(FileAutomationException): """Raised when a shell command fails (non-zero exit, timeout, or bad argv).""" -def run_shell( - argv: Sequence[str], - *, - timeout: float = _DEFAULT_TIMEOUT_SECONDS, - cwd: str | None = None, - env: Mapping[str, str] | None = None, - check: bool = True, - capture_output: bool = True, -) -> dict[str, object]: - """Run ``argv`` as a subprocess with a hard timeout. - - ``argv`` must be a non-empty list / tuple of strings. Strings are - rejected to block shell-injection via concatenated user input. - - Returns ``{"returncode": int, "stdout": str, "stderr": str}`` when - ``capture_output=True``; ``stdout``/``stderr`` are empty strings - otherwise. - - Raises :class:`ShellException` when ``check=True`` and the process - returns a non-zero exit code or times out. - """ +def _validate_argv(argv: Sequence[str], timeout: float) -> list[str]: if isinstance(argv, str) or not isinstance(argv, Sequence): raise ShellException("argv must be a list of strings, not a single string") argv_list = list(argv) @@ -53,12 +33,18 @@ def run_shell( raise ShellException("argv must be a non-empty list of strings") if timeout <= 0: raise ShellException("timeout must be positive") + return argv_list - cwd_path = str(Path(cwd)) if cwd else None - env_dict = dict(env) if env is not None else None +def _run_subprocess( + argv_list: list[str], + timeout: float, + cwd_path: str | None, + env_dict: dict[str, str] | None, + capture_output: bool, +) -> subprocess.CompletedProcess[str]: try: - completed = subprocess.run( + return subprocess.run( # nosec B603 nosemgrep — argv_list validated; shell=False by default argv_list, timeout=timeout, cwd=cwd_path, @@ -75,6 +61,33 @@ def run_shell( except OSError as err: raise ShellException(f"subprocess failed: {err!r}") from err + +def run_shell( + argv: Sequence[str], + *, + timeout: float = _DEFAULT_TIMEOUT_SECONDS, + cwd: str | None = None, + env: Mapping[str, str] | None = None, + check: bool = True, + capture_output: bool = True, +) -> dict[str, object]: + """Run ``argv`` as a subprocess with a hard timeout. + + ``argv`` must be a non-empty list / tuple of strings. Strings are + rejected to block shell-injection via concatenated user input. + + Returns ``{"returncode": int, "stdout": str, "stderr": str}`` when + ``capture_output=True``; ``stdout``/``stderr`` are empty strings + otherwise. + + Raises :class:`ShellException` when ``check=True`` and the process + returns a non-zero exit code or times out. + """ + argv_list = _validate_argv(argv, timeout) + cwd_path = str(Path(cwd)) if cwd else None + env_dict = dict(env) if env is not None else None + completed = _run_subprocess(argv_list, timeout, cwd_path, env_dict, capture_output) + result: dict[str, object] = { "returncode": completed.returncode, "stdout": completed.stdout or "" if capture_output else "", diff --git a/automation_file/local/sync_ops.py b/automation_file/local/sync_ops.py index 32c4976..18ed95a 100644 --- a/automation_file/local/sync_ops.py +++ b/automation_file/local/sync_ops.py @@ -69,10 +69,10 @@ def sync_dir( src_entries = _walk_relative(source) for rel in src_entries: - _process_source_entry(source, destination, rel, compare, dry_run, summary) + _process_source_entry(source, destination, rel, compare, dry_run=dry_run, summary=summary) if delete: - _delete_extras(source, destination, src_entries, dry_run, summary) + _delete_extras(destination, src_entries, dry_run, summary) file_automation_logger.info( "sync_dir %s -> %s: copied=%d skipped=%d deleted=%d errors=%d (dry_run=%s)", @@ -103,6 +103,7 @@ def _process_source_entry( destination: Path, rel: Path, compare: str, + *, dry_run: bool, summary: dict[str, Any], ) -> None: @@ -149,7 +150,6 @@ def _copy_one(src: Path, dst: Path) -> None: def _delete_extras( - source: Path, destination: Path, src_entries: list[Path], dry_run: bool, diff --git a/automation_file/local/tar_ops.py b/automation_file/local/tar_ops.py index 36ab503..f33bb19 100644 --- a/automation_file/local/tar_ops.py +++ b/automation_file/local/tar_ops.py @@ -87,8 +87,6 @@ def extract_tar(source: str, target_dir: str) -> list[str]: else: archive.extract(member, str(dest)) extracted.append(member.name) - except PathTraversalException: - raise except (OSError, tarfile.TarError) as err: raise TarException(f"extract_tar failed: {err}") from err @@ -96,21 +94,25 @@ def extract_tar(source: str, target_dir: str) -> list[str]: return extracted +def _verify_member_link(member: tarfile.TarInfo, dest_resolved: Path) -> None: + if not (member.issym() or member.islnk()): + return + link = member.linkname + link_path_source = Path(link) + link_path = ( + link_path_source.resolve() + if link_path_source.is_absolute() + else (dest_resolved / link).resolve() + ) + if not is_within(str(dest_resolved), str(link_path)): + kind = "symlink" if member.issym() else "hardlink" + raise PathTraversalException(f"tar {kind} escapes target: {member.name} -> {link}") + + def _verify_members(archive: tarfile.TarFile, dest: Path) -> None: dest_resolved = dest.resolve() for member in archive.getmembers(): candidate = (dest_resolved / member.name).resolve() if not is_within(str(dest_resolved), str(candidate)): raise PathTraversalException(f"tar member escapes target: {member.name}") - if member.issym() or member.islnk(): - link = member.linkname - link_path = ( - (dest_resolved / link).resolve() - if not Path(link).is_absolute() - else Path(link).resolve() - ) - if not is_within(str(dest_resolved), str(link_path)): - raise PathTraversalException( - f"tar {('symlink' if member.issym() else 'hardlink')} escapes target: " - f"{member.name} -> {link}" - ) + _verify_member_link(member, dest_resolved) diff --git a/automation_file/notify/manager.py b/automation_file/notify/manager.py index f50b996..fc05164 100644 --- a/automation_file/notify/manager.py +++ b/automation_file/notify/manager.py @@ -67,6 +67,11 @@ def list(self) -> list[dict[str, Any]]: sinks = list(self._sinks.values()) return [_describe(sink) for sink in sinks] + def has_sinks(self) -> bool: + """Return whether at least one sink is currently registered.""" + with self._lock: + return bool(self._sinks) + def notify( self, subject: str, @@ -148,9 +153,8 @@ def notify_on_failure(context: str, error: FileAutomationException | Exception) Does nothing when no sinks are registered, so callers can call this unconditionally without having to check the configuration. """ - with notification_manager._lock: - if not notification_manager._sinks: - return + if not notification_manager.has_sinks(): + return try: notification_manager.notify( f"automation_file: {context} failed", repr(error), level="error" diff --git a/automation_file/notify/sinks.py b/automation_file/notify/sinks.py index 3576e47..47733c7 100644 --- a/automation_file/notify/sinks.py +++ b/automation_file/notify/sinks.py @@ -23,6 +23,7 @@ LEVELS = frozenset({"info", "warning", "error"}) _DEFAULT_TIMEOUT = 10.0 _MAX_BODY_BYTES = 64 * 1024 +_JSON_HEADERS: dict[str, str] = {"Content-Type": "application/json"} class NotificationException(FileAutomationException): @@ -72,7 +73,7 @@ def __init__( def send(self, subject: str, body: str, level: str = "info") -> None: self._check_level(level) payload = {"subject": subject, "body": self._truncate(body), "level": level} - headers = {"Content-Type": "application/json"} + headers = dict(_JSON_HEADERS) headers.update(self.extra_headers) try: response = requests.post( @@ -125,7 +126,7 @@ def send(self, subject: str, body: str, level: str = "info") -> None: response = requests.post( self._url, data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"}, + headers=_JSON_HEADERS, timeout=self.timeout, allow_redirects=False, ) @@ -144,6 +145,7 @@ class EmailSink(NotificationSink): held as an instance attribute and used once per ``send``. """ + # pylint: disable-next=too-many-arguments def __init__( self, *, @@ -236,7 +238,7 @@ def send(self, subject: str, body: str, level: str = "info") -> None: response = requests.post( self._url, data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"}, + headers=_JSON_HEADERS, timeout=self.timeout, allow_redirects=False, ) @@ -284,7 +286,7 @@ def send(self, subject: str, body: str, level: str = "info") -> None: response = requests.post( self._url, data=json.dumps({"content": content}).encode("utf-8"), - headers={"Content-Type": "application/json"}, + headers=_JSON_HEADERS, timeout=self.timeout, allow_redirects=False, ) @@ -331,7 +333,7 @@ def send(self, subject: str, body: str, level: str = "info") -> None: response = requests.post( self._url, data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"}, + headers=_JSON_HEADERS, timeout=self.timeout, allow_redirects=False, ) @@ -391,7 +393,7 @@ def send(self, subject: str, body: str, level: str = "info") -> None: response = requests.post( self._ENQUEUE_URL, data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"}, + headers=_JSON_HEADERS, timeout=self.timeout, allow_redirects=False, ) diff --git a/automation_file/remote/ftp/client.py b/automation_file/remote/ftp/client.py index 2e6f0a0..a2767ae 100644 --- a/automation_file/remote/ftp/client.py +++ b/automation_file/remote/ftp/client.py @@ -10,7 +10,7 @@ import contextlib from dataclasses import dataclass -from ftplib import FTP, FTP_TLS # nosec B321 - plaintext FTP is opt-in via tls=False +from ftplib import FTP, FTP_TLS # nosec B321,B402 - plaintext FTP is opt-in via tls=False from typing import Any from automation_file.exceptions import FileAutomationException diff --git a/automation_file/remote/ftp/upload_ops.py b/automation_file/remote/ftp/upload_ops.py index bbc8345..b9384c7 100644 --- a/automation_file/remote/ftp/upload_ops.py +++ b/automation_file/remote/ftp/upload_ops.py @@ -3,7 +3,7 @@ from __future__ import annotations import posixpath -from ftplib import error_perm +from ftplib import error_perm # nosec B402 — FTP transport is opt-in via client configuration from pathlib import Path from automation_file.exceptions import FileNotExistsException diff --git a/automation_file/remote/google_drive/download_ops.py b/automation_file/remote/google_drive/download_ops.py index a99ee0a..1c38108 100644 --- a/automation_file/remote/google_drive/download_ops.py +++ b/automation_file/remote/google_drive/download_ops.py @@ -48,7 +48,7 @@ def drive_download_file_from_folder(folder_name: str) -> dict[str, str] | None: try: folders = ( service.files() - .list(q=(f"mimeType = 'application/vnd.google-apps.folder' and name = '{folder_name}'")) + .list(q=f"mimeType = 'application/vnd.google-apps.folder' and name = '{folder_name}'") .execute() ) folder_list = folders.get("files", []) diff --git a/automation_file/remote/http_download.py b/automation_file/remote/http_download.py index 4f135d0..f0e0fa3 100644 --- a/automation_file/remote/http_download.py +++ b/automation_file/remote/http_download.py @@ -4,7 +4,9 @@ import contextlib import os +from dataclasses import dataclass from pathlib import Path +from typing import Any import requests from tqdm import tqdm @@ -31,6 +33,19 @@ ) +@dataclass(frozen=True) +class _StreamContext: + """Bundle of per-download stream knobs; avoids 10+ positional args.""" + + write_mode: str + chunk_size: int + start_byte: int + total_size: int + max_bytes: int + reporter: ProgressReporter | None + token: Any + + @retry_on_transient(max_attempts=3, backoff_base=0.5, retriable=_RETRIABLE_EXCEPTIONS) def _open_stream( file_url: str, @@ -49,6 +64,121 @@ def _open_stream( return response +def _resume_layout(target: Path, resume: bool) -> tuple[Path, int, str]: + part_path = target.with_suffix(target.suffix + ".part") if resume else target + start_byte = part_path.stat().st_size if resume and part_path.exists() else 0 + write_mode = "ab" if start_byte > 0 else "wb" + return part_path, start_byte, write_mode + + +def _open_stream_logged(file_url: str, timeout: int, start_byte: int) -> requests.Response | None: + try: + return _open_stream(file_url, timeout, start_byte=start_byte) + except RetryExhaustedException as error: + file_automation_logger.error("download_file retries exhausted: %r", error) + except requests.exceptions.HTTPError as error: + file_automation_logger.error("download_file HTTP error: %r", error) + except requests.exceptions.RequestException as error: + file_automation_logger.error("download_file request error: %r", error) + return None + + +def _stream_to_disk( + response: requests.Response, + part_path: Path, + target: Path, + ctx: _StreamContext, +) -> int | None: + """Stream ``response`` into ``part_path``. Returns bytes written, or None on failure.""" + written = ctx.start_byte + with ( + open(part_path, ctx.write_mode) as output, # pylint: disable=unspecified-encoding + _progress(ctx.total_size, str(target)) as progress, + ): + if ctx.start_byte > 0: + progress.update(ctx.start_byte) + for chunk in response.iter_content(chunk_size=ctx.chunk_size): + if ctx.token is not None: + ctx.token.raise_if_cancelled() + if not chunk: + continue + written += len(chunk) + if written > ctx.max_bytes: + file_automation_logger.error( + "download_file aborted: stream exceeded %d bytes", ctx.max_bytes + ) + return None + output.write(chunk) + progress.update(len(chunk)) + if ctx.reporter is not None: + ctx.reporter.update(len(chunk)) + return written + + +def _verify_and_finalize( + target: Path, + expected_sha256: str | None, + reporter: ProgressReporter | None, +) -> bool: + if expected_sha256 and not verify_checksum(target, expected_sha256): + file_automation_logger.error("download_file checksum mismatch for %s; removing", target) + with contextlib.suppress(OSError): + target.unlink() + if reporter is not None: + reporter.finish(status="checksum_failed") + return False + if reporter is not None: + reporter.finish(status="done") + return True + + +def _reject_oversize(total_size: int, max_bytes: int) -> bool: + if total_size > max_bytes: + file_automation_logger.error( + "download_file rejected: content-length %d > %d", total_size, max_bytes + ) + return True + return False + + +def _make_reporter( + progress_name: str | None, + total_size: int, + start_byte: int, +) -> tuple[ProgressReporter | None, Any]: + if not progress_name: + return None, None + reporter, token = progress_registry.create(progress_name, total=total_size or None) + if start_byte > 0: + reporter.update(start_byte) + return reporter, token + + +def _run_stream( + response: requests.Response, + part_path: Path, + target: Path, + ctx: _StreamContext, + progress_name: str | None, +) -> int | None: + try: + written = _stream_to_disk(response, part_path, target, ctx) + except CancelledException: + file_automation_logger.warning("download_file cancelled: %s", progress_name) + if ctx.reporter is not None: + ctx.reporter.finish(status="cancelled") + return None + except OSError as error: + file_automation_logger.error("download_file write error: %r", error) + if ctx.reporter is not None: + ctx.reporter.finish(status="error") + return None + if written is None and ctx.reporter is not None: + ctx.reporter.finish(status="aborted") + return written + + +# pylint: disable-next=too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-return-statements def download_file( file_url: str, file_name: str, @@ -82,86 +212,37 @@ def download_file( return False target = Path(file_name) - part_path = target.with_suffix(target.suffix + ".part") if resume else target - start_byte = part_path.stat().st_size if resume and part_path.exists() else 0 - write_mode = "ab" if start_byte > 0 else "wb" + part_path, start_byte, write_mode = _resume_layout(target, resume) - try: - response = _open_stream(file_url, timeout, start_byte=start_byte) - except RetryExhaustedException as error: - file_automation_logger.error("download_file retries exhausted: %r", error) - return False - except requests.exceptions.HTTPError as error: - file_automation_logger.error("download_file HTTP error: %r", error) - return False - except requests.exceptions.RequestException as error: - file_automation_logger.error("download_file request error: %r", error) + response = _open_stream_logged(file_url, timeout, start_byte) + if response is None: return False - remaining = int(response.headers.get("content-length", 0)) - total_size = start_byte + remaining - if total_size > max_bytes: - file_automation_logger.error( - "download_file rejected: content-length %d > %d", total_size, max_bytes - ) + total_size = start_byte + int(response.headers.get("content-length", 0)) + if _reject_oversize(total_size, max_bytes): return False - reporter: ProgressReporter | None = None - token = None - if progress_name: - reporter, token = progress_registry.create(progress_name, total=total_size or None) - if start_byte > 0: - reporter.update(start_byte) + reporter, token = _make_reporter(progress_name, total_size, start_byte) + ctx = _StreamContext( + write_mode=write_mode, + chunk_size=chunk_size, + start_byte=start_byte, + total_size=total_size, + max_bytes=max_bytes, + reporter=reporter, + token=token, + ) - written = start_byte - try: - with ( - open(part_path, write_mode) as output, - _progress(total_size, str(target)) as progress, - ): - if start_byte > 0: - progress.update(start_byte) - for chunk in response.iter_content(chunk_size=chunk_size): - if token is not None: - token.raise_if_cancelled() - if not chunk: - continue - written += len(chunk) - if written > max_bytes: - file_automation_logger.error( - "download_file aborted: stream exceeded %d bytes", max_bytes - ) - if reporter is not None: - reporter.finish(status="aborted") - return False - output.write(chunk) - progress.update(len(chunk)) - if reporter is not None: - reporter.update(len(chunk)) - except CancelledException: - file_automation_logger.warning("download_file cancelled: %s", progress_name) - if reporter is not None: - reporter.finish(status="cancelled") - return False - except OSError as error: - file_automation_logger.error("download_file write error: %r", error) - if reporter is not None: - reporter.finish(status="error") + written = _run_stream(response, part_path, target, ctx, progress_name) + if written is None: return False if resume and part_path != target: os.replace(part_path, target) - if expected_sha256 and not verify_checksum(target, expected_sha256): - file_automation_logger.error("download_file checksum mismatch for %s; removing", target) - with contextlib.suppress(OSError): - target.unlink() - if reporter is not None: - reporter.finish(status="checksum_failed") + if not _verify_and_finalize(target, expected_sha256, reporter): return False - if reporter is not None: - reporter.finish(status="done") file_automation_logger.info("download_file: %s -> %s (%d bytes)", file_url, target, written) return True diff --git a/automation_file/remote/s3/download_ops.py b/automation_file/remote/s3/download_ops.py index 137590f..4f91bff 100644 --- a/automation_file/remote/s3/download_ops.py +++ b/automation_file/remote/s3/download_ops.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Callable from pathlib import Path from automation_file.core.progress import ( @@ -25,21 +26,25 @@ def s3_download_file( """ client = s3_instance.require_client() Path(target_path).parent.mkdir(parents=True, exist_ok=True) - callback = None + callback: Callable[[int], None] | None = None reporter = None token = None if progress_name: - total = None + total: int | None = None try: head = client.head_object(Bucket=bucket, Key=key) total = int(head.get("ContentLength", 0)) or None except Exception: # pylint: disable=broad-except total = None reporter, token = progress_registry.create(progress_name, total=total) + _reporter = reporter + _token = token - def callback(bytes_transferred: int) -> None: - token.raise_if_cancelled() - reporter.update(bytes_transferred) + def _progress_callback(bytes_transferred: int) -> None: + _token.raise_if_cancelled() + _reporter.update(bytes_transferred) + + callback = _progress_callback try: client.download_file(bucket, key, target_path, Callback=callback) diff --git a/automation_file/remote/s3/upload_ops.py b/automation_file/remote/s3/upload_ops.py index dadd85b..83c893e 100644 --- a/automation_file/remote/s3/upload_ops.py +++ b/automation_file/remote/s3/upload_ops.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Callable from pathlib import Path from automation_file.core.progress import ( @@ -29,15 +30,19 @@ def s3_upload_file( if not path.is_file(): raise FileNotExistsException(str(path)) client = s3_instance.require_client() - callback = None + callback: Callable[[int], None] | None = None reporter = None token = None if progress_name: reporter, token = progress_registry.create(progress_name, total=path.stat().st_size) + _reporter = reporter + _token = token - def callback(bytes_transferred: int) -> None: - token.raise_if_cancelled() - reporter.update(bytes_transferred) + def _progress_callback(bytes_transferred: int) -> None: + _token.raise_if_cancelled() + _reporter.update(bytes_transferred) + + callback = _progress_callback try: client.upload_file(str(path), bucket, key, Callback=callback) diff --git a/automation_file/scheduler/cron.py b/automation_file/scheduler/cron.py index 0dc6d87..08b0b5d 100644 --- a/automation_file/scheduler/cron.py +++ b/automation_file/scheduler/cron.py @@ -64,6 +64,40 @@ def _expand_range(start: int, end: int, step: int, low: int, high: int) -> set[i return set(range(start, end + 1, step)) +def _split_step(chunk: str) -> tuple[str, int]: + if "/" not in chunk: + return chunk, 1 + base, step_text = chunk.split("/", 1) + try: + return base, int(step_text) + except ValueError as error: + raise CronException(f"cron: bad step {step_text!r}") from error + + +def _expand_chunk( + chunk: str, + step: int, + field_index: int, + low: int, + high: int, + *, + effective_high: int, +) -> set[int]: + if chunk == "*": + return _expand_range(low, high, step, low, high) + if "-" in chunk: + start_text, end_text = chunk.split("-", 1) + start = _parse_value(start_text, field_index) + end = _parse_value(end_text, field_index) + return _expand_range(start, end, step, low, effective_high) + value = _parse_value(chunk, field_index) + if value < low or value > effective_high: + raise CronException(f"cron: value {value} outside [{low},{high}]") + if step == 1: + return {value} + return _expand_range(value, effective_high, step, low, effective_high) + + def _parse_field(raw: str, field_index: int) -> frozenset[int]: low, high = _FIELD_BOUNDS[field_index] # DoW accepts 7 as an alias for Sunday (0) before range validation. @@ -73,30 +107,8 @@ def _parse_field(raw: str, field_index: int) -> frozenset[int]: chunk = part.strip() if not chunk: raise CronException(f"cron: empty chunk in field {field_index}") - step = 1 - if "/" in chunk: - base, step_text = chunk.split("/", 1) - try: - step = int(step_text) - except ValueError as error: - raise CronException(f"cron: bad step {step_text!r}") from error - chunk = base - if chunk == "*": - result |= _expand_range(low, high, step, low, high) - continue - if "-" in chunk: - start_text, end_text = chunk.split("-", 1) - start = _parse_value(start_text, field_index) - end = _parse_value(end_text, field_index) - result |= _expand_range(start, end, step, low, effective_high) - continue - value = _parse_value(chunk, field_index) - if value < low or value > effective_high: - raise CronException(f"cron: value {value} outside [{low},{high}]") - if step == 1: - result.add(value) - else: - result |= _expand_range(value, effective_high, step, low, effective_high) + base, step = _split_step(chunk) + result |= _expand_chunk(base, step, field_index, low, high, effective_high=effective_high) if field_index == 4 and 7 in result: result.discard(7) result.add(0) diff --git a/automation_file/ui/tabs/local_tab.py b/automation_file/ui/tabs/local_tab.py index 327dba6..4abbd3a 100644 --- a/automation_file/ui/tabs/local_tab.py +++ b/automation_file/ui/tabs/local_tab.py @@ -41,12 +41,40 @@ class _ActionEntry(NamedTuple): build: Callable[[LocalOpsTab], QWidget] +# pylint: disable-next=too-many-instance-attributes class LocalOpsTab(BaseTab): - """Dropdown-driven local file, directory, and ZIP operations.""" + """Dropdown-driven local file, directory, and ZIP operations. + + Each operation owns its own input widgets (one QLineEdit per field, + plus a content QTextEdit for create-file). That is intentional — the + dropdown switches between pre-built pages rather than reusing widgets + across operations — so the attribute count is inherent to the design. + """ def __init__(self, log, pool) -> None: super().__init__(log, pool) + # Widgets populated lazily by the _page_* builders. + self._create_path: QLineEdit + self._create_content: QTextEdit + self._copy_src: QLineEdit + self._copy_dst: QLineEdit + self._rename_src: QLineEdit + self._rename_dst: QLineEdit + self._remove_path: QLineEdit + self._dir_create: QLineEdit + self._dir_copy_src: QLineEdit + self._dir_copy_dst: QLineEdit + self._dir_rename_src: QLineEdit + self._dir_rename_dst: QLineEdit + self._dir_remove: QLineEdit + self._zip_file_path: QLineEdit + self._zip_file_name: QLineEdit + self._zip_dir_path: QLineEdit + self._zip_dir_name: QLineEdit + self._unzip_archive: QLineEdit + self._unzip_target: QLineEdit + entries = self._entries() self._picker = QComboBox() self._stack = QStackedWidget() diff --git a/automation_file/ui/tabs/progress_tab.py b/automation_file/ui/tabs/progress_tab.py index 2515388..35354e8 100644 --- a/automation_file/ui/tabs/progress_tab.py +++ b/automation_file/ui/tabs/progress_tab.py @@ -102,15 +102,15 @@ def _refresh(self) -> None: self._set_cell(row, 4, _format_bytes(total) if total else "unknown") def _progress_widget(self, data: dict[str, Any]) -> QProgressBar: - bar = QProgressBar() + progress_bar = QProgressBar() total = data["total"] transferred = data["transferred"] if total and total > 0: - bar.setRange(0, int(total)) - bar.setValue(min(int(transferred), int(total))) + progress_bar.setRange(0, int(total)) + progress_bar.setValue(min(int(transferred), int(total))) else: - bar.setRange(0, 0) # busy indicator - return bar + progress_bar.setRange(0, 0) # busy indicator + return progress_bar def _set_cell(self, row: int, col: int, text: str) -> None: self._table.setItem(row, col, QTableWidgetItem(text)) diff --git a/automation_file/utils/deduplicate.py b/automation_file/utils/deduplicate.py index 7808ac3..d72f2b2 100644 --- a/automation_file/utils/deduplicate.py +++ b/automation_file/utils/deduplicate.py @@ -57,6 +57,18 @@ def find_duplicates( return groups +def _classify_entry(entry: os.DirEntry[str]) -> tuple[str, int | None]: + """Return ``("dir", None)``, ``("file", size)``, or ``("skip", None)``.""" + try: + if entry.is_dir(follow_symlinks=False): + return "dir", None + if not entry.is_file(follow_symlinks=False): + return "skip", None + return "file", entry.stat(follow_symlinks=False).st_size + except OSError: + return "skip", None + + def _group_by_size(root: Path, min_size: int) -> dict[int, list[str]]: buckets: dict[int, list[str]] = defaultdict(list) stack: list[str] = [str(root)] @@ -64,22 +76,15 @@ def _group_by_size(root: Path, min_size: int) -> dict[int, list[str]]: current = stack.pop() try: iterator = os.scandir(current) - except (PermissionError, FileNotFoundError, OSError): + except OSError: continue with iterator as entries: for entry in entries: - try: - if entry.is_dir(follow_symlinks=False): - stack.append(entry.path) - continue - if not entry.is_file(follow_symlinks=False): - continue - size = entry.stat(follow_symlinks=False).st_size - except OSError: - continue - if size < min_size: - continue - buckets[size].append(os.path.abspath(entry.path)) + kind, size = _classify_entry(entry) + if kind == "dir": + stack.append(entry.path) + elif kind == "file" and size is not None and size >= min_size: + buckets[size].append(os.path.abspath(entry.path)) return {size: paths for size, paths in buckets.items() if len(paths) > 1} diff --git a/automation_file/utils/fast_find.py b/automation_file/utils/fast_find.py index fd6bdc2..464ff68 100644 --- a/automation_file/utils/fast_find.py +++ b/automation_file/utils/fast_find.py @@ -21,7 +21,7 @@ import os import platform import shutil -import subprocess +import subprocess # nosec B404 — invoked only with fixed-name OS indexers (mdfind/locate) via argv from collections.abc import Iterable, Iterator from pathlib import Path @@ -92,6 +92,22 @@ def fast_find( return list(scandir_find(root_path, pattern, limit=limit, files_only=files_only)) +def _iter_directory(current: str) -> Iterator[os.DirEntry[str]]: + try: + iterator = os.scandir(current) + except OSError: + return + with iterator as entries: + yield from entries + + +def _entry_is_dir(entry: os.DirEntry[str]) -> bool | None: + try: + return entry.is_dir(follow_symlinks=False) + except OSError: + return None + + def scandir_find( root: str | os.PathLike[str], pattern: str = _DEFAULT_PATTERN, @@ -115,25 +131,19 @@ def scandir_find( lowered_pattern = pattern.lower() while stack: current = stack.pop() - try: - iterator = os.scandir(current) - except (PermissionError, FileNotFoundError, OSError): - continue - with iterator as entries: - for entry in entries: - try: - is_dir = entry.is_dir(follow_symlinks=False) - except OSError: + for entry in _iter_directory(current): + is_dir = _entry_is_dir(entry) + if is_dir is None: + continue + if is_dir: + stack.append(entry.path) + if files_only: continue - if is_dir: - stack.append(entry.path) - if files_only: - continue - if _matches(entry.name, lowered_pattern): - yield os.path.abspath(entry.path) - yielded += 1 - if limit is not None and yielded >= limit: - return + if _matches(entry.name, lowered_pattern): + yield os.path.abspath(entry.path) + yielded += 1 + if limit is not None and yielded >= limit: + return def _matches(name: str, lowered_pattern: str) -> bool: @@ -157,7 +167,8 @@ def _run_indexer( def _capture(argv: list[str]) -> list[str]: - completed = subprocess.run( + # argv[0] is a fixed-name indexer (mdfind/locate/es); shell=False. + completed = subprocess.run( # nosec B603 nosemgrep argv, capture_output=True, timeout=_INDEX_TIMEOUT_SECONDS, diff --git a/automation_file/utils/rotate.py b/automation_file/utils/rotate.py index 55f04f6..1f28f9c 100644 --- a/automation_file/utils/rotate.py +++ b/automation_file/utils/rotate.py @@ -26,6 +26,41 @@ class RotateException(FileAutomationException): """Raised when rotate_backups receives invalid arguments.""" +def _validate_keep_counts(counts: dict[str, int]) -> None: + for name, value in counts.items(): + if value < 0: + raise RotateException(f"{name} must be >= 0") + + +def _gather_candidates(root: Path, pattern: str) -> list[tuple[Path, float]]: + candidates = [ + (entry, entry.stat().st_mtime) + for entry in root.iterdir() + if entry.is_file() and fnmatch.fnmatch(entry.name, pattern) + ] + candidates.sort(key=lambda item: item[1], reverse=True) + return candidates + + +def _delete_unkept( + candidates: list[tuple[Path, float]], + kept_paths: set[Path], + dry_run: bool, +) -> list[str]: + deleted: list[str] = [] + for path, _mtime in candidates: + if path in kept_paths: + continue + deleted.append(str(path)) + if dry_run: + continue + try: + path.unlink() + except OSError as err: + file_automation_logger.error("rotate_backups: unlink %s failed: %r", path, err) + return deleted + + def rotate_backups( directory: str, pattern: str = "*", @@ -51,28 +86,20 @@ def rotate_backups( any enabled bucket. ``0`` disables a bucket. No file can appear in both ``kept`` and ``deleted``. """ - for name, value in ( - ("keep_daily", keep_daily), - ("keep_weekly", keep_weekly), - ("keep_monthly", keep_monthly), - ("keep_yearly", keep_yearly), - ): - if value < 0: - raise RotateException(f"{name} must be >= 0") + _validate_keep_counts( + { + "keep_daily": keep_daily, + "keep_weekly": keep_weekly, + "keep_monthly": keep_monthly, + "keep_yearly": keep_yearly, + } + ) root = Path(directory) if not root.is_dir(): raise RotateException(f"not a directory: {directory}") - candidates: list[tuple[Path, float]] = [] - for entry in root.iterdir(): - if not entry.is_file(): - continue - if not fnmatch.fnmatch(entry.name, pattern): - continue - candidates.append((entry, entry.stat().st_mtime)) - candidates.sort(key=lambda item: item[1], reverse=True) - + candidates = _gather_candidates(root, pattern) kept = _select_kept( candidates, keep_daily=keep_daily, @@ -81,17 +108,7 @@ def rotate_backups( keep_yearly=keep_yearly, ) kept_paths = {path for path, _ in kept} - deleted: list[str] = [] - for path, _mtime in candidates: - if path in kept_paths: - continue - deleted.append(str(path)) - if dry_run: - continue - try: - path.unlink() - except OSError as err: - file_automation_logger.error("rotate_backups: unlink %s failed: %r", path, err) + deleted = _delete_unkept(candidates, kept_paths, dry_run) file_automation_logger.info( "rotate_backups: kept=%d deleted=%d dry_run=%s", diff --git a/docs/source.zh-CN/index.rst b/docs/source.zh-CN/index.rst index c116e5d..ecc2c05 100644 --- a/docs/source.zh-CN/index.rst +++ b/docs/source.zh-CN/index.rst @@ -3,22 +3,111 @@ automation_file 语言：`English <../html/index.html>`_ | `繁體中文 <../html-zh-TW/index.html>`_ | **简体中文** -以自动化为核心的 Python 库，涵盖本地文件 / 目录 / zip 操作、HTTP 下载 -以及远程存储（Google Drive、S3、Azure Blob、Dropbox、SFTP）。内置 PySide6 图形界面， -把每一项功能以标签页的形式呈现。所有动作以 JSON 描述，统一通过 -:class:`~automation_file.core.action_registry.ActionRegistry` 调度。 +以自动化为核心的模块化框架，涵盖本地文件 / 目录 / ZIP 操作、经 SSRF 校验的 +HTTP 下载、远程存储（Google Drive、S3、Azure Blob、Dropbox、SFTP、FTP、 +WebDAV、SMB、fsspec），以及通过内建 TCP / HTTP 服务器执行的 JSON 动作。 +内置 PySide6 图形界面，把每一项功能以标签页形式呈现；所有公开功能统一从 +顶层 ``automation_file`` 外观模块重新导出。 -快速开始 +功能亮点 -------- -从 PyPI 安装并执行 JSON 动作列表： +**核心原语** + +* JSON 动作列表由共享的 + :class:`~automation_file.core.action_executor.ActionExecutor` 执行，支持 + 校验、dry-run、并行、DAG。 +* 路径穿越防护（:func:`~automation_file.local.safe_paths.safe_join`）、 + 对外 URL 的 SSRF 校验、默认仅绑定 loopback 的 TCP / HTTP 服务器， + 可选共享密钥验证与每动作 ACL。 +* 可靠性辅助：``retry_on_transient`` 装饰器、``Quota`` 流量与时间预算、 + 流式 checksum、可续传 HTTP 下载。 + +**后端集成** + +* 本地文件 / 目录 / ZIP / tar 操作。 +* HTTP 下载：SSRF 防护、大小 / 超时上限、重试、续传、可选 SHA-256 校验。 +* 一等公民后端：Google Drive、S3、Azure Blob、Dropbox、SFTP、FTP / FTPS、 + WebDAV、SMB / CIFS、fsspec — 全部自动注册。 +* 跨后端复制，使用 URI 语法（``local://``、``s3://``、``drive://``、 + ``sftp://``、``azure://``、``dropbox://``、``ftp://`` …）。 + +**事件驱动** + +* 文件监听触发器 ``FA_watch_*`` — 路径变动时自动执行动作列表。 +* Cron 调度（``FA_schedule_*``）采用纯标准库的 5 字段解析器， + 提供重叠保护，失败时自动通知。 +* 传输进度与取消 Token，通过 ``progress_name`` 对外暴露。 + +**可观测性与集成** + +* 通知 Sink — webhook / Slack / SMTP / Telegram / Discord / Teams / + PagerDuty，每个 Sink 独立隔离错误，并采用滑动窗口去重。 +* Prometheus 指标导出器（``start_metrics_server``）、SQLite 审计日志、 + 文件完整性监视器。 +* HTMX 网页面板（``start_web_ui``）、MCP 服务器将注册表桥接到 + Claude Desktop / MCP CLI，走 JSON-RPC 2.0。 +* PySide6 桌面 GUI（``python -m automation_file ui``）。 + +**供应链** + +* 配置文件与机密信息 — 在 ``automation_file.toml`` 声明 sink 与默认值； + ``${env:…}`` / ``${file:…}`` 引用通过 Env / File / Chained provider + 解析，避免把密钥写死在文件中。 +* 入口点插件 — 第三方包通过 + ``[project.entry-points."automation_file.actions"]`` + 注册自己的 ``FA_*`` 动作。 + +架构鸟瞰 +-------- + +.. code-block:: text + + 用户 / CLI / JSON batch + │ + ▼ + ┌─────────────────────────────────────────┐ + │ automation_file（外观） │ + │ execute_action、driver_instance、 │ + │ start_autocontrol_socket_server、 │ + │ start_http_action_server、Quota … │ + └─────────────────────────────────────────┘ + │ + ▼ + ┌──────────────┐ ┌────────────────────┐ + │ core │────▶│ ActionRegistry │ + │ executor、 │ │ （FA_* 指令） │ + │ retry、 │ └────────────────────┘ + │ quota、 │ │ + │ progress │ ▼ + └──────────────┘ ┌────────────────────┐ + │ local / remote / │ + │ server / triggers /│ + │ scheduler / ui │ + └────────────────────┘ + +完整的模块树与设计模式见 :doc:`architecture`。 + +安装 +---- .. code-block:: bash pip install automation_file + +所有后端（S3、Azure Blob、Dropbox、SFTP、PySide6）都是一等运行期 +依赖，常见使用场景不需要额外的 extras。 + +快速开始 +-------- + +用 CLI 执行 JSON 动作列表： + +.. code-block:: bash + python -m automation_file --execute_file my_actions.json -或直接通过 Python 代码调用： +直接从 Python 调用： .. code-block:: python @@ -27,8 +116,47 @@ automation_file execute_action([ ["FA_create_dir", {"dir_path": "build"}], ["FA_create_file", {"file_path": "build/hello.txt", "content": "hi"}], + ["FA_zip_dir", {"source": "build", "target": "build.zip"}], ]) +执行前先校验动作列表，或并行执行： + +.. code-block:: python + + from automation_file import executor + + problems = executor.validate(actions) + if problems: + raise SystemExit("\n".join(problems)) + executor.execute_action_parallel(actions, max_workers=4) + +启动 PySide6 图形界面： + +.. code-block:: bash + + python -m automation_file ui + +以共享密钥在 loopback 提供 HTTP 动作服务器： + +.. code-block:: python + + from automation_file import start_http_action_server + + server = start_http_action_server(port=8765, shared_secret="s3kret") + +动作列表的格式 +-------------- + +一个动作是三种 list 形式之一，按名称通过注册表调度： + +.. code-block:: python + + ["FA_create_dir"] # 无参数 + ["FA_create_dir", {"dir_path": "build"}] # 关键字参数 + ["FA_copy_file", ["src.txt", "dst.txt"]] # 位置参数 + +JSON 动作列表就是上述 list 的 list。 + .. toctree:: :maxdepth: 2 :caption: 目录 diff --git a/docs/source.zh-TW/index.rst b/docs/source.zh-TW/index.rst index 2c89396..ca6c70b 100644 --- a/docs/source.zh-TW/index.rst +++ b/docs/source.zh-TW/index.rst @@ -3,22 +3,111 @@ automation_file 語言：`English <../html/index.html>`_ | **繁體中文** | `简体中文 <../html-zh-CN/index.html>`_ -以自動化為核心的 Python 函式庫，涵蓋本地檔案 / 目錄 / zip 操作、HTTP 下載 -與遠端儲存（Google Drive、S3、Azure Blob、Dropbox、SFTP）。內建 PySide6 圖形介面， -把每一項功能以分頁形式呈現。所有動作以 JSON 描述，統一透過 -:class:`~automation_file.core.action_registry.ActionRegistry` 調度。 +以自動化為核心的模組化框架，涵蓋本地檔案 / 目錄 / ZIP 操作、經 SSRF 驗證的 +HTTP 下載、遠端儲存（Google Drive、S3、Azure Blob、Dropbox、SFTP、FTP、 +WebDAV、SMB、fsspec），以及透過內建 TCP / HTTP 伺服器執行的 JSON 動作。 +內建 PySide6 圖形介面，把每一項功能以分頁形式呈現；所有公開功能統一從 +頂層 ``automation_file`` 外觀模組重新匯出。 -快速開始 +功能亮點 -------- -從 PyPI 安裝並執行 JSON 動作清單： +**核心原語** + +* JSON 動作清單由共用的 + :class:`~automation_file.core.action_executor.ActionExecutor` 執行，支援 + 驗證、dry-run、平行、DAG。 +* 路徑穿越防護（:func:`~automation_file.local.safe_paths.safe_join`）、 + 對外 URL 的 SSRF 驗證、預設僅綁定 loopback 的 TCP / HTTP 伺服器， + 可選共享金鑰驗證與每動作 ACL。 +* 可靠性輔助：``retry_on_transient`` 裝飾器、``Quota`` 流量與時間上限、 + 串流式 checksum、可續傳 HTTP 下載。 + +**後端整合** + +* 本地檔案 / 目錄 / ZIP / tar 操作。 +* HTTP 下載：SSRF 防護、大小 / 逾時上限、重試、續傳、可選 SHA-256 驗證。 +* 第一方整合：Google Drive、S3、Azure Blob、Dropbox、SFTP、FTP / FTPS、 + WebDAV、SMB / CIFS、fsspec — 全部自動註冊。 +* 跨後端複製，使用 URI 語法（``local://``、``s3://``、``drive://``、 + ``sftp://``、``azure://``、``dropbox://``、``ftp://`` …）。 + +**事件驅動** + +* 檔案監看觸發器 ``FA_watch_*`` — 路徑變動時自動執行動作清單。 +* Cron 排程（``FA_schedule_*``）採用純標準函式庫的 5 欄位解析器， + 提供重疊保護，失敗時自動通知。 +* 傳輸進度與取消 Token，透過 ``progress_name`` 對外暴露。 + +**可觀測性與整合** + +* 通知 Sink — webhook / Slack / SMTP / Telegram / Discord / Teams / + PagerDuty，各 Sink 獨立隔離錯誤並採用滑動視窗去重。 +* Prometheus 指標匯出器（``start_metrics_server``）、SQLite 稽核日誌、 + 檔案完整性監視器。 +* HTMX 網頁面板（``start_web_ui``）、MCP 伺服器將註冊表橋接到 + Claude Desktop / MCP CLI，走 JSON-RPC 2.0。 +* PySide6 桌面 GUI（``python -m automation_file ui``）。 + +**供應鏈** + +* 設定檔與機敏資訊 — 在 ``automation_file.toml`` 宣告 sink 與預設值； + ``${env:…}`` / ``${file:…}`` 參考透過 Env / File / Chained provider + 解析，避免把金鑰寫死在檔案裡。 +* 進入點外掛 — 第三方套件透過 + ``[project.entry-points."automation_file.actions"]`` + 自行註冊 ``FA_*`` 動作。 + +架構鳥瞰 +-------- + +.. code-block:: text + + 使用者 / CLI / JSON batch + │ + ▼ + ┌─────────────────────────────────────────┐ + │ automation_file（外觀） │ + │ execute_action、driver_instance、 │ + │ start_autocontrol_socket_server、 │ + │ start_http_action_server、Quota … │ + └─────────────────────────────────────────┘ + │ + ▼ + ┌──────────────┐ ┌────────────────────┐ + │ core │────▶│ ActionRegistry │ + │ executor、 │ │ （FA_* 指令） │ + │ retry、 │ └────────────────────┘ + │ quota、 │ │ + │ progress │ ▼ + └──────────────┘ ┌────────────────────┐ + │ local / remote / │ + │ server / triggers /│ + │ scheduler / ui │ + └────────────────────┘ + +完整的模組樹與設計模式請見 :doc:`architecture`。 + +安裝 +---- .. code-block:: bash pip install automation_file + +所有後端（S3、Azure Blob、Dropbox、SFTP、PySide6）皆為第一方執行期 +依賴，常見使用情境不需要額外 extras。 + +快速開始 +-------- + +用 CLI 執行 JSON 動作清單： + +.. code-block:: bash + python -m automation_file --execute_file my_actions.json -或直接從 Python 程式碼呼叫： +直接從 Python 呼叫： .. code-block:: python @@ -27,8 +116,47 @@ automation_file execute_action([ ["FA_create_dir", {"dir_path": "build"}], ["FA_create_file", {"file_path": "build/hello.txt", "content": "hi"}], + ["FA_zip_dir", {"source": "build", "target": "build.zip"}], ]) +執行前先驗證動作清單，或以平行方式執行： + +.. code-block:: python + + from automation_file import executor + + problems = executor.validate(actions) + if problems: + raise SystemExit("\n".join(problems)) + executor.execute_action_parallel(actions, max_workers=4) + +啟動 PySide6 圖形介面： + +.. code-block:: bash + + python -m automation_file ui + +以共享金鑰在 loopback 提供 HTTP 動作伺服器： + +.. code-block:: python + + from automation_file import start_http_action_server + + server = start_http_action_server(port=8765, shared_secret="s3kret") + +動作清單的格式 +-------------- + +一個動作是三種 list 形式之一，依名稱透過註冊表調度： + +.. code-block:: python + + ["FA_create_dir"] # 無參數 + ["FA_create_dir", {"dir_path": "build"}] # 關鍵字參數 + ["FA_copy_file", ["src.txt", "dst.txt"]] # 位置參數 + +JSON 動作清單就是上述 list 的 list。 + .. toctree:: :maxdepth: 2 :caption: 目錄 diff --git a/docs/source/index.rst b/docs/source/index.rst index a50b4f8..961db5f 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -3,23 +3,113 @@ automation_file Languages: **English** | `繁體中文 <../html-zh-TW/index.html>`_ | `简体中文 <../html-zh-CN/index.html>`_ -Automation-first Python library for local file / directory / zip operations, -HTTP downloads, and remote storage (Google Drive, S3, Azure Blob, Dropbox, -SFTP). Ships with a PySide6 GUI that surfaces every feature through tabs. -Actions are defined as JSON and dispatched through a central -:class:`~automation_file.core.action_registry.ActionRegistry`. +A modular automation framework for local file / directory / ZIP operations, +SSRF-validated HTTP downloads, remote storage (Google Drive, S3, Azure Blob, +Dropbox, SFTP, FTP, WebDAV, SMB, fsspec), and JSON-driven action execution +over embedded TCP / HTTP servers. Ships with a PySide6 GUI that exposes every +feature through tabs. All public functionality is re-exported from the +top-level ``automation_file`` facade. -Getting started ---------------- +Highlights +---------- -Install from PyPI and run a JSON action list: +**Core primitives** + +* JSON action lists executed by a shared + :class:`~automation_file.core.action_executor.ActionExecutor` — validate, + dry-run, parallel, DAG. +* Path traversal guard (:func:`~automation_file.local.safe_paths.safe_join`), + SSRF validator for outbound URLs, loopback-first TCP / HTTP servers with + optional shared-secret auth and per-action ACLs. +* Reliability helpers: ``retry_on_transient`` decorator, ``Quota`` size and + time budgets, streaming checksums, resumable HTTP downloads. + +**Backends** + +* Local file / directory / ZIP / tar operations. +* HTTP downloads with SSRF protection, size / timeout caps, retries, + resumable transfers, and optional SHA-256 verification. +* First-class Google Drive, S3, Azure Blob, Dropbox, SFTP, FTP / FTPS, + WebDAV, SMB / CIFS, and fsspec backends — all auto-registered. +* Cross-backend copy through URI syntax (``local://``, ``s3://``, + ``drive://``, ``sftp://``, ``azure://``, ``dropbox://``, ``ftp://``, …). + +**Event-driven execution** + +* File-watcher triggers via ``FA_watch_*`` — run action lists on path + changes. +* Cron scheduler (``FA_schedule_*``) on a stdlib-only 5-field parser, with + overlap guard and auto-notify on failure. +* Transfer progress + cancellation tokens exposed through ``progress_name``. + +**Observability + integrations** + +* Notification sinks — webhook / Slack / SMTP / Telegram / Discord / Teams / + PagerDuty with per-sink error isolation and sliding-window dedup. +* Prometheus metrics exporter (``start_metrics_server``), SQLite audit log, + file integrity monitor. +* HTMX web UI (``start_web_ui``), MCP server bridging the registry to Claude + Desktop / MCP CLIs over JSON-RPC 2.0. +* PySide6 desktop GUI (``python -m automation_file ui``). + +**Supply chain** + +* Config + secrets — declare sinks and defaults in ``automation_file.toml``; + ``${env:…}`` / ``${file:…}`` references resolve through Env / File / + Chained providers so secrets stay out of the file. +* Entry-point plugins — third-party packages register their own ``FA_*`` + actions via ``[project.entry-points."automation_file.actions"]``. + +Architecture at a glance +------------------------ + +.. code-block:: text + + User / CLI / JSON batch + │ + ▼ + ┌─────────────────────────────────────────┐ + │ automation_file (facade) │ + │ execute_action, driver_instance, │ + │ start_autocontrol_socket_server, │ + │ start_http_action_server, Quota, … │ + └─────────────────────────────────────────┘ + │ + ▼ + ┌──────────────┐ ┌────────────────────┐ + │ core │────▶│ ActionRegistry │ + │ executor, │ │ (FA_* commands) │ + │ retry, │ └────────────────────┘ + │ quota, │ │ + │ progress │ ▼ + └──────────────┘ ┌────────────────────┐ + │ local / remote / │ + │ server / triggers /│ + │ scheduler / ui │ + └────────────────────┘ + +See :doc:`architecture` for the full module tree and design patterns. + +Installation +------------ .. code-block:: bash pip install automation_file + +All backends (S3, Azure Blob, Dropbox, SFTP, PySide6) are first-class +runtime dependencies — no extras required for common use. + +Quick start +----------- + +Run a JSON action list from the CLI: + +.. code-block:: bash + python -m automation_file --execute_file my_actions.json -Or drive the library directly from Python: +Drive the library from Python: .. code-block:: python @@ -28,8 +118,48 @@ Or drive the library directly from Python: execute_action([ ["FA_create_dir", {"dir_path": "build"}], ["FA_create_file", {"file_path": "build/hello.txt", "content": "hi"}], + ["FA_zip_dir", {"source": "build", "target": "build.zip"}], ]) +Validate a batch before running it, or run actions in parallel: + +.. code-block:: python + + from automation_file import executor + + problems = executor.validate(actions) + if problems: + raise SystemExit("\n".join(problems)) + executor.execute_action_parallel(actions, max_workers=4) + +Start the PySide6 GUI: + +.. code-block:: bash + + python -m automation_file ui + +Expose the registry over a loopback HTTP server with shared-secret auth: + +.. code-block:: python + + from automation_file import start_http_action_server + + server = start_http_action_server(port=8765, shared_secret="s3kret") + +Action-list shape +----------------- + +An action is a list of one of three shapes, dispatched by name through the +registry: + +.. code-block:: python + + ["FA_create_dir"] # no-args + ["FA_create_dir", {"dir_path": "build"}] # keyword args + ["FA_copy_file", ["src.txt", "dst.txt"]] # positional args + +A JSON action list is simply a list of these lists. + .. toctree:: :maxdepth: 2 :caption: Contents diff --git a/tests/.pylintrc b/tests/.pylintrc new file mode 100644 index 0000000..0f68e8a --- /dev/null +++ b/tests/.pylintrc @@ -0,0 +1,52 @@ +# Pylint configuration for the test suite. +# +# Pytest-style fixtures intentionally shadow outer names, tests poke at +# protected members to verify invariants, and many assertion patterns use +# explicit equality for readability. These idioms are standard for a +# pytest codebase — disable the rules here rather than peppering every +# test file with suppression comments. + +[MAIN] +extension-pkg-allow-list=PySide6,PySide6.QtCore,PySide6.QtGui,PySide6.QtWidgets + +[MESSAGES CONTROL] +# See project root .pylintrc for non-test disables. +# Additional disables for tests: +# C1803 — use-implicit-booleaness-not-comparison; explicit `== []` / `== {}` +# is clearer in assertions. +# R0801 — duplicate-code; test fixtures and table-driven cases share shape. +# R1732 — consider-using-with; tests often keep resources open across asserts. +# W0212 — protected-access; tests verify private invariants. +# W0613 — unused-argument; pytest fixtures are requested by name regardless +# of whether they're read inside the test body. +# W0621 — redefined-outer-name; pytest fixture parameters shadow the +# fixture function intentionally. +disable= + C0114, + C0115, + C0116, + C0415, + C1803, + R0801, + R0903, + R1732, + W0212, + W0511, + W0613, + W0621 + +[TYPECHECK] +generated-members=files,permissions + +[DESIGN] +# Tests can legitimately have more locals/statements than prod code. +max-args=10 +max-locals=20 +max-returns=8 +max-branches=20 +max-statements=75 +max-attributes=20 +max-public-methods=30 + +[FORMAT] +max-line-length=100 diff --git a/tests/test_action_executor.py b/tests/test_action_executor.py index 03df1a7..e717017 100644 --- a/tests/test_action_executor.py +++ b/tests/test_action_executor.py @@ -46,9 +46,13 @@ def test_execute_action_unknown_records_error() -> None: assert "unknown action" in value +def _raise_runtime_error() -> None: + raise RuntimeError("nope") + + def test_execute_action_runtime_error_is_caught() -> None: executor = _fresh_executor() - executor.registry.register("boom", lambda: (_ for _ in ()).throw(RuntimeError("nope"))) + executor.registry.register("boom", _raise_runtime_error) results = executor.execute_action([["boom"]]) [value] = results.values() assert "RuntimeError" in value diff --git a/tests/test_audit.py b/tests/test_audit.py index 756c372..ede0676 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -27,7 +27,7 @@ def test_record_and_recent_roundtrip(tmp_path: Path) -> None: assert row["payload"] == {"src": "a", "dst": "b"} assert row["result"] == {"ok": True} assert row["error"] is None - assert row["duration_ms"] == 12.5 + assert row["duration_ms"] == pytest.approx(12.5) def test_record_error_stores_repr(tmp_path: Path) -> None: diff --git a/tests/test_config.py b/tests/test_config.py index b5b8fc4..a42f297 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -97,7 +97,7 @@ def test_apply_to_registers_and_sets_dedup(tmp_path: Path, monkeypatch: pytest.M manager = NotificationManager(dedup_seconds=0.0) count = config.apply_to(manager) assert count == 1 - assert manager.dedup_seconds == 120.5 + assert manager.dedup_seconds == pytest.approx(120.5) descriptions = manager.list() assert descriptions[0]["name"] == "team" diff --git a/tests/test_notify.py b/tests/test_notify.py index 80ff482..7ef94ba 100644 --- a/tests/test_notify.py +++ b/tests/test_notify.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Iterator from unittest.mock import MagicMock, patch import pytest @@ -19,7 +20,7 @@ @pytest.fixture(autouse=True) -def _reset_manager() -> None: +def _reset_manager() -> Iterator[None]: notification_manager.unregister_all() yield notification_manager.unregister_all() @@ -38,7 +39,7 @@ def test_webhook_posts_json() -> None: post.assert_called_once() kwargs = post.call_args.kwargs assert kwargs["allow_redirects"] is False - assert kwargs["timeout"] == 10.0 + assert kwargs["timeout"] == pytest.approx(10.0) import json body = json.loads(kwargs["data"].decode("utf-8")) @@ -88,7 +89,7 @@ def test_email_send_uses_smtp_context_manager() -> None: sender="me@example.com", recipients=["you@example.com"], username="me", - password="pw", + password="pw", # NOSONAR test fixture — not a real credential use_tls=True, ) with patch("automation_file.notify.sinks.smtplib.SMTP") as smtp_cls: @@ -109,7 +110,7 @@ def test_email_repr_hides_password() -> None: sender="me@example.com", recipients=["you@example.com"], username="me", - password="secret-pw", + password="secret-pw", # NOSONAR test fixture — repr-hiding assertion ) assert "secret-pw" not in repr(sink) @@ -199,7 +200,7 @@ class _Recorder(NotificationSink): name = "rec" def send(self, subject: str, body: str, level: str = "info") -> None: - pass + """No-op sink — the test only checks manager register/unregister.""" manager = NotificationManager(dedup_seconds=0.0) manager.register(_Recorder()) diff --git a/tests/test_quota.py b/tests/test_quota.py index 57bcf4b..900523b 100644 --- a/tests/test_quota.py +++ b/tests/test_quota.py @@ -25,7 +25,7 @@ def test_check_size_zero_disables_cap() -> None: def test_time_budget_passes_fast_block() -> None: with Quota(max_seconds=1.0).time_budget("fast"): - pass + _ = 1 + 1 # no-op — this test only asserts the context manager doesn't raise def test_time_budget_fails_slow_block() -> None: diff --git a/tests/test_secrets.py b/tests/test_secrets.py index 3253ad4..4ed49df 100644 --- a/tests/test_secrets.py +++ b/tests/test_secrets.py @@ -62,7 +62,12 @@ def test_resolve_secret_refs_walks_nested_containers( { "notify": { "sinks": [ - {"username": "${env:U}", "password": "${env:P}", "port": 587}, + { + "username": "${env:U}", + # NOSONAR env-ref placeholder, not a real credential + "password": "${env:P}", # NOSONAR + "port": 587, + }, ] } },