75 lines
3.2 KiB
Python
75 lines
3.2 KiB
Python
import importlib.util
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
MODULE_PATH = Path(__file__).parents[2] / "autoinstall/offline/python-environment-checkpoint.py"
|
|
SPEC = importlib.util.spec_from_file_location("python_environment_checkpoint", MODULE_PATH)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
class PythonEnvironmentCheckpointTests(unittest.TestCase):
|
|
def test_create_verify_modify_and_restore(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "venv"
|
|
target.mkdir()
|
|
(target / "package.py").write_text("old\n")
|
|
(target / "link").symlink_to("package.py")
|
|
checkpoint = root / "checkpoint"
|
|
created = MODULE.create(target, checkpoint, "op-1")
|
|
self.assertEqual(MODULE.verify(checkpoint, target)["inventory_sha256"],
|
|
created["inventory_sha256"])
|
|
(target / "package.py").write_text("new\n")
|
|
MODULE.restore(checkpoint, target)
|
|
self.assertEqual((target / "package.py").read_text(), "old\n")
|
|
self.assertEqual((target / "link").readlink(), Path("package.py"))
|
|
|
|
def test_payload_tamper_fails(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "venv"
|
|
target.mkdir()
|
|
(target / "package.py").write_text("old\n")
|
|
checkpoint = root / "checkpoint"
|
|
MODULE.create(target, checkpoint, "op-1")
|
|
(checkpoint / "payload/package.py").write_text("tampered\n")
|
|
with self.assertRaisesRegex(MODULE.CheckpointError, "PAYLOAD_INVALID"):
|
|
MODULE.verify(checkpoint, target)
|
|
|
|
def test_absent_target_restore_quarantines_new_tree(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "venv"
|
|
checkpoint = root / "checkpoint"
|
|
MODULE.create(target, checkpoint, "op-1")
|
|
target.mkdir()
|
|
(target / "new").write_text("new")
|
|
MODULE.restore(checkpoint, target)
|
|
self.assertFalse(target.exists())
|
|
self.assertTrue((root / ".venv.rollback-discard/new").exists())
|
|
|
|
def test_recover_after_quarantine_and_after_activation(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "venv"
|
|
target.mkdir()
|
|
(target / "package.py").write_text("old\n")
|
|
checkpoint = root / "checkpoint"
|
|
MODULE.create(target, checkpoint, "op-1")
|
|
(target / "package.py").write_text("new\n")
|
|
quarantine = root / ".venv.rollback-discard"
|
|
target.rename(quarantine)
|
|
MODULE.recover_restore(checkpoint, target)
|
|
self.assertEqual((target / "package.py").read_text(), "old\n")
|
|
# A retry after activation is idempotent and preserves the failed tree.
|
|
MODULE.recover_restore(checkpoint, target)
|
|
self.assertEqual((quarantine / "package.py").read_text(), "new\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|