mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
276 lines
13 KiB
Python
276 lines
13 KiB
Python
"""The VM wizard must not accept failed ISO downloads as installation media.
|
|
|
|
Run its function logic against a loopback HTTP fixture. Redirect the fixed
|
|
native ISO directory only in the loaded test copy and stub Proxmox/UI services.
|
|
No product script is sourced, no guest operation runs, and no public network
|
|
or real ISO storage is accessed.
|
|
"""
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
CREATOR = ROOT / "scripts/vm/vm_creator.sh"
|
|
PAYLOAD = b"fixture installation media\x00\x01\n"
|
|
REAL_WGET = shutil.which("wget")
|
|
|
|
|
|
class MediaHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path == "/missing.iso":
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
if self.path == "/partial.iso":
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", "1024")
|
|
self.end_headers()
|
|
self.wfile.write(b"partial")
|
|
self.close_connection = True
|
|
return
|
|
body = b"" if self.path == "/empty.iso" else PAYLOAD
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Content-Disposition", 'attachment; filename="remote-name.iso"')
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
class VmIsoDownloadTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if REAL_WGET is None:
|
|
raise RuntimeError("These offline tests require wget, as the VM wizard does")
|
|
cls.server = ThreadingHTTPServer(("127.0.0.1", 0), MediaHandler)
|
|
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
|
|
cls.thread.start()
|
|
cls.http = f"http://127.0.0.1:{cls.server.server_port}"
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.shutdown()
|
|
cls.server.server_close()
|
|
cls.thread.join()
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="vm-iso-download-")
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.directory = Path(self.temp.name)
|
|
self.iso = self.directory / "installation media.iso"
|
|
self.qm_log = self.directory / "qm.log"
|
|
self.wget_log = self.directory / "wget.log"
|
|
self.bin = self.directory / "bin"
|
|
self.bin.mkdir()
|
|
# Route only the optional fixed VirtIO/SourceForge endpoints to our
|
|
# fixture. Wget still performs the actual HTTP transfer and file I/O.
|
|
assert REAL_WGET is not None
|
|
wrapper = self.bin / "wget"
|
|
wrapper.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
'printf "%s\\n" "$*" >> "$WGET_LOG"\n'
|
|
'args=("$@")\n'
|
|
'for i in "${!args[@]}"; do\n'
|
|
' case "${args[$i]}" in\n'
|
|
' https://fedorapeople.org/*) args[$i]="$VIRTIO_FIXTURE_URL";;\n'
|
|
' https://sourceforge.net/*) args[$i]="$HTTP_FIXTURE/success.iso";;\n'
|
|
' esac\n'
|
|
' if [[ "${args[$i]}" == http://* || "${args[$i]}" == https://* ]]; then\n'
|
|
' [[ "${args[$i]}" == "$HTTP_FIXTURE"/* ]] || exit 98\n'
|
|
' fi\n'
|
|
'done\n'
|
|
'for i in "${!args[@]}"; do\n'
|
|
' if [[ "${args[$i]}" == -O ]]; then\n'
|
|
' case "${args[$((i+1))]}" in "$FIXTURE_DIR"/*) ;; *) exit 97;; esac\n'
|
|
' fi\n'
|
|
'done\n'
|
|
f"exec {shlex.quote(REAL_WGET)} --no-config --no-proxy --tries=1 --timeout=1 "
|
|
'"${args[@]}"\n'
|
|
)
|
|
wrapper.chmod(0o755)
|
|
|
|
def run_wizard(self, route="/success.iso", windows=False, overwrite=True, url=None,
|
|
caller_iso_dir=None):
|
|
source = CREATOR.read_text()
|
|
functions = "\n\n".join(re.findall(r"^function \w+\(\) \{\n.*?^\}", source, re.M | re.S))
|
|
self.assertIn("function create_vm()", functions)
|
|
# Only relocate this external filesystem anchor. Keep all predicates,
|
|
# transfers and promotion logic unchanged, without a production hook.
|
|
self.assertIn("/var/lib/vz/template/iso", functions)
|
|
functions = functions.replace("/var/lib/vz/template/iso", "${FIXTURE_DIR}")
|
|
prelude = r'''
|
|
translate() { printf '%s' "$1"; }
|
|
msg_ok() { printf 'OK:%s\n' "$*"; }
|
|
msg_error() { printf 'ERROR:%s\n' "$*"; }
|
|
msg_warn() { printf 'WARN:%s\n' "$*"; }
|
|
msg_info() { :; }
|
|
msg_success() { :; }
|
|
qm() { printf '%q ' "$@" >> "$QM_LOG"; printf '\n' >> "$QM_LOG"; }
|
|
pvesm() { printf 'Name Type Status\nfixture dir active\n'; }
|
|
select_storage_target() { printf fixture; }
|
|
# No next menu, start, or guest-facing process is launched.
|
|
bash() { :; }
|
|
whiptail() {
|
|
if [[ " $* " == *' --yesno '* ]]; then
|
|
return "$OVERWRITE_STATUS"
|
|
fi
|
|
if [[ ! -f "$DIALOG_ONCE" ]]; then
|
|
: > "$DIALOG_ONCE"
|
|
printf '1' >&2
|
|
return 0
|
|
fi
|
|
return 1
|
|
}
|
|
IMPORT_DISKS=(); PASSTHROUGH_DISKS=(); VIRTUAL_DISKS=(); CONTROLLER_NVME_PCIS=()
|
|
VMID=99999; HN=fixture; CORE_COUNT=1; RAM_SIZE=512; BRG=fixture
|
|
MAC='02:00:00:00:00:01'; VLAN=''; MTU=''; BIOS_TYPE=''; CPU_TYPE=''
|
|
MACHINE=''; SERIAL_PORT=''; START_VM=no; WIZARD_ADD_GPU=no
|
|
TAB=''; GN=''; CL=''
|
|
create_vm
|
|
'''
|
|
env = dict(os.environ, PATH=str(self.bin) + os.pathsep + os.environ["PATH"],
|
|
ISO_PATH=str(self.iso), ISO_URL=url if url is not None else self.http + route,
|
|
ISO_VOLID="local:iso/installation media.iso",
|
|
OS_TYPE="2" if windows else "3",
|
|
QM_LOG=str(self.qm_log), WGET_LOG=str(self.wget_log),
|
|
DIALOG_ONCE=str(self.directory / "dialog-once"), FIXTURE_DIR=str(self.directory),
|
|
OVERWRITE_STATUS="0" if overwrite else "1", HTTP_FIXTURE=self.http,
|
|
VIRTIO_FIXTURE_URL=self.http + route)
|
|
env.pop("ISO_DIR", None)
|
|
if caller_iso_dir is not None:
|
|
env["ISO_DIR"] = str(caller_iso_dir)
|
|
return subprocess.run(["bash", "--noprofile", "--norc", "-c", functions + prelude],
|
|
env=env, stdin=subprocess.DEVNULL, capture_output=True,
|
|
text=True, timeout=15)
|
|
|
|
def qm_calls(self):
|
|
return self.qm_log.read_text() if self.qm_log.exists() else ""
|
|
|
|
def test_http_failure_stops_before_creating_a_guest(self):
|
|
result = self.run_wizard("/missing.iso")
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.qm_calls(), "")
|
|
self.assertNotIn("OK:ISO image downloaded", result.stdout)
|
|
self.assertIn("ERROR:Failed to download ISO image", result.stdout)
|
|
|
|
def test_partial_transfer_is_not_published_as_cached_media(self):
|
|
result = self.run_wizard("/partial.iso")
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.qm_calls(), "")
|
|
self.assertFalse(self.iso.exists(), "A failed transfer must not become reusable media")
|
|
|
|
def test_an_empty_cached_download_is_retried(self):
|
|
self.iso.write_bytes(b"")
|
|
result = self.run_wizard()
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.iso.read_bytes(), PAYLOAD)
|
|
self.assertIn("create ", self.qm_calls())
|
|
|
|
def test_virtio_download_keeps_native_storage_despite_caller_environment(self):
|
|
self.iso.write_bytes(PAYLOAD)
|
|
unrelated = self.directory / "unrelated directory"
|
|
unrelated.mkdir()
|
|
result = self.run_wizard(windows=True, caller_iso_dir=unrelated)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
driver = self.directory / "virtio-win.iso"
|
|
self.assertTrue(driver.is_file(), result.stdout + result.stderr)
|
|
self.assertEqual(driver.read_bytes(), PAYLOAD)
|
|
self.assertIn("-ide3 ", self.qm_calls())
|
|
self.assertEqual(list(unrelated.iterdir()), [])
|
|
|
|
def test_failed_virtio_fetch_is_not_selected_for_attachment(self):
|
|
self.iso.write_bytes(PAYLOAD)
|
|
result = self.run_wizard("/missing.iso", windows=True)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertNotIn("-ide3 ", self.qm_calls())
|
|
self.assertNotIn("OK:VirtIO driver ISO downloaded successfully.", result.stdout)
|
|
self.assertIn("ERROR:Failed to download VirtIO driver ISO.", result.stdout)
|
|
self.assertFalse((self.directory / "virtio-win.iso").exists())
|
|
|
|
|
|
def test_successful_virtio_replacement_preserves_existing_permissions(self):
|
|
self.iso.write_bytes(PAYLOAD)
|
|
driver = self.directory / "virtio-win.iso"
|
|
driver.write_bytes(b"previous driver")
|
|
driver.chmod(0o640)
|
|
result = self.run_wizard(windows=True)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(driver.read_bytes(), PAYLOAD)
|
|
self.assertEqual(driver.stat().st_mode & 0o777, 0o640)
|
|
|
|
def test_empty_http_success_is_rejected(self):
|
|
result = self.run_wizard("/empty.iso")
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.qm_calls(), "")
|
|
self.assertFalse(self.iso.exists())
|
|
self.assertEqual({p.name for p in self.directory.iterdir()}, {"bin", "wget.log"})
|
|
|
|
def test_existing_nonempty_media_is_reused_without_fetching(self):
|
|
self.iso.write_bytes(b"existing installation media")
|
|
result = self.run_wizard("/missing.iso")
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.iso.read_bytes(), b"existing installation media")
|
|
self.assertFalse(self.wget_log.exists())
|
|
self.assertIn("-ide2 ", self.qm_calls())
|
|
|
|
def test_sourceforge_download_keeps_the_selected_destination_name(self):
|
|
result = self.run_wizard(url="https://sourceforge.net/projects/fixture/iso/download")
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.iso.read_bytes(), PAYLOAD)
|
|
self.assertFalse((self.directory / "remote-name.iso").exists())
|
|
self.assertEqual({p.name for p in self.directory.iterdir()},
|
|
{"bin", "wget.log", "qm.log", self.iso.name})
|
|
|
|
def test_failed_virtio_replacement_preserves_previous_media(self):
|
|
self.iso.write_bytes(PAYLOAD)
|
|
driver = self.directory / "virtio-win.iso"
|
|
driver.write_bytes(b"previous driver")
|
|
driver.chmod(0o640)
|
|
result = self.run_wizard("/partial.iso", windows=True)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(driver.read_bytes(), b"previous driver")
|
|
self.assertEqual(driver.stat().st_mode & 0o777, 0o640)
|
|
self.assertNotIn("-ide3 ", self.qm_calls())
|
|
self.assertNotIn("OK:VirtIO driver ISO downloaded successfully.", result.stdout)
|
|
|
|
def test_declining_virtio_replacement_reuses_previous_media(self):
|
|
self.iso.write_bytes(PAYLOAD)
|
|
driver = self.directory / "virtio-win.iso"
|
|
driver.write_bytes(b"previous driver")
|
|
result = self.run_wizard("/missing.iso", windows=True, overwrite=False)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(driver.read_bytes(), b"previous driver")
|
|
self.assertFalse(self.wget_log.exists())
|
|
self.assertIn("-ide3 ", self.qm_calls())
|
|
|
|
def test_empty_cached_virtio_media_is_retried_not_attached(self):
|
|
self.iso.write_bytes(PAYLOAD)
|
|
driver = self.directory / "virtio-win.iso"
|
|
driver.write_bytes(b"")
|
|
result = self.run_wizard("/missing.iso", windows=True, overwrite=False)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertTrue(self.wget_log.exists())
|
|
self.assertNotIn("-ide3 ", self.qm_calls())
|
|
self.assertIn("ERROR:Failed to download VirtIO driver ISO.", result.stdout)
|
|
|
|
def test_a_directory_cannot_be_published_as_media(self):
|
|
self.iso.mkdir()
|
|
result = self.run_wizard()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.qm_calls(), "")
|
|
self.assertTrue(self.iso.is_dir())
|
|
self.assertEqual(list(self.iso.iterdir()), [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|