75 lines
3.8 KiB
Python
75 lines
3.8 KiB
Python
import importlib.util
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
MODULE_PATH = Path(__file__).parents[2] / "autoinstall/offline/atomic-profile-journal.py"
|
|
SPEC = importlib.util.spec_from_file_location("atomic_profile_journal", MODULE_PATH)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
IDENTITY = {"operation_id": "op-1", "profile": "cpu-only", "manifest_sha256": "a" * 64}
|
|
|
|
|
|
class AtomicProfileJournalTests(unittest.TestCase):
|
|
def test_atomic_sequence_and_permissions(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "journal.json"
|
|
first = MODULE.update(path, expected_generation=0, state="profile_selected",
|
|
checkpoint_id=None, **IDENTITY)
|
|
self.assertEqual(first["generation"], 1)
|
|
self.assertEqual(os.stat(path).st_mode & 0o777, 0o600)
|
|
second = MODULE.update(path, expected_generation=1, state="checkpoint_ready",
|
|
checkpoint_id="checkpoint-1", **IDENTITY)
|
|
third = MODULE.update(path, expected_generation=2, state="installing",
|
|
checkpoint_id=None, **IDENTITY)
|
|
self.assertEqual(third["checkpoint_id"], "checkpoint-1")
|
|
self.assertEqual(third["previous_record_sha256"], second["record_sha256"])
|
|
|
|
def test_stale_generation_and_identity_change_fail(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "journal.json"
|
|
MODULE.update(path, expected_generation=0, state="profile_selected",
|
|
checkpoint_id=None, **IDENTITY)
|
|
with self.assertRaisesRegex(MODULE.JournalError, "GENERATION_STALE"):
|
|
MODULE.update(path, expected_generation=0, state="checkpoint_ready",
|
|
checkpoint_id="checkpoint-1", **IDENTITY)
|
|
with self.assertRaisesRegex(MODULE.JournalError, "IDENTITY_CHANGED"):
|
|
MODULE.update(path, expected_generation=1, state="checkpoint_ready",
|
|
checkpoint_id="checkpoint-1", **{**IDENTITY, "profile": "nvidia-cu128"})
|
|
|
|
def test_checkpoint_required_before_installing(self):
|
|
current = MODULE.next_record(None, 0, "op-1", "cpu-only", "a" * 64,
|
|
"profile_selected", None)
|
|
with self.assertRaisesRegex(MODULE.JournalError, "TRANSITION_INVALID"):
|
|
MODULE.next_record(current, 1, "op-1", "cpu-only", "a" * 64, "installing", None)
|
|
with self.assertRaisesRegex(MODULE.JournalError, "CHECKPOINT_ID_REQUIRED"):
|
|
MODULE.next_record(current, 1, "op-1", "cpu-only", "a" * 64,
|
|
"checkpoint_ready", None)
|
|
|
|
def test_tamper_is_detected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "journal.json"
|
|
MODULE.update(path, expected_generation=0, state="profile_selected",
|
|
checkpoint_id=None, **IDENTITY)
|
|
path.write_text(path.read_text().replace("cpu-only", "nvidia-cu128"))
|
|
with self.assertRaisesRegex(MODULE.JournalError, "DIGEST_INVALID"):
|
|
MODULE.read_verified(path)
|
|
|
|
def test_rollback_is_terminal(self):
|
|
current = MODULE.next_record(None, 0, "op-1", "cpu-only", "a" * 64,
|
|
"profile_selected", None)
|
|
rolled_back = MODULE.next_record(current, 1, "op-1", "cpu-only", "a" * 64,
|
|
"rolled_back", None)
|
|
with self.assertRaisesRegex(MODULE.JournalError, "TRANSITION_INVALID"):
|
|
MODULE.next_record(rolled_back, 2, "op-1", "cpu-only", "a" * 64,
|
|
"profile_selected", None)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|