"""Engineer image > Engineering Layer upload. Run from the webapp folder: PYTHONPATH=.: python -m unittest tests.test_engineering_layer -v """ import io import os import re import shutil import sys import tempfile import time import unittest import zipfile _T = tempfile.mkdtemp(prefix="pxe-englayer-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", "STARTNET_BACKUPS_DIR": "snb"}.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())) 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 engineering_layer as EL # noqa: E402 EXE_BYTES = b"MZ" + b"\x00" * 1000 def make_zip(entries): buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as z: for name, data in entries.items(): z.writestr(name, data) return buf.getvalue() def payload_zip(version="3.1", prefix=""): return make_zip({ f"{prefix}engineering_layer/Engineering_Install_v{version}.exe": EXE_BYTES, f"{prefix}prereqs/sxs/netfx3.cab": b"cab", f"{prefix}prereqs/redistrib/v14/vc_redist.x64.exe": b"MZ..", f"{prefix}Install-EngineeringLayer.ps1": b"# from the zip - must be ignored", }) class EngineeringLayer(unittest.TestCase): def setUp(self): for d in (config.SAMBA_SHARE, config.UPDATES_DIR): shutil.rmtree(d, ignore_errors=True) os.makedirs(d) os.makedirs(os.path.join(config.SAMBA_SHARE, "gea-engineer", "Deploy")) os.makedirs(EL.root()) with open(os.path.join(EL.root(), "Install-EngineeringLayer.ps1"), "w") as f: f.write("# repo copy") def up(self, name, data): return EL.upload(io.BytesIO(data), name, len(data)) def test_zip_installs_both_parts_and_keeps_the_script(self): res = self.up("payload.zip", payload_zip("3.1")) self.assertEqual(res["version"], "3.1") st = EL.status() self.assertEqual(st["active"]["name"], "Engineering_Install_v3.1.exe") self.assertTrue(st["prereqs"]["present"] and st["prereqs"]["sxs"] and st["prereqs"]["redistrib"]) with open(os.path.join(EL.root(), "Install-EngineeringLayer.ps1")) as f: self.assertEqual(f.read(), "# repo copy") def test_zip_under_extra_engineering_folder(self): self.up("extra_engineering.zip", payload_zip("3.2", prefix="extra_engineering/")) self.assertEqual(EL.status()["active"]["version"], "3.2") def test_exe_replaces_installer_and_keeps_prereqs(self): self.up("payload.zip", payload_zip("3.1")) res = self.up("Engineering_Install_v3.2.exe", EXE_BYTES) st = EL.status() self.assertEqual([i["name"] for i in st["installers"]], ["Engineering_Install_v3.2.exe"]) self.assertTrue(st["prereqs"]["present"]) self.assertEqual(res["replaced"], ["engineering_layer"]) self.assertEqual(st["archives"][0]["version"], "3.1") def test_rejections(self): with self.assertRaisesRegex(ValueError, "upload Engineering_Install"): self.up("setup.exe", EXE_BYTES) with self.assertRaisesRegex(ValueError, "not a Windows executable"): self.up("Engineering_Install_v3.1.exe", b"PK not an exe") with self.assertRaisesRegex(ValueError, "no engineering_layer"): self.up("p.zip", make_zip({"prereqs/x.cab": b"x"})) with self.assertRaisesRegex(ValueError, "unsafe path"): self.up("p.zip", make_zip({"engineering_layer/Engineering_Install_v3.1.exe": EXE_BYTES, "engineering_layer/../../evil.txt": b"x"})) with self.assertRaisesRegex(ValueError, "damaged"): self.up("p.zip", b"not a zip") self.assertIsNone(EL.status()["active"]) # nothing changed self.assertEqual(os.listdir(EL._state("uploads")), []) # no leftovers def test_no_engineer_image_refused(self): shutil.rmtree(os.path.join(config.SAMBA_SHARE, "gea-engineer")) with self.assertRaisesRegex(RuntimeError, "import it first"): self.up("Engineering_Install_v3.1.exe", EXE_BYTES) def test_archives_pruned_to_two_and_restore(self): for v in ("3.1", "3.2", "3.3", "3.4"): self.up(f"Engineering_Install_v{v}.exe", EXE_BYTES) time.sleep(1.05) # archive names are per second st = EL.status() self.assertEqual(st["active"]["version"], "3.4") self.assertEqual([a["version"] for a in st["archives"]], ["3.3", "3.2"]) EL.restore(st["archives"][1]["name"]) st = EL.status() self.assertEqual(st["active"]["version"], "3.2") self.assertIn("3.4", [a["version"] for a in st["archives"]]) with self.assertRaises(ValueError): EL.restore("../../etc") class EngineeringLayerPages(unittest.TestCase): def setUp(self): shutil.rmtree(config.SAMBA_SHARE, ignore_errors=True) for img in ("gea-engineer", "gea-standard"): os.makedirs(os.path.join(config.SAMBA_SHARE, img, "Deploy", "Control")) self.c = A.app.test_client() def test_tab_only_on_engineer(self): eng = self.c.get("/images/gea-engineer").get_data(as_text=True) std = self.c.get("/images/gea-standard").get_data(as_text=True) self.assertIn("Engineering Layer", eng) self.assertNotIn("engineering-layer", std) r = self.c.get("/images/gea-standard/engineering-layer") self.assertEqual(r.status_code, 302) def test_page_and_upload_route(self): r = self.c.get("/images/gea-engineer/engineering-layer") self.assertEqual(r.status_code, 200) html = r.get_data(as_text=True) tok = re.search(r'name="csrf-token" content="([^"]+)"', html).group(1) r = self.c.post("/images/gea-engineer/engineering-layer/upload?name=Engineering_Install_v3.5.exe", data=EXE_BYTES, headers={"X-CSRF-Token": tok, "Content-Type": "application/octet-stream"}) self.assertEqual(r.status_code, 200, r.get_data(as_text=True)) self.assertTrue(r.get_json()["ok"]) html = self.c.get("/images/gea-engineer/engineering-layer").get_data(as_text=True) self.assertIn("Engineering_Install_v3.5.exe", html) r = self.c.post("/images/gea-engineer/engineering-layer/upload?name=bad.exe", data=EXE_BYTES, headers={"X-CSRF-Token": tok}) self.assertEqual(r.status_code, 400) if __name__ == "__main__": unittest.main()