From 5f49605d24174d74bb752285ea9240545845321e Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 11 Sep 2026 13:19:37 +0800 Subject: [PATCH] fix: address CUPS runtime review findings and add regression tests --- .gitignore | 3 + README.md | 23 ++ cups-driver-manager/driver_manager.py | 47 +++- setup_cups.sh | 74 ++--- tests/test_regressions.py | 373 ++++++++++++++++++++++++++ watchdog/network-watchdog.sh | 4 +- watchdog/print-watchdog.sh | 4 +- 7 files changed, 490 insertions(+), 38 deletions(-) create mode 100644 .gitignore create mode 100644 tests/test_regressions.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e3aed4d --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.py[cod] +.test-work/ diff --git a/README.md b/README.md index 22e85ea..be8be1f 100644 --- a/README.md +++ b/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 打印机选项 diff --git a/cups-driver-manager/driver_manager.py b/cups-driver-manager/driver_manager.py index 6a4f886..1e43e53 100644 --- a/cups-driver-manager/driver_manager.py +++ b/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: - results.extend(install_deb(str(deb_file))) - return results + return install_deb_bundle(deb_files) 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'): diff --git a/setup_cups.sh b/setup_cups.sh index 00951c7..31184ee 100755 --- a/setup_cups.sh +++ b/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 "强制结束自动更新进程..." - killall unattended-upgr 2>/dev/null || true - sleep 2 + warn "等待自动更新超时，请稍后重新运行安装脚本" + return 1 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 - # 清理可能残留的锁文件 - 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 - + # 锁由内核随进程释放；删除锁文件会破坏其他 apt/dpkg 进程的互斥。 # 修复可能中断的安装 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/dev/null | grep -q "PDF"; then - # 尝试使用通用 PPD - lpadmin -p PDF \ + -o printer-is-shared=true; then + # 从 CUPS 实际提供的模型中选备用 PPD，不创建无法转换 PDF 的 raw 队列。 + local pdf_model + pdf_model=$(lpinfo -m 2>/dev/null | awk 'tolower($0) ~ /cups-pdf/ && $1 != "lsb/usr/cups-pdf/CUPS-PDF_opt.ppd" {print $1; exit}') + if [ -z "$pdf_model" ] || ! 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 "" diff --git a/tests/test_regressions.py b/tests/test_regressions.py new file mode 100644 index 0000000..67b33f5 --- /dev/null +++ b/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() diff --git a/watchdog/network-watchdog.sh b/watchdog/network-watchdog.sh index 420f308..e05d3e7 100644 --- a/watchdog/network-watchdog.sh +++ b/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() { diff --git a/watchdog/print-watchdog.sh b/watchdog/print-watchdog.sh index f5fb6a7..333e81f 100644 --- a/watchdog/print-watchdog.sh +++ b/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() {