374 lines
18 KiB
Python
374 lines
18 KiB
Python
"""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()
|