ge-aerospace/pxe-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
pxe-webapp-redesign
pxe-server/webapp/tests/test_clonezilla_images.py
cproudlock 9d56a5369e Clonezilla back up / restore by machine number 
Boot menu: "Clonezilla: back up this PC" and "restore this PC" ask for the
machine number and boot Clonezilla (64-bit with Debian shim, or 32-bit
i686 by cpuid) with ocs_repository=smb3://clonezilla:<pw>@server/clonezilla/,
ocs_prerun fetching clonezilla-scripts/pxe-cz.sh and ocs_live_run running
it. The script validates the machine number, saves
<machine>-<YYYY-MM-DD-HHMM> (server time), writes pxe-meta.txt, runs
ocs-chkimg, and restores only that machine's images with -k1 -r after a
typed YES. The old entry is "Clonezilla: manual".

Playbook: clonezilla Samba user for the [clonezilla] share only, password
from clonezilla_smb_password (validated, filled into the menu
placeholder), scripts installed to the web root.

Webapp: Clonezilla backups page lists image folders and zips grouped by
machine number with serial/model/version/check, serial mismatch flag,
delete, move to another machine, streamed tar download, disk space bar
and an explicit keep-last-N retention with preview. Tests and a VM test
plan added.
2026-10-15 15:26:30 -04:00

423 lines
21 KiB
Python
Raw Permalink Blame History

"""Clonezilla backups page: image listing, grouping, metadata, serial
mismatch, delete/move safety, retention preview/apply, tar download.
Run from the webapp folder:
PYTHONPATH=.:<deps> python -m unittest discover -s tests -v
Every config path points into a temp folder (env vars before config is
imported, then patched per test), so nothing on the host is touched.
"""
import io
import json
import os
import shutil
import sys
import tarfile
import tempfile
import time
import unittest
from unittest import mock
_T = tempfile.mkdtemp(prefix="pxe-cz-test-")
for _k, _sub in {"SAMBA_SHARE": "samba", "CLONEZILLA_SHARE": "cz", "BLANCCO_REPORTS": "br",
"ENROLLMENT_SHARE": "enr", "WEB_ROOT": "www", "IMAGING_DIR": "img", "UPDATES_DIR": "upd",
"TFTP_ROOT": "tftp", "BIOS_DIR": "bios", "UPLOAD_DIR": "upl", "SAMBA_LOG_DIR": "smblog"}.items():
os.environ.setdefault(_k, os.path.join(_T, _sub))
os.makedirs(os.environ[_k], exist_ok=True)
for _k in ("AUDIT_LOG", "HARDWARE_STATE", "BLANCCO_STATE", "DNSMASQ_LEASES", "APACHE_ACCESS_LOG",
"DNSMASQ_SYSLOG", "NIC_STATUS_FILE", "NIC_CONF", "DNSMASQ_NIC_CONF", "DNSMASQ_CONF", "CLONEZILLA_STATE"):
os.environ.setdefault(_k, os.path.join(_T, _k.lower()))
os.makedirs(os.path.join(os.environ["ENROLLMENT_SHARE"], "ppkgs"), exist_ok=True)
WEBAPP = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, WEBAPP)
os.chdir(WEBAPP)
import config # noqa: E402
import app as A # noqa: E402
from services import clonezilla_images as CZ # noqa: E402
DMI = """This image was saved from this machine with DMI info at 2025-01-01:
# dmidecode 3.5
Handle 0x0001, DMI type 1, 27 bytes
System Information
\tManufacturer: Dell Inc.
\tProduct Name: OptiPlex 7010
\tSerial Number: DMI777
\tUUID: 4c4c4544
Handle 0x0002, DMI type 2, 15 bytes
Base Board Information
\tSerial Number: BOARD1
"""
def make_image(root, name, meta=None, disk="sda", parts="sda1 sda2", data=b"x" * 3000, dmi=None, saved_by=None):
d = os.path.join(root, name)
os.makedirs(d)
with open(os.path.join(d, "disk"), "w") as f:
f.write(disk + "\n")
with open(os.path.join(d, "parts"), "w") as f:
f.write(parts + "\n")
with open(os.path.join(d, "sda1.ntfs-ptcl-img.zst"), "wb") as f:
f.write(data)
with open(os.path.join(d, "clonezilla-img"), "w") as f:
f.write("This image was saved by Clonezilla at 2025-01-01.\n")
if saved_by:
f.write(f"Saved by {saved_by}.\n")
if dmi:
with open(os.path.join(d, "Info-dmi.txt"), "w") as f:
f.write(dmi)
if meta is not None:
with open(os.path.join(d, "pxe-meta.txt"), "w") as f:
f.write("".join(f"{k}={v}\n" for k, v in meta.items()))
return d
def good_meta(machine, serial="SER1", model="OptiPlex 7090", **kw):
m = {"machine": machine, "serial": serial, "model": model, "manufacturer": "Dell Inc.",
"disk": "sda", "disk_model": "WDC SSD", "disk_size_bytes": "256060514304",
"clonezilla": "clonezilla-live-3.3.3-37-amd64", "bits": "64", "firmware": "uefi",
"save": "ok", "check": "ok", "date": "2026-10-15 14:03 EDT"}
m.update(kw)
return m
class Base(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="cz-case-")
self.share = os.path.join(self.tmp, "share")
os.makedirs(self.share)
self.state = os.path.join(self.tmp, "state", "clonezilla-settings.json")
p1 = mock.patch.object(config, "CLONEZILLA_SHARE", self.share)
p2 = mock.patch.object(config, "CLONEZILLA_STATE", self.state)
p1.start()
p2.start()
self.addCleanup(p1.stop)
self.addCleanup(p2.stop)
self.addCleanup(shutil.rmtree, self.tmp, True)
self.c = A.app.test_client()
with self.c.session_transaction() as s:
s["_csrf_token"] = "t"
def post(self, url, data=None):
d = {"_csrf_token": "t"}
d.update(data or {})
return self.c.post(url, data=d)
class TestListing(Base):
def test_detects_only_image_dirs(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
os.makedirs(os.path.join(self.share, "random-folder"))
os.makedirs(os.path.join(self.share, "only-disk"))
open(os.path.join(self.share, "only-disk", "disk"), "w").close() # disk but no image file
open(os.path.join(self.share, "disk"), "w").close() # a stray file at the top
os.makedirs(os.path.join(self.share, ".hidden-2026-01-01-0000"))
names = [i["name"] for i in CZ.list_images()]
self.assertEqual(names, ["6501-2026-10-15-1403"])
def test_symlinked_dir_is_not_listed(self):
outside = os.path.join(self.tmp, "outside")
os.makedirs(outside)
make_image(outside, "6501-2026-01-01-0000")
os.symlink(os.path.join(outside, "6501-2026-01-01-0000"), os.path.join(self.share, "6501-2026-01-01-0000"))
self.assertEqual(CZ.list_images(), [])
def test_meta_is_read(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", check_date="2026-10-15 14:40 EDT"))
i = CZ.list_images()[0]
self.assertEqual((i["machine"], i["machine_source"]), ("6501", "meta"))
self.assertEqual(i["serial"], "SER1")
self.assertEqual(i["model"], "OptiPlex 7090")
self.assertEqual(i["bits"], "64")
self.assertEqual(i["clonezilla"], "clonezilla-live-3.3.3-37-amd64")
self.assertEqual(i["status"], "ok")
self.assertEqual(i["disks"], ["sda"])
self.assertEqual(i["parts"], ["sda1", "sda2"])
self.assertEqual(i["disk_size"], 256060514304)
self.assertGreater(i["size"], 0)
def test_manual_image_falls_back_to_name_dmi_and_clonezilla_img(self):
d = make_image(self.share, "wj-7001-2025-03-04-0915", dmi=DMI, saved_by="clonezilla-live-3.2.0-5-i686")
old = time.time() - 30 * 86400
os.utime(d, (old, old))
i = CZ.list_images()[0]
self.assertEqual((i["machine"], i["machine_source"]), ("WJ-7001", "name"))
self.assertEqual((i["serial"], i["model"], i["manufacturer"]), ("DMI777", "OptiPlex 7010", "Dell Inc."))
self.assertEqual(i["clonezilla"], "clonezilla-live-3.2.0-5-i686")
self.assertEqual(i["bits"], "32")
self.assertEqual(i["status"], "manual")
def test_fresh_image_without_meta_is_in_progress(self):
make_image(self.share, "6501-2026-10-15-1403")
self.assertEqual(CZ.list_images()[0]["status"], "in-progress")
def test_failed_statuses(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", save="failed", check="not-run"))
make_image(self.share, "6501-2026-10-14-1403", good_meta("6501", check="failed"))
st = {i["name"]: i["status"] for i in CZ.list_images()}
self.assertEqual(st, {"6501-2026-10-15-1403": "failed", "6501-2026-10-14-1403": "check-failed"})
def test_unreadable_and_odd_contents_never_500(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
bad = make_image(self.share, "6502-2026-10-15-1403", good_meta("6502"))
with open(os.path.join(bad, "pxe-meta.txt"), "wb") as f:
f.write(b"\xff\xfe machine=\x00garbage\nno equals sign\n=\n")
os.makedirs(os.path.join(self.share, "locked-2026-01-01-0000"))
os.chmod(os.path.join(self.share, "locked-2026-01-01-0000"), 0)
self.addCleanup(os.chmod, os.path.join(self.share, "locked-2026-01-01-0000"), 0o755)
with open(os.path.join(self.share, "weird \u00e9 name.zip"), "wb") as f:
f.write(b"PK")
r = self.c.get("/backups")
self.assertEqual(r.status_code, 200)
self.assertIn(b"6501-2026-10-15-1403", r.data)
def test_share_missing_still_renders(self):
shutil.rmtree(self.share)
r = self.c.get("/backups")
self.assertEqual(r.status_code, 200)
self.assertIn(b"No backups yet", r.data)
class TestGrouping(Base):
def test_grouped_newest_first_with_zips_and_unassigned(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
make_image(self.share, "6501-2026-09-01-0800", good_meta("6501"))
make_image(self.share, "6501-2026-10-15-1403-2", good_meta("6501"))
make_image(self.share, "6502-2026-08-01-0800", good_meta("6502", serial="S2"))
d = make_image(self.share, "2026-07-01-10-img") # Clonezilla's default autoname
old = time.time() - 30 * 86400
os.utime(d, (old, old))
with open(os.path.join(self.share, "6502.zip"), "wb") as f:
f.write(b"PK")
groups = CZ.grouped()
self.assertEqual([g["machine"] for g in groups], ["6502", "6501", ""]) # zip just written: 6502 newest
g6501 = next(g for g in groups if g["machine"] == "6501")
self.assertEqual([i["name"] for i in g6501["images"]],
["6501-2026-10-15-1403-2", "6501-2026-10-15-1403", "6501-2026-09-01-0800"])
g6502 = next(g for g in groups if g["machine"] == "6502")
self.assertEqual([z["filename"] for z in g6502["zips"]], ["6502.zip"])
page = self.c.get("/backups").data
self.assertIn(b"No machine number", page)
self.assertIn(b"6502.zip", page)
def test_machine_case_is_normalised(self):
make_image(self.share, "wj-1-2026-10-15-1403", good_meta("wj-1"))
make_image(self.share, "WJ-1-2026-10-14-1403")
self.assertEqual([g["machine"] for g in CZ.grouped()], ["WJ-1"])
def test_serial_mismatch_flag(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", serial="AAA111"))
make_image(self.share, "6501-2026-10-14-1403", good_meta("6501", serial="aaa111 "))
make_image(self.share, "6502-2026-10-14-1403", good_meta("6502", serial="X"))
groups = {g["machine"]: g for g in CZ.grouped()}
self.assertFalse(groups["6501"]["mismatch"])
make_image(self.share, "6501-2026-10-13-1403", good_meta("6501", serial="BBB222"))
groups = {g["machine"]: g for g in CZ.grouped()}
self.assertTrue(groups["6501"]["mismatch"])
self.assertEqual(groups["6501"]["serials"], ["AAA111", "BBB222"])
self.assertEqual([i["serial_differs"] for i in groups["6501"]["images"]], [False, False, True])
self.assertFalse(groups["6502"]["mismatch"])
page = self.c.get("/backups").data
self.assertIn(b"Serial mismatch", page)
self.assertIn(b"different serial numbers", page)
class TestDeleteMove(Base):
def test_delete_image(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
r = self.post("/backups/image/6501-2026-10-15-1403/delete")
self.assertEqual(r.status_code, 302)
self.assertFalse(os.path.exists(os.path.join(self.share, "6501-2026-10-15-1403")))
def test_delete_needs_csrf(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
r = self.c.post("/backups/image/6501-2026-10-15-1403/delete")
self.assertEqual(r.status_code, 403)
self.assertTrue(os.path.isdir(os.path.join(self.share, "6501-2026-10-15-1403")))
def test_delete_refuses_unsafe_names(self):
outside = os.path.join(self.tmp, "outside")
make_image(outside, "victim")
os.makedirs(os.path.join(self.share, "plain-folder"))
os.symlink(os.path.join(outside, "victim"), os.path.join(self.share, "link"))
for bad in ("..", ".", "...", "%2e%2e", "..%2Foutside%2Fvictim", "link", "plain-folder", "nope", ".hidden"):
with self.subTest(bad=bad):
self.post(f"/backups/image/{bad}/delete")
for bad in ("..", "../outside/victim", "link", "plain-folder", "", "a/b", "a\\b", "x\x00"):
with self.subTest(direct=bad), self.assertRaises(CZ.ImageError):
CZ.delete_image(bad)
self.assertTrue(os.path.isdir(os.path.join(outside, "victim")))
self.assertTrue(os.path.isdir(os.path.join(self.share, "plain-folder")))
self.assertTrue(os.path.islink(os.path.join(self.share, "link")))
def test_move_renames_and_updates_meta(self):
make_image(self.share, "6510-2026-10-15-1403", good_meta("6510"))
r = self.post("/backups/image/6510-2026-10-15-1403/move", {"machine": "6501"})
self.assertEqual(r.status_code, 302)
new = os.path.join(self.share, "6501-2026-10-15-1403")
self.assertTrue(os.path.isdir(new))
meta = CZ.read_meta(new)
self.assertEqual(meta["machine"], "6501")
self.assertEqual(meta["moved_from"], "6510")
self.assertEqual(meta["serial"], "SER1")
self.assertEqual([i["name"] for i in CZ.list_images()], ["6501-2026-10-15-1403"])
def test_move_clash_gets_suffix_and_case_twin_counts(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
make_image(self.share, "6510-2026-10-15-1403", good_meta("6510"))
self.assertEqual(CZ.move_image("6510-2026-10-15-1403", "6501"), "6501-2026-10-15-1403-2")
make_image(self.share, "ab-2026-10-15-1403", good_meta("ab"))
make_image(self.share, "x-2026-10-15-1403", good_meta("x"))
self.assertEqual(CZ.move_image("x-2026-10-15-1403", "AB"), "AB-2026-10-15-1403-2")
def test_move_manual_image_without_stamp(self):
d = make_image(self.share, "2026-07-01-10-img")
os.utime(d, (1700000000, 1700000000))
new = CZ.move_image("2026-07-01-10-img", "6501")
self.assertRegex(new, r"^6501-\d{4}-\d{2}-\d{2}-\d{4}$")
self.assertEqual(CZ.read_meta(os.path.join(self.share, new))["machine"], "6501")
def test_move_refuses_bad_machine(self):
make_image(self.share, "6510-2026-10-15-1403", good_meta("6510"))
for bad in ("", "-6501", "65 01", "../x", "a/b", "x" * 33, "65_01"):
with self.subTest(bad=bad), self.assertRaises(CZ.ImageError):
CZ.move_image("6510-2026-10-15-1403", bad)
self.assertTrue(os.path.isdir(os.path.join(self.share, "6510-2026-10-15-1403")))
class TestRetention(Base):
def _three(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
make_image(self.share, "6501-2026-10-01-1403", good_meta("6501"))
make_image(self.share, "6501-2026-09-01-1403", good_meta("6501"))
make_image(self.share, "6502-2026-09-01-1403", good_meta("6502"))
def test_settings_validate_and_persist(self):
self.assertEqual(CZ.settings(), {"keep_last": 0, "warn_percent": 85})
r = self.post("/backups/settings", {"keep_last": "2", "warn_percent": "90"})
self.assertEqual(r.status_code, 302)
with open(self.state) as f:
self.assertEqual(json.load(f), {"keep_last": 2, "warn_percent": 90})
for bad in (("-1", "85"), ("101", "85"), ("2", "40"), ("x", "85")):
with self.subTest(bad=bad), self.assertRaises(CZ.ImageError):
CZ.save_settings(*bad)
self.assertEqual(CZ.settings()["keep_last"], 2)
def test_preview_lists_without_deleting(self):
self._three()
CZ.save_settings(1, 85)
r = self.c.get("/backups/retention")
self.assertEqual(r.status_code, 200)
self.assertIn(b"6501-2026-10-01-1403", r.data)
self.assertIn(b"6501-2026-09-01-1403", r.data)
self.assertIn(b"will be DELETED", r.data)
self.assertEqual(len(CZ.list_images()), 4)
def test_plan_keeps_newest_good_and_newer_failed(self):
make_image(self.share, "6501-2026-10-16-0900", good_meta("6501", save="failed")) # newer, failed: kept
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501", check="failed")) # check failed: not good
make_image(self.share, "6501-2026-10-01-1403", good_meta("6501")) # newest good: kept
make_image(self.share, "6501-2026-09-01-1403", good_meta("6501")) # delete
d = make_image(self.share, "2026-07-01-10-img") # unassigned: never
os.utime(d, (1700000000, 1700000000))
plan = [p["name"] for p in CZ.retention_plan(1)]
self.assertEqual(plan, ["6501-2026-09-01-1403"])
self.assertEqual(CZ.retention_plan(0), [])
self.assertEqual(CZ.retention_plan(5), [])
def test_apply_deletes_only_planned_names(self):
self._three()
CZ.save_settings(1, 85)
r = self.c.post("/backups/retention/apply", data={
"_csrf_token": "t",
"name": ["6501-2026-10-01-1403", "6501-2026-09-01-1403", "6501-2026-10-15-1403", "6502-2026-09-01-1403", "../x"]})
self.assertEqual(r.status_code, 302)
left = sorted(i["name"] for i in CZ.list_images())
self.assertEqual(left, ["6501-2026-10-15-1403", "6502-2026-09-01-1403"])
def test_stale_preview_cannot_delete_a_now_kept_image(self):
self._three()
CZ.save_settings(1, 85)
planned = [p["name"] for p in CZ.retention_plan()]
shutil.rmtree(os.path.join(self.share, "6501-2026-10-15-1403")) # newest removed meanwhile
deleted, skipped = CZ.apply_retention(planned)
self.assertEqual(deleted, ["6501-2026-09-01-1403"])
self.assertEqual(skipped, ["6501-2026-10-01-1403"])
self.assertTrue(os.path.isdir(os.path.join(self.share, "6501-2026-10-01-1403")))
def test_apply_needs_csrf_and_setting(self):
self._three()
r = self.c.post("/backups/retention/apply", data={"name": ["6501-2026-09-01-1403"]})
self.assertEqual(r.status_code, 403)
self.post("/backups/retention/apply", {"name": "6501-2026-09-01-1403"}) # keep_last 0: nothing
self.assertEqual(len(CZ.list_images()), 4)
class TestTarAndSpace(Base):
def test_tar_download_streams_the_folder(self):
data = os.urandom(3 * 1024 * 1024 + 123)
d = make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"), data=data)
os.makedirs(os.path.join(d, "sub"))
with open(os.path.join(d, "sub", "f.txt"), "w") as f:
f.write("hello")
os.symlink("/etc/passwd", os.path.join(d, "evil-link"))
r = self.c.get("/backups/image/6501-2026-10-15-1403/tar")
self.assertEqual(r.status_code, 200)
self.assertEqual(r.mimetype, "application/x-tar")
self.assertIn('filename="6501-2026-10-15-1403.tar"', r.headers["Content-Disposition"])
self.assertTrue(r.is_streamed)
body = r.get_data()
self.assertEqual(len(body) % tarfile.RECORDSIZE, 0)
with tarfile.open(fileobj=io.BytesIO(body)) as t:
names = t.getnames()
self.assertIn("6501-2026-10-15-1403/sda1.ntfs-ptcl-img.zst", names)
self.assertIn("6501-2026-10-15-1403/sub/f.txt", names)
self.assertNotIn("6501-2026-10-15-1403/evil-link", names)
self.assertEqual(t.extractfile("6501-2026-10-15-1403/sda1.ntfs-ptcl-img.zst").read(), data)
self.assertEqual(t.extractfile("6501-2026-10-15-1403/sub/f.txt").read(), b"hello")
def test_tar_of_shrinking_file_stays_valid(self):
d = make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"), data=b"y" * 5000)
gen = CZ.tar_stream("6501-2026-10-15-1403")
chunks = []
for c in gen:
chunks.append(c)
if len(chunks) == 1: # first header out: shrink a file before it is read
with open(os.path.join(d, "sda1.ntfs-ptcl-img.zst"), "wb") as f:
f.write(b"y" * 10)
with tarfile.open(fileobj=io.BytesIO(b"".join(chunks))) as t:
m = t.getmember("6501-2026-10-15-1403/sda1.ntfs-ptcl-img.zst")
self.assertEqual(len(t.extractfile(m).read()), m.size)
def test_tar_bad_name_redirects(self):
for bad in ("..", "nope", "plain"):
with self.subTest(bad=bad):
r = self.c.get(f"/backups/image/{bad}/tar")
self.assertEqual(r.status_code, 302)
def test_disk_space_and_warning(self):
make_image(self.share, "6501-2026-10-15-1403", good_meta("6501"))
sp = CZ.disk_space(50)
self.assertGreater(sp["total"], 0)
fake = shutil._ntuple_diskusage(100, 95, 5)
with mock.patch.object(CZ.shutil, "disk_usage", return_value=fake):
self.assertTrue(CZ.disk_space(90)["warn"])
self.assertFalse(CZ.disk_space(96)["warn"])
page = self.c.get("/backups").data
self.assertIn(b"Share disk space", page)
self.assertIn(b"95% full", page)
def test_zip_routes_still_work(self):
with open(os.path.join(self.share, "6501.zip"), "wb") as f:
f.write(b"PK-zip")
r = self.c.get("/backups/download/6501.zip")
self.assertEqual(r.data, b"PK-zip")
r.close()
self.post("/backups/delete/6501.zip")
self.assertFalse(os.path.exists(os.path.join(self.share, "6501.zip")))
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink