fix: address CUPS runtime review findings and add regression tests

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.py[cod]
+.test-work/

README.md

@@ -181,7 +181,30 @@ lpinfo -v | grep usb
 - Gitea: https://git.workyai.cn/237899745/S905L3A
 - Gitee: https://gitee.com/yu-yon/S905L3A（镜像）
 
+## 回归测试
+
+需要 Python 3.9+、Flask 和 Bash。在仓库根目录运行：
+
+```bash
+python3 -B -m unittest discover -s tests -v
+```
+
+测试会模拟系统命令和驱动安装，不会修改真实网络、安装驱动或重启服务。
+覆盖守护锁释放与连续失败恢复、多网卡地址选择、APT 等待/超时、PDF 备用驱动、DEB 架构筛选和基础访问控制。
+通过这些测试不等于通过实机验证；USB 打印、AirPrint、NetworkManager/netplan 切换仍需在目标设备验证。
+
+## 已部署设备升级提示
+
+更新 Git 仓库不会自动更新 `/opt` 下正在运行的文件。已安装设备需要部署更新后的
+`watchdog/network-watchdog.sh`、`watchdog/print-watchdog.sh` 到 `/opt/cups-watchdog/`，
+以及 `cups-driver-manager/driver_manager.py` 到 `/opt/cups-driver-manager/`，保留原有配置和账号密码。
+
+旧版守护脚本可能在 `/run/cups-watchdog/` 留下 `network.lock`、`print.lock` 目录。
+部署新文件后重启设备一次，可清除 `/run` 中的旧锁并加载新代码。
+不要在守护脚本仍运行时直接删除锁目录，以免破坏并发互斥。
+
 ## 更新日志
 
+- 2026-09-11: 修复守护锁、多网卡静态 IP、APT 等待、PDF 备用驱动和多架构 DEB 安装问题，添加隔离回归测试。
 - 2024-12-01: 添加卸载功能、优化打印速度
 - 2024-11-30: 修复 CSS 兼容 CUPS 2.4.7、添加 PDF 打印机选项

cups-driver-manager/driver_manager.py

@@ -216,6 +216,47 @@ def install_deb(filepath):
 
     return results
 
+def install_deb_bundle(filepaths):
+    """先检查包元数据，仅安装本机架构和架构无关的 DEB。"""
+    host = run_command(['dpkg', '--print-architecture'])
+    architecture = host['stdout'].strip()
+    if not host['success'] or not architecture:
+        return [('检测系统架构', {
+            **host, 'success': False, 'returncode': host['returncode'] or 1,
+            'stderr': host['stderr'] or '无法检测 dpkg 系统架构，未安装任何驱动包'
+        })]
+
+    results = []
+    compatible = []
+    # 完成全部检查后再安装，避免读取损坏的包时已经修改了系统。
+    for filepath in filepaths:
+        metadata = run_command(['dpkg-deb', '--field', str(filepath), 'Architecture'])
+        package_architecture = metadata['stdout'].strip()
+        if not metadata['success'] or not package_architecture:
+            results.append((f'检查 DEB 架构: {Path(filepath).name}', {
+                **metadata, 'success': False, 'returncode': metadata['returncode'] or 1,
+                'stderr': metadata['stderr'] or '无法读取包的 Architecture 字段，未安装任何驱动包'
+            }))
+            return results
+        if package_architecture in (architecture, 'all'):
+            compatible.append(str(filepath))
+        else:
+            results.append(('跳过其他架构 DEB', {
+                'success': True, 'returncode': 0, 'stderr': '',
+                'stdout': f'{Path(filepath).name}: {package_architecture}，当前系统为 {architecture}'
+            }))
+
+    if not compatible:
+        results.append(('选择兼容 DEB', {
+            'success': False, 'returncode': 1, 'stdout': '',
+            'stderr': f'未找到适用于 {architecture} 或 all 架构的驱动包'
+        }))
+        return results
+
+    for filepath in compatible:
+        results.extend(install_deb(filepath))
+    return results
+
 def install_ppd(filepath):
     """安装 .ppd 文件"""
     results = []
@@ -266,9 +307,7 @@ def install_extracted_dir(extract_dir):
 
     deb_files = find_files_by_suffix(extract_dir, ('.deb',))
     if deb_files:
-        for deb_file in deb_files:
+        return install_deb_bundle(deb_files)
-            results.extend(install_deb(str(deb_file)))
-        return results
 
     rpm_files = find_files_by_suffix(extract_dir, ('.rpm',))
     if rpm_files:
@@ -424,7 +463,7 @@ def install_script(filepath):
 def install_driver(filepath, file_type):
     """根据文件类型安装驱动"""
     if file_type == 'deb':
-        return install_deb(filepath)
+        return install_deb_bundle([filepath])
     elif file_type == 'ppd':
         return install_ppd(filepath)
     elif file_type in ('tar.gz', 'tar', 'tgz'):

setup_cups.sh

@@ -77,16 +77,15 @@ fix_apt_lock() {
         local count=0
         while pgrep -x "unattended-upgr" > /dev/null 2>&1 && [ $count -lt 30 ]; do
             sleep 1
-            ((count++))
+            count=$((count + 1))
             echo -ne "\r  等待自动更新进程结束... ${count}s"
         done
         echo ""
 
-        # 如果还在运行，强制结束
+        # 不强制中断正在写入软件包数据库的进程。
         if pgrep -x "unattended-upgr" > /dev/null 2>&1; then
-            warn "强制结束自动更新进程..."
+            warn "等待自动更新超时，请稍后重新运行安装脚本"
-            killall unattended-upgr 2>/dev/null || true
+            return 1
-            sleep 2
         fi
 
         success "自动更新服务已停止"
@@ -98,17 +97,17 @@ fix_apt_lock() {
         local count=0
         while fuser /var/lib/dpkg/lock-frontend > /dev/null 2>&1 && [ $count -lt 60 ]; do
             sleep 1
-            ((count++))
+            count=$((count + 1))
             echo -ne "\r  等待 apt 锁释放... ${count}s"
         done
         echo ""
+        if fuser /var/lib/dpkg/lock-frontend > /dev/null 2>&1; then
+            warn "等待 apt 锁超时，请稍后重新运行安装脚本"
+            return 1
+        fi
     fi
 
-    # 清理可能残留的锁文件
+    # 锁由内核随进程释放；删除锁文件会破坏其他 apt/dpkg 进程的互斥。
-    rm -f /var/lib/dpkg/lock-frontend 2>/dev/null || true
-    rm -f /var/lib/dpkg/lock 2>/dev/null || true
-    rm -f /var/cache/apt/archives/lock 2>/dev/null || true
-
     # 修复可能中断的安装
     dpkg --configure -a 2>/dev/null || true
 
@@ -244,23 +243,28 @@ EOF
 
 # 获取本机IP地址
 get_ip() {
-    ip addr show | grep -oP '(?<=inet\s)\d+(\.\d+){3}' | grep -v '127.0.0.1' | head -n1
+    local iface="${1:-$(get_interface)}"
+    [ -n "$iface" ] || return 1
+    ip -4 -o addr show dev "$iface" scope global | awk '{split($4, address, "/"); print address[1]; exit}'
 }
 
 # 获取默认网关
 get_gateway() {
-    ip route | grep default | awk '{print $3}' | head -n1
+    local iface="${1:-$(get_interface)}"
+    [ -n "$iface" ] || return 1
+    ip -4 route show default dev "$iface" | awk '{for (i=1; i<NF; i++) if ($i == "via") {print $(i+1); exit}}'
 }
 
 # 获取主网络接口名称
 get_interface() {
-    ip route | grep default | awk '{print $5}' | head -n1
+    ip -4 route show default | awk '{for (i=1; i<NF; i++) if ($i == "dev") {print $(i+1); exit}}'
 }
 
 # 获取子网掩码（CIDR格式）
 get_netmask() {
-    local iface=$(get_interface)
+    local iface="${1:-$(get_interface)}"
-    ip addr show "$iface" | grep -oP 'inet \K[\d.]+/\d+' | head -n1 | cut -d'/' -f2
+    [ -n "$iface" ] || return 1
+    ip -4 -o addr show dev "$iface" scope global | awk '{split($4, address, "/"); print address[2]; exit}'
 }
 
 # 获取DNS服务器
@@ -277,12 +281,13 @@ configure_static_ip() {
     echo -e "${YELLOW}═══════════════════════════════════════════════════════════${NC}"
     echo ""
 
-    local current_ip=$(get_ip)
-    local current_gateway=$(get_gateway)
     local current_interface=$(get_interface)
-    local current_netmask=$(get_netmask)
+    local current_ip=$(get_ip "$current_interface")
+    local current_gateway=$(get_gateway "$current_interface")
+    local current_netmask=$(get_netmask "$current_interface")
     local current_dns=$(get_dns)
 
+    [ -n "$current_interface" ] || error "未找到 IPv4 默认路由对应的网卡，无法设置静态 IP"
     echo -e "${BLUE}当前网络配置:${NC}"
     echo "  接口名称: $current_interface"
     echo "  IP 地址:  $current_ip"
@@ -305,6 +310,9 @@ configure_static_ip() {
     local static_dns=""
 
     if [[ "$use_current" =~ ^[Yy]$ ]]; then
+        if [ -z "$current_ip" ] || [ -z "$current_netmask" ] || [ -z "$current_gateway" ]; then
+            error "当前网卡的 IPv4 地址、掩码或网关不完整，未修改网络配置"
+        fi
         static_ip="$current_ip"
         static_gateway="$current_gateway"
         static_netmask="$current_netmask"
@@ -1368,10 +1376,10 @@ install_watchdogs() {
 
     [ -n "$tmp_dir" ] && rm -rf "$tmp_dir"
 
-    local current_ip=$(get_ip)
-    local current_gateway=$(get_gateway)
     local current_interface=$(get_interface)
-    local current_netmask=$(get_netmask)
+    local current_ip=$(get_ip "$current_interface")
+    local current_gateway=$(get_gateway "$current_interface")
+    local current_netmask=$(get_netmask "$current_interface")
     local current_dns=$(get_dns)
     local current_host_octet="${current_ip##*.}"
     current_dns="${current_dns:-114.114.114.114 223.5.5.5}"
@@ -1451,24 +1459,26 @@ install_pdf_printer() {
     else
         # 添加 PDF 虚拟打印机
         # 使用 lpadmin 命令创建打印机
-        lpadmin -p PDF \
+        if ! lpadmin -p PDF \
             -v cups-pdf:/ \
             -E \
             -m lsb/usr/cups-pdf/CUPS-PDF_opt.ppd \
             -D "虚拟PDF打印机 (测试用)" \
             -L "本地" \
-            -o printer-is-shared=true
+            -o printer-is-shared=true; then
-
+            # 从 CUPS 实际提供的模型中选备用 PPD，不创建无法转换 PDF 的 raw 队列。
-        # 如果上面的 PPD 文件不存在，尝试其他路径
+            local pdf_model
-        if ! lpstat -p PDF 2>/dev/null | grep -q "PDF"; then
+            pdf_model=$(lpinfo -m 2>/dev/null | awk 'tolower($0) ~ /cups-pdf/ && $1 != "lsb/usr/cups-pdf/CUPS-PDF_opt.ppd" {print $1; exit}')
-            # 尝试使用通用 PPD
+            if [ -z "$pdf_model" ] || ! lpadmin -p PDF \
-            lpadmin -p PDF \
                 -v cups-pdf:/ \
                 -E \
-                -m raw \
+                -m "$pdf_model" \
                 -D "虚拟PDF打印机 (测试用)" \
                 -L "本地" \
-                -o printer-is-shared=true 2>/dev/null || true
+                -o printer-is-shared=true; then
+                warn "未能使用可用的 CUPS-PDF 驱动创建队列，跳过虚拟 PDF 打印机"
+                return 0
+            fi
         fi
 
         # 启用打印机
@@ -1596,7 +1606,7 @@ main() {
     show_banner
     check_root
 
-    LOCAL_IP=$(get_ip)
+    LOCAL_IP=$(get_ip) || error "未找到 IPv4 默认路由对应的网卡"
     info "检测到本机IP: ${LOCAL_IP}"
     echo ""
 

tests/test_regressions.py

@@ -0,0 +1,373 @@
+"""Safe regression tests: system commands and driver installation are mocked."""
+
+import base64
+import importlib.util
+import io
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import tarfile
+import tempfile
+import unittest
+from unittest.mock import patch
+import zipfile
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.dont_write_bytecode = True
+
+
+def bash_path():
+    candidate = os.environ.get("BASH_PATH") or shutil.which("bash")
+    if not candidate and os.name == "nt":
+        candidate = "C:/Program Files/Git/bin/bash.exe"
+    if not candidate or not Path(candidate).is_file():
+        raise unittest.SkipTest("Bash is required for shell regression tests")
+    return candidate
+
+
+def shell_path(path):
+    value = str(path.resolve()).replace("\\", "/")
+    if os.name == "nt" and value[1:2] == ":":
+        return "/" + value[0].lower() + value[2:]
+    return value
+
+
+def function(source, name):
+    match = re.search(rf"^{re.escape(name)}\(\) \{{\n.*?^\}}", source, re.M | re.S)
+    if not match:
+        raise AssertionError(f"Shell function not found: {name}")
+    return match.group(0)
+
+
+class IsolatedTest(unittest.TestCase):
+    def setUp(self):
+        workspace = ROOT / ".test-work"
+        workspace.mkdir(exist_ok=True)
+        self.temp = tempfile.TemporaryDirectory(dir=workspace)
+        self.directory = Path(self.temp.name).resolve()
+        self.assertTrue(self.directory.is_relative_to(workspace.resolve()))
+        self.addCleanup(self.temp.cleanup)
+
+
+class ShellTests(IsolatedTest):
+    def setUp(self):
+        super().setUp()
+        self.bash = bash_path()
+        self.setup = (ROOT / "setup_cups.sh").read_text(encoding="utf-8")
+
+    def run_shell(self, source, env=None):
+        return subprocess.run(
+            [self.bash, "--noprofile", "--norc", "-s"], input=source,
+            text=True, encoding="utf-8", errors="replace", capture_output=True,
+            env={**os.environ, **(env or {})}, timeout=15,
+        )
+
+    def test_shell_syntax(self):
+        for path in [ROOT / "setup_cups.sh", *sorted((ROOT / "watchdog").glob("*.sh"))]:
+            with self.subTest(script=path.name):
+                result = subprocess.run(
+                    [self.bash, "-n"], input=path.read_text(encoding="utf-8"),
+                    capture_output=True, text=True, encoding="utf-8", timeout=10,
+                )
+                self.assertEqual(result.returncode, 0, result.stderr)
+
+    def watchdog(self, kind, failed=False):
+        state = self.directory / kind
+        log = self.directory / f"{kind}.log"
+        config = self.directory / f"{kind}.conf"
+        config.write_text(
+            f'STATE_DIR="{shell_path(state)}"\nLOG_FILE="{shell_path(log)}"\n'
+            'SERVICES=""\nCHECK_CUPS_HTTP=0\nCHECK_LPSTAT=0\n', encoding="utf-8",
+        )
+        source = (ROOT / "watchdog" / f"{kind}-watchdog.sh").read_text(encoding="utf-8")
+        preamble = 'ip() { echo "default via 192.168.1.1 dev fixture0"; }; ping() { return 0; };\n'
+        if failed:
+            head, dispatch = source.rsplit('case "${1:-check}" in', 1)
+            recovery = shell_path(self.directory / "recoveries")
+            # Replace only the OS-changing boundary; keep lock/count/threshold logic intact.
+            if kind == "network":
+                overrides = f'connectivity_ok() {{ return 1; }}\nrecover_after_failure() {{ echo recovered >> "{recovery}"; }}\n'
+            else:
+                overrides = f'check_http() {{ return 1; }}\nrestart_print_stack() {{ echo recovered >> "{recovery}"; }}\n'
+            source = head + overrides + 'case "${1:-check}" in' + dispatch
+        return preamble + source, {f"CUPS_{kind.upper()}_WATCHDOG_CONFIG": shell_path(config)}, state
+
+    def test_watchdogs_release_locks_after_every_check(self):
+        for kind in ("network", "print"):
+            with self.subTest(watchdog=kind):
+                source, env, state = self.watchdog(kind)
+                for _ in range(3):
+                    result = self.run_shell(source, env)
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertEqual(result.stderr, "")
+                    self.assertNotIn("another run is active", result.stdout)
+                    self.assertFalse((state / f"{kind}.lock").exists())
+
+    def test_watchdogs_preserve_an_existing_lock(self):
+        for kind in ("network", "print"):
+            with self.subTest(watchdog=kind):
+                source, env, state = self.watchdog(kind)
+                lock = state / f"{kind}.lock"
+                lock.mkdir(parents=True)
+                result = self.run_shell(source, env)
+                self.assertEqual(result.returncode, 0)
+                self.assertIn("another run is active", result.stdout)
+                self.assertTrue(lock.is_dir())
+
+    def test_watchdogs_release_their_lock_on_termination(self):
+        for kind in ("network", "print"):
+            with self.subTest(watchdog=kind):
+                state = self.directory / kind
+                state.mkdir()
+                source = (ROOT / "watchdog" / f"{kind}-watchdog.sh").read_text(encoding="utf-8")
+                script = f'set -u\nSTATE_DIR="{shell_path(state)}"\n'
+                script += function(source, "acquire_lock") + '\nacquire_lock\nkill -TERM "$$"\n'
+                result = self.run_shell(script)
+                self.assertEqual(result.returncode, 143, result.stderr)
+                self.assertEqual(result.stderr, "")
+                self.assertFalse((state / f"{kind}.lock").exists())
+
+    def test_watchdogs_recover_after_consecutive_failures(self):
+        for kind, threshold in (("network", 3), ("print", 2)):
+            with self.subTest(watchdog=kind):
+                source, env, state = self.watchdog(kind, failed=True)
+                count_before = (self.directory / "recoveries").read_text().count("recovered") if (self.directory / "recoveries").exists() else 0
+                for _ in range(threshold):
+                    result = self.run_shell(source, env)
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertEqual(result.stderr, "")
+                self.assertEqual((state / f"{kind}.fail_count").read_text().strip(), "0")
+                self.assertEqual((self.directory / "recoveries").read_text().count("recovered"), count_before + 1)
+
+    def apt_fixture(self, busy_tool, release_after):
+        return (
+            'set -e\ninfo() { :; }; warn() { :; }; success() { :; };\n'
+            'pgrep() { return 1; }; fuser() { return 1; }; systemctl() { :; };\n'
+            'rm() { echo UNEXPECTED_RM; exit 99; }; killall() { echo UNEXPECTED_KILL; exit 99; };\n'
+            'dpkg() { echo DPKG_REACHED; }; sleeps=0; polls=0;\n'
+            'trap \'echo "WAITED=$sleeps"\' EXIT\n'
+            'sleep() { sleeps=$((sleeps + 1)); };\n'
+            f'{busy_tool}() {{ polls=$((polls + 1)); [ "$polls" -le {release_after} ]; }}\n'
+            + function(self.setup, "fix_apt_lock")
+            + '\nfix_apt_lock\nprintf "DONE sleeps=%s\\n" "$sleeps"\n'
+        )
+
+    def test_apt_waits_then_continues(self):
+        for tool in ("fuser", "pgrep"):
+            with self.subTest(tool=tool):
+                result = self.run_shell(self.apt_fixture(tool, 3))
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertIn("DPKG_REACHED", result.stdout)
+                self.assertIn("DONE sleeps=2", result.stdout)
+                self.assertNotIn("UNEXPECTED", result.stdout)
+
+    def test_apt_wait_timeout_does_not_remove_locks_or_kill(self):
+        for tool in ("fuser", "pgrep"):
+            with self.subTest(tool=tool):
+                result = self.run_shell(self.apt_fixture(tool, 1000))
+                self.assertEqual(result.returncode, 1, result.stderr)
+                self.assertNotIn("DPKG_REACHED", result.stdout)
+                self.assertNotIn("UNEXPECTED", result.stdout)
+                self.assertIn("WAITED=60" if tool == "fuser" else "WAITED=30", result.stdout)
+
+    def ip_fixture(self, missing=False):
+        preamble = '''ip() {
+    case "$*" in
+        "-4 route show default"|"-4 route show default dev eth0"|"route")
+            echo 'default via 192.168.1.1 dev eth0 proto dhcp src 192.168.1.219 metric 100' ;;
+        "-4 route show default dev wlan0") echo 'default via 10.0.0.1 dev wlan0' ;;
+        "-4 -o addr show dev eth0 scope global"|"addr show eth0")
+            echo '3: eth0 inet 192.168.1.219/24 brd 192.168.1.255 scope global eth0' ;;
+        "-4 -o addr show dev wlan0 scope global") echo '4: wlan0 inet 10.0.0.20/16 scope global wlan0' ;;
+        "addr show") printf '2: docker0\n    inet 172.17.0.1/16 scope global docker0\n3: eth0\n    inet 192.168.1.219/24 scope global eth0\n' ;;
+        *) return 1 ;;
+    esac
+}
+'''
+        if missing:
+            preamble = "ip() { return 0; }\n"
+        return preamble + "\n".join(function(self.setup, name) for name in ("get_ip", "get_gateway", "get_interface", "get_netmask"))
+
+    def test_default_interface_address_is_not_docker_address(self):
+        result = self.run_shell(self.ip_fixture() + '\nprintf "%s %s %s %s" "$(get_interface)" "$(get_ip)" "$(get_netmask)" "$(get_gateway)"\n')
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "eth0 192.168.1.219 24 192.168.1.1")
+
+    def test_explicit_interface_is_used_for_all_parameters(self):
+        result = self.run_shell(self.ip_fixture() + '\nprintf "%s %s %s" "$(get_ip wlan0)" "$(get_netmask wlan0)" "$(get_gateway wlan0)"\n')
+        self.assertEqual(result.stdout, "10.0.0.20 16 10.0.0.1")
+
+    def test_missing_route_does_not_use_an_unrelated_address(self):
+        for name in ("get_ip", "get_gateway", "get_netmask"):
+            with self.subTest(helper=name):
+                result = self.run_shell(self.ip_fixture(missing=True) + f"\n{name}\n")
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(result.stdout, "")
+
+    def pdf_fixture(self, mode):
+        source = function(self.setup, "install_pdf_printer").replace(
+            "/etc/cups/cups-pdf.conf", shell_path(self.directory / "absent.conf"))
+        return f'MODE={mode}\n' + '''set -e
+info() { :; }; warn() { echo WARNING; }; success() { :; }; sleep() { :; }; apt() { :; }
+calls=0; ready=0
+if [ "$MODE" = existing ]; then ready=1; fi
+lpstat() { if [ "$ready" = 1 ]; then echo 'printer PDF is idle'; else return 1; fi; }
+lpinfo() { if [ "$MODE" != missing ]; then echo 'generic/CUPS-PDF.ppd Generic CUPS-PDF Printer'; fi; }
+lpadmin() {
+    calls=$((calls + 1))
+    echo "LPADMIN $*"
+    if [ "$MODE" = primary ] || { [ "$MODE" = fallback ] && [ "$calls" = 2 ]; }; then ready=1; return 0; fi
+    return 1
+}
+cupsenable() { :; }; cupsaccept() { :; }
+''' + source + '\ninstall_pdf_printer\necho "DONE calls=$calls ready=$ready"\n'
+
+    def test_pdf_primary_model(self):
+        result = self.run_shell(self.pdf_fixture("primary"))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("DONE calls=1 ready=1", result.stdout)
+
+    def test_pdf_existing_queue_is_not_recreated(self):
+        result = self.run_shell(self.pdf_fixture("existing"))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("DONE calls=0 ready=1", result.stdout)
+
+    def test_pdf_fallback_uses_discovered_model(self):
+        result = self.run_shell(self.pdf_fixture("fallback"))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("-m generic/CUPS-PDF.ppd", result.stdout)
+        self.assertIn("DONE calls=2 ready=1", result.stdout)
+        self.assertNotIn("-m raw", result.stdout)
+
+    def test_pdf_unavailable_models_do_not_abort_remaining_setup(self):
+        for mode, calls in (("missing", 1), ("broken", 2)):
+            with self.subTest(mode=mode):
+                result = self.run_shell(self.pdf_fixture(mode))
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertIn(f"DONE calls={calls} ready=0", result.stdout)
+                self.assertIn("WARNING", result.stdout)
+                self.assertNotIn("-m raw", result.stdout)
+
+
+class DriverManagerTests(IsolatedTest):
+    @classmethod
+    def setUpClass(cls):
+        spec = importlib.util.spec_from_file_location("driver_manager", ROOT / "cups-driver-manager" / "driver_manager.py")
+        cls.dm = importlib.util.module_from_spec(spec)
+        with patch("os.makedirs"):
+            spec.loader.exec_module(cls.dm)
+
+    def setUp(self):
+        super().setUp()
+        self.uploads = self.directory / "uploads"
+        self.uploads.mkdir()
+        self.dm.app.config.update(TESTING=True, UPLOAD_FOLDER=str(self.uploads))
+        self.client = self.dm.app.test_client()
+        token = base64.b64encode(f"{self.dm.ADMIN_USERNAME}:{self.dm.ADMIN_PASSWORD}".encode()).decode()
+        self.headers = {"Authorization": "Basic " + token}
+        self.architectures = {"a-amd64.deb": "arm64", "b-arm64.deb": "amd64", "common.deb": "all"}
+
+    @staticmethod
+    def result(success=True, stdout="", stderr=""):
+        return {"success": success, "stdout": stdout, "stderr": stderr, "returncode": 0 if success else 1}
+
+    def command(self, cmd, **kwargs):
+        if cmd == ['dpkg', '--print-architecture']:
+            return self.result(stdout="arm64\n")
+        if cmd[:2] == ['dpkg-deb', '--field'] and cmd[-1] == 'Architecture':
+            architecture = self.architectures[Path(cmd[2]).name]
+            return self.result(stdout=architecture + "\n") if architecture else self.result(False, stderr="Invalid DEB")
+        raise AssertionError(f"Unexpected system command: {cmd}")
+
+    def archive_request(self, names):
+        buffer = io.BytesIO()
+        with zipfile.ZipFile(buffer, "w") as archive:
+            for name in names:
+                archive.writestr("vendor/" + name, b"mock DEB metadata is supplied by the test")
+        buffer.seek(0)
+        installed = []
+
+        def install(path):
+            installed.append(Path(path).name)
+            return [("mock installation", self.result())]
+
+        real_mkdtemp = tempfile.mkdtemp
+        with patch.object(self.dm, "run_command", side_effect=self.command), \
+             patch.object(self.dm, "install_deb", side_effect=install), \
+             patch.object(self.dm.tempfile, "mkdtemp", side_effect=lambda prefix: real_mkdtemp(prefix=prefix, dir=self.directory)):
+            response = self.client.post("/api/install", headers=self.headers,
+                                        data={"driver_file": (buffer, "drivers.zip")})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(list(self.uploads.iterdir()), [])
+        return response.json, installed
+
+    def test_archive_uses_metadata_not_filename_and_includes_all(self):
+        response, installed = self.archive_request(self.architectures)
+        self.assertTrue(response['success'], response)
+        self.assertEqual(installed, ["a-amd64.deb", "common.deb"])
+
+    def test_archive_with_no_compatible_packages_fails_without_installation(self):
+        response, installed = self.archive_request(["b-arm64.deb"])
+        self.assertFalse(response['success'])
+        self.assertEqual(installed, [])
+
+    def test_architecture_independent_bundle_is_supported(self):
+        response, installed = self.archive_request(["common.deb"])
+        self.assertTrue(response['success'], response)
+        self.assertEqual(installed, ["common.deb"])
+
+    def test_corrupt_metadata_prevents_partial_installation(self):
+        self.architectures["common.deb"] = None
+        response, installed = self.archive_request(["a-amd64.deb", "common.deb"])
+        self.assertFalse(response['success'])
+        self.assertEqual(installed, [])
+
+    def test_architecture_detection_failure_does_not_install(self):
+        with patch.object(self.dm, "run_command", return_value=self.result(False, stderr="dpkg unavailable")), \
+             patch.object(self.dm, "install_deb") as install:
+            results = self.dm.install_driver("driver.deb", "deb")
+        self.assertFalse(all(step[1]['success'] for step in results))
+        install.assert_not_called()
+
+    def test_direct_foreign_deb_is_rejected_before_dependency_repair(self):
+        with patch.object(self.dm, "run_command", side_effect=self.command), \
+             patch.object(self.dm, "install_deb") as install:
+            results = self.dm.install_driver("b-arm64.deb", "deb")
+        self.assertFalse(all(step[1]['success'] for step in results))
+        install.assert_not_called()
+
+    def test_deb_dependency_repair_still_retries(self):
+        with patch.object(self.dm, "run_command", side_effect=[self.result(False), self.result(), self.result()]) as command:
+            results = self.dm.install_deb("native.deb")
+        self.assertTrue(all(step[1]['success'] for step in results))
+        self.assertEqual(command.call_args_list[-1].args[0], ['dpkg', '-i', 'native.deb'])
+
+    def test_auth_and_private_network_filter(self):
+        self.assertEqual(self.client.get("/").status_code, 401)
+        self.assertEqual(self.client.get("/", headers=self.headers).status_code, 200)
+        self.assertEqual(self.client.get("/", headers=self.headers, environ_overrides={"REMOTE_ADDR": "8.8.8.8"}).status_code, 403)
+
+    def test_invalid_upload_is_rejected_without_dispatch(self):
+        with patch.object(self.dm, "install_driver") as install:
+            response = self.client.post("/api/install", headers=self.headers,
+                                        data={"driver_file": (io.BytesIO(b"test"), "bad.txt")})
+        self.assertFalse(response.json['success'])
+        install.assert_not_called()
+
+    def test_tar_traversal_is_rejected(self):
+        buffer = io.BytesIO()
+        with tarfile.open(fileobj=buffer, mode="w") as archive:
+            entry = tarfile.TarInfo("../escape.txt")
+            entry.size = 1
+            archive.addfile(entry, io.BytesIO(b"x"))
+        buffer.seek(0)
+        with tarfile.open(fileobj=buffer) as archive, self.assertRaises(ValueError):
+            self.dm.safe_extract_tar(archive, str(self.directory / "extract"))
+
+
+if __name__ == "__main__":
+    unittest.main()

watchdog/network-watchdog.sh

@@ -46,7 +46,9 @@ acquire_lock() {
         echo "$(date '+%Y-%m-%d %H:%M:%S') [network-watchdog] another run is active"
         exit 0
     fi
-    trap 'rmdir "$lock_dir" 2>/dev/null || true' EXIT
+    trap 'rmdir "$STATE_DIR/network.lock" 2>/dev/null || true' EXIT
+    trap 'exit 130' INT
+    trap 'exit 143' TERM
 }
 
 rotate_log() {

watchdog/print-watchdog.sh

@@ -35,7 +35,9 @@ acquire_lock() {
         echo "$(date '+%Y-%m-%d %H:%M:%S') [print-watchdog] another run is active"
         exit 0
     fi
-    trap 'rmdir "$lock_dir" 2>/dev/null || true' EXIT
+    trap 'rmdir "$STATE_DIR/print.lock" 2>/dev/null || true' EXIT
+    trap 'exit 130' INT
+    trap 'exit 143' TERM
 }
 
 rotate_log() {