92 lines
4.8 KiB
Python
92 lines
4.8 KiB
Python
import importlib.util
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
MODULE_PATH = Path(__file__).parents[2] / "autoinstall/offline/install-python-profile.py"
|
|
SPEC = importlib.util.spec_from_file_location("install_python_profile", MODULE_PATH)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
class InstallPythonProfileTests(unittest.TestCase):
|
|
def test_service_probe_must_be_absolute_regular_and_executable(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
with self.assertRaisesRegex(ValueError, "SERVICE_HEALTH_PROBE_NOT_ABSOLUTE"):
|
|
MODULE.validate_service_probe(Path("relative-probe"))
|
|
probe = root / "probe"
|
|
probe.write_text("#!/bin/sh\nexit 0\n")
|
|
with self.assertRaisesRegex(ValueError, "SERVICE_HEALTH_PROBE_NOT_EXECUTABLE"):
|
|
MODULE.validate_service_probe(probe)
|
|
os.chmod(probe, 0o700)
|
|
MODULE.validate_service_probe(probe)
|
|
|
|
def test_service_probe_failure_and_timeout_are_failures(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
probe = Path(directory) / "probe"
|
|
probe.write_text("#!/bin/sh\nexit 23\n")
|
|
os.chmod(probe, 0o700)
|
|
with self.assertRaises(Exception):
|
|
MODULE.service_health(probe, Path(directory), "cpu-only", 1)
|
|
with self.assertRaisesRegex(ValueError, "SERVICE_HEALTH_TIMEOUT_INVALID"):
|
|
MODULE.install(Path("journal"), Path("checkpoint"), Path("target"),
|
|
Path("wheels"), Path("requirements"), Path(sys.executable),
|
|
probe, 0)
|
|
|
|
def test_failed_offline_resolution_rolls_back_without_target_change(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "venv"
|
|
target.mkdir()
|
|
(target / "old").write_text("old")
|
|
checkpoint = root / "checkpoint"
|
|
manifest = MODULE.checkpoint_api.create(target, checkpoint, "op-1")
|
|
journal = root / "journal.json"
|
|
identity = dict(operation_id="op-1", profile="cpu-only", manifest_sha256="a" * 64)
|
|
current = MODULE.journal_api.update(journal, expected_generation=0,
|
|
state="profile_selected", checkpoint_id=None, **identity)
|
|
MODULE.journal_api.update(journal, expected_generation=current["generation"],
|
|
state="checkpoint_ready",
|
|
checkpoint_id=manifest["inventory_sha256"], **identity)
|
|
requirements = root / "requirements.txt"
|
|
requirements.write_text("definitely-not-a-real-package==1\n")
|
|
wheels = root / "wheels"
|
|
wheels.mkdir()
|
|
with self.assertRaises(Exception):
|
|
MODULE.install(journal, checkpoint, target, wheels, requirements, Path(sys.executable))
|
|
self.assertEqual((target / "old").read_text(), "old")
|
|
self.assertEqual(MODULE.journal_api.read_verified(journal)["state"], "rolled_back")
|
|
|
|
def test_recover_activated_target_restores_checkpoint(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
target = root / "venv"
|
|
target.mkdir()
|
|
(target / "marker").write_text("old")
|
|
checkpoint = root / "checkpoint"
|
|
manifest = MODULE.checkpoint_api.create(target, checkpoint, "op-1")
|
|
journal = root / "journal.json"
|
|
identity = dict(operation_id="op-1", profile="cpu-only", manifest_sha256="a" * 64)
|
|
current = MODULE.journal_api.update(journal, expected_generation=0,
|
|
state="profile_selected", checkpoint_id=None, **identity)
|
|
current = MODULE.journal_api.update(journal, expected_generation=current["generation"],
|
|
state="checkpoint_ready",
|
|
checkpoint_id=manifest["inventory_sha256"], **identity)
|
|
MODULE.journal_api.update(journal, expected_generation=current["generation"],
|
|
state="installing", checkpoint_id=manifest["inventory_sha256"],
|
|
**identity)
|
|
(target / "marker").write_text("activated")
|
|
result = MODULE.recover(journal, checkpoint, target)
|
|
self.assertEqual(result["state"], "rolled_back")
|
|
self.assertEqual((target / "marker").read_text(), "old")
|
|
self.assertEqual((root / ".venv.rollback-discard/marker").read_text(), "activated")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|