Automate post-import refresh and validation workflow

reset_helpdesk_mail_settings.py

@@ -7,20 +7,17 @@ settings so test mail flows through Mailpit and imported real credentials cannot
 be used accidentally.
 """
 
-from __future__ import annotations
-
 import argparse
 import json
 import os
 import subprocess
 import sys
-from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 
 DEFAULT_SSH_HOST = "reddev@192.168.50.170"
-DEFAULT_SSH_KEY = Path("/tmp/reddev")
+DEFAULT_SSH_KEY = Path("/home/iadnah/reddev")
 DEFAULT_REMOTE_REDMINE = "/usr/share/redmine"
 DEFAULT_MAILPIT_HOST = "192.168.1.105"
 
@@ -55,15 +52,16 @@ class ResetError(RuntimeError):
     pass
 
 
-@dataclass(frozen=True)
 class RemoteRedmine:
-    ssh_host: str
-    ssh_key: Path
-    remote_redmine: str
+    def __init__(self, ssh_host, ssh_key, remote_redmine, local=False):
+        self.ssh_host = ssh_host
+        self.ssh_key = ssh_key
+        self.remote_redmine = remote_redmine
+        self.local = local
 
-    def mysql_json_lines(self, sql: str) -> list[dict[str, Any]]:
+    def mysql_json_lines(self, sql):
         stdout = self.mysql(sql)
-        rows: list[dict[str, Any]] = []
+        rows = []
         for line in stdout.splitlines():
             if not line.strip():
                 continue
@@ -73,24 +71,29 @@ class RemoteRedmine:
                 raise ResetError(f"Remote query returned an unexpected row: {line[:200]}") from exc
         return rows
 
-    def mysql(self, sql: str) -> str:
-        command = [
-            "ssh",
-            "-i",
-            str(self.ssh_key),
-            "-o",
-            "IdentitiesOnly=yes",
-            self.ssh_host,
-            self._mysql_runner_command(),
-        ]
+    def mysql(self, sql):
+        command = self._mysql_runner_command()
+        shell = True
+        if not self.local:
+            command = [
+                "ssh",
+                "-i",
+                str(self.ssh_key),
+                "-o",
+                "IdentitiesOnly=yes",
+                self.ssh_host,
+                self._mysql_runner_command(),
+            ]
+            shell = False
         try:
             result = subprocess.run(
                 command,
                 input=sql,
-                text=True,
+                universal_newlines=True,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 check=False,
+                shell=shell,
             )
         except OSError as exc:
             raise ResetError(f"Could not run ssh: {exc}") from exc
@@ -99,7 +102,7 @@ class RemoteRedmine:
             raise ResetError(result.stderr.strip() or "Remote MySQL command failed.")
         return result.stdout
 
-    def _mysql_runner_command(self) -> str:
+    def _mysql_runner_command(self):
         ruby = (
             "require 'yaml'; "
             "c = YAML.load_file('config/database.yml')['production']; "
@@ -112,12 +115,13 @@ class RemoteRedmine:
         return f"cd {shell_quote(self.remote_redmine)} && ruby -e {shell_quote(ruby)}"
 
 
-def main() -> int:
+def main():
     parser = argparse.ArgumentParser(
         description="Reset Helpdesk mail settings for all active projects."
     )
     parser.add_argument("--ssh-host", default=os.getenv("REDMINE_SSH_HOST", DEFAULT_SSH_HOST))
     parser.add_argument("--ssh-key", type=Path, default=Path(os.getenv("REDMINE_SSH_KEY", str(DEFAULT_SSH_KEY))))
+    parser.add_argument("--local", action="store_true", help="Read the Redmine database locally instead of over SSH.")
     parser.add_argument("--remote-redmine", default=os.getenv("REDMINE_REMOTE_PATH", DEFAULT_REMOTE_REDMINE))
     parser.add_argument("--mailpit-host", default=DEFAULT_MAILPIT_HOST, help="Host Redmine should use to reach Mailpit.")
     parser.add_argument("--pop3-port", type=int, default=1110)
@@ -139,7 +143,7 @@ def main() -> int:
     parser.add_argument("--dry-run", action="store_true", help="Show affected projects and settings without writing.")
     args = parser.parse_args()
 
-    remote = RemoteRedmine(args.ssh_host, args.ssh_key, args.remote_redmine)
+    remote = RemoteRedmine(args.ssh_host, args.ssh_key, args.remote_redmine, local=args.local)
 
     try:
         projects = find_active_projects(remote, args.project)
@@ -166,7 +170,7 @@ def main() -> int:
         return 1
 
 
-def find_active_projects(remote: RemoteRedmine, filters: list[str]) -> list[dict[str, Any]]:
+def find_active_projects(remote, filters):
     where = ["p.status = 1"]
     if filters:
         clauses = []
@@ -190,8 +194,8 @@ ORDER BY p.identifier;
     )
 
 
-def build_values(args: argparse.Namespace, projects: list[dict[str, Any]]) -> list[tuple[int, str, str]]:
-    rows: list[tuple[int, str, str]] = []
+def build_values(args, projects):
+    rows = []
     for project in projects:
         project_id = int(project["id"])
         answer_from = args.from_pattern.format(
@@ -227,7 +231,7 @@ def build_values(args: argparse.Namespace, projects: list[dict[str, Any]]) -> li
     return rows
 
 
-def apply_values(remote: RemoteRedmine, rows: list[tuple[int, str, str]]) -> None:
+def apply_values(remote, rows):
     statements = ["START TRANSACTION;"]
     for project_id, name, value in rows:
         project_id_sql = sql_int(project_id)
@@ -254,8 +258,8 @@ WHERE NOT EXISTS (
     remote.mysql("\n".join(statements))
 
 
-def print_plan(rows: list[tuple[int, str, str]]) -> None:
-    current_project_id: int | None = None
+def print_plan(rows):
+    current_project_id = None
     for project_id, name, value in rows:
         if project_id != current_project_id:
             current_project_id = project_id
@@ -264,18 +268,18 @@ def print_plan(rows: list[tuple[int, str, str]]) -> None:
         print(f"  {name} = {display_value}")
 
 
-def sql_int(value: Any) -> int:
+def sql_int(value):
     try:
         return max(0, int(value))
     except (TypeError, ValueError):
         return 0
 
 
-def sql_string(value: Any) -> str:
+def sql_string(value):
     return "'" + str(value).replace("\\", "\\\\").replace("'", "\\'") + "'"
 
 
-def shell_quote(value: str) -> str:
+def shell_quote(value):
     return "'" + value.replace("'", "'\"'\"'") + "'"