railiance-platform/tests/test_telemetry_primary_archive.py
codex 805e0e5a2b
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s
Verify telemetry essentials recovery from versioned Scaleway archive
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
2026-09-06 23:46:20 +02:00

40 lines
1.9 KiB
Python

import importlib.util
import json
from pathlib import Path
from types import SimpleNamespace
import tempfile
import unittest
from unittest.mock import patch
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
import telemetry_primary_archive as archive
class PrimaryArchiveGuards(unittest.TestCase):
def test_unresumed_capture_and_changed_bytes_never_request_credentials(self):
with tempfile.TemporaryDirectory() as directory:
source = Path(directory) / 'source.age'
source.write_bytes(b'age-encryption.org/v1\nfixture')
receipt = Path(directory) / 'receipt.json'
args = SimpleNamespace(source=source, source_receipt=receipt)
for metadata in [
{'status': 'captured', 'production_resumed': False},
{'status': 'captured', 'production_resumed': True, 'ciphertext_sha256': 'wrong'},
]:
receipt.write_text(json.dumps(metadata))
with patch.object(archive, 'assert_cluster') as access:
with self.assertRaises(ValueError):
archive.run(args, {})
access.assert_not_called()
def test_plaintext_never_requests_credentials(self):
with tempfile.TemporaryDirectory() as directory:
source = Path(directory) / 'source.age'
source.write_bytes(b'plaintext even if the extension claims age')
receipt = Path(directory) / 'receipt.json'
receipt.write_text(json.dumps({'status': 'captured', 'production_resumed': True, 'ciphertext_sha256': archive.digest(source)}))
with patch.object(archive, 'assert_cluster') as access:
with self.assertRaises(ValueError):
archive.run(SimpleNamespace(source=source, source_receipt=receipt), {})
access.assert_not_called()