Address SonarCloud open issues

Reduce cognitive complexity (S3776) in download_file, rotate_backups, the
tar-member verifier, json_edit set/delete walkers, run_shell, execute_action_dag,
_group_by_size, scandir_find, and the cron field parser by extracting helpers
with single responsibilities. Behaviour is unchanged; the full test suite still
passes.

Collapse redundant except tuples (S5713) where the listed subclass already
inherits from another entry (shutil.Error/OSError, PermissionError/OSError,
ModuleNotFoundError/ImportError, ManifestException/FileAutomationException).

Extract the duplicated 'application/json' content-type to a module-level
_JSON_HEADERS constant in notify/sinks.py (S1192). Switch the substitution regex
to the concise '\w' class with re.ASCII to preserve ASCII-only matching (S6353).
Drop the unused 'source' parameter from _delete_extras (S1172). Replace tuple
float equality in three tests with pytest.approx (S1244) and mark the obvious
test-fixture credentials with NOSONAR (S2068). Clean up test_notify.py's
Iterator annotation (S5886), the empty _Recorder.send body (S1186), the empty
time_budget block in test_quota.py (S108), and the generator-throw trick in
test_action_executor.py (S7500).

Author: JE-Chen

automation_file/core/dag_executor.py

@@ -31,6 +31,68 @@
 __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 _skip_dependents(self, node_id: str) -> None:
+        for dependent in self.graph.get(node_id, ()):
+            self.indegree[dependent] -= 1
+            _mark_skipped(dependent, node_id, self.graph, self.indegree, self.results, self.lock)
+
+    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:
+                _mark_skipped(
+                    dependent, node_id, self.graph, self.indegree, self.results, self.lock
+                )
+            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 +108,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:

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": [],

automation_file/core/package_loader.py

@@ -36,7 +36,7 @@ def load(self, package: str) -> ModuleType | None:
             # `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:
+        except ImportError as error:
             file_automation_logger.error("PackageLoader import error: %r", error)
             return None
         self._cache[package] = module

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

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
 

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
 

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

automation_file/local/shell_ops.py

@@ -25,40 +25,26 @@ 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)
     if not argv_list or not all(isinstance(part, str) for part in argv_list):
         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(
             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 "",