108 lines
6.2 KiB
Python
108 lines
6.2 KiB
Python
"""Immutable-source compatibility evidence, not installed/release acceptance."""
|
|
import hashlib
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
import types
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
TAG = 'v2.0.3-alpha'
|
|
COMMIT = '1a4ea15df9c8800b6d288277d2839c7b93662ff7'
|
|
|
|
|
|
def historical(path, ref=TAG):
|
|
return subprocess.check_output(['git', '-C', str(ROOT), 'show', ref + ':' + path], text=True)
|
|
|
|
|
|
class ImmutableCompatibilityTests(unittest.TestCase):
|
|
def test_exact_immutable_dispatch_and_standalone_client_contract(self):
|
|
actual = subprocess.check_output(['git', '-C', str(ROOT), 'rev-parse', TAG + '^{commit}'], text=True).strip()
|
|
self.assertEqual(actual, COMMIT)
|
|
update = historical('core/archipelago/src/update.rs')
|
|
entry = update.split('pub async fn apply_update(', 1)[1].split('\n}\n', 1)[0]
|
|
self.assertIn('d09_lifecycle::prepare_apply', entry)
|
|
self.assertNotIn('apply_legacy_update', entry)
|
|
self.assertNotIn('runtime_paths', entry)
|
|
lifecycle = historical('core/archipelago/src/d09_lifecycle.rs')
|
|
self.assertIn('const CLIENT: &str = "/usr/libexec/archipelago-d09ctl"', lifecycle)
|
|
invoke = lifecycle.split('async fn invoke(', 1)[1].split('\n}\n', 1)[0]
|
|
self.assertIn('if crate::d09_transport::installed()', invoke)
|
|
self.assertIn('!crate::host_access::installed()', invoke)
|
|
self.assertIn('Command::new(CLIENT)', invoke)
|
|
self.assertIn('metadata.uid() == 0', invoke)
|
|
|
|
def test_historical_runtime_hook_is_reachable_but_nonfatal_and_after_frontend(self):
|
|
update = historical('core/archipelago/src/update.rs', 'v1.8.22-alpha')
|
|
entry = update.split('pub async fn apply_update(', 1)[1].split('/// Rollback', 1)[0]
|
|
self.assertIn('runtime_paths', entry)
|
|
self.assertIn('"archipelago" => 2', entry)
|
|
self.assertIn('n if n.contains("runtime") && n.ends_with(".tar.gz") => 1', entry)
|
|
self.assertIn('ExecStartPre=+-/opt/archipelago/scripts/ota-crash-guard.sh', entry)
|
|
self.assertIn('"archipelago-doctor.service"', entry)
|
|
bootstrap = historical('core/archipelago/src/bootstrap.rs')
|
|
self.assertIn('("scripts", "/opt/archipelago/scripts")', bootstrap)
|
|
self.assertIn('sync_doctor_script().await', bootstrap)
|
|
self.assertIn('"archipelago-doctor.service"', bootstrap)
|
|
# A helper left only in /opt/scripts or the doctor hook is not a stable
|
|
# external worker: the receiving immutable runtime promotes over it.
|
|
|
|
def test_immutable_runtime_guard_changes_candidate_template_to_preserve_live_routes(self):
|
|
namespace = {'__name__': 'immutable_guard'}
|
|
exec(compile(historical('scripts/dashboard-public-guard.py'), TAG + ':dashboard-public-guard.py', 'exec'), namespace)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
active, candidate = root / 'active.conf', root / 'candidate.conf'
|
|
guarded = namespace['guarded']
|
|
# Retain the first failed fixture input: the real guard correctly
|
|
# refuses a partial HTTP-only dashboard instead of weakening it.
|
|
http_only = 'server {\n listen 80 default_server;\n server_name _;\n location /operator-route { return 200; }\n}\n'
|
|
with self.assertRaisesRegex(ValueError, 'both default HTTP and HTTPS'):
|
|
guarded(http_only)
|
|
https = 'server {\n listen 443 ssl default_server;\n server_name _;\n location / { return 200; }\n}\n'
|
|
active.write_text(guarded(http_only + https))
|
|
candidate.write_text(guarded(http_only.replace('/operator-route', '/') + https))
|
|
original_live = active.read_bytes()
|
|
original_candidate = hashlib.sha256(candidate.read_bytes()).hexdigest()
|
|
commands = []
|
|
|
|
def nginx_validation_double(args, **kwargs):
|
|
commands.append(args)
|
|
return types.SimpleNamespace(returncode=0)
|
|
|
|
namespace['install_runtime'](active, candidate, command=nginx_validation_double, lock_path=root / 'lock')
|
|
self.assertEqual(active.read_bytes(), original_live)
|
|
self.assertEqual(candidate.read_bytes(), original_live)
|
|
self.assertNotEqual(hashlib.sha256(candidate.read_bytes()).hexdigest(), original_candidate)
|
|
self.assertIn(['nginx', '-t'], commands)
|
|
self.assertTrue(list(root.glob('runtime-template.before-preservation-*')))
|
|
|
|
def test_public_manifest_is_not_a_d09_admission_or_bootstrap_policy(self):
|
|
provision = historical('scripts/provision-d09-lifecycle.py')
|
|
for required in ['exclusion_evidence_digest', 'effective_config', 'prerequisite',
|
|
'target_binding_digest', 'restore_artifacts', 'admission_epoch_id']:
|
|
self.assertIn(required, provision)
|
|
self.assertIn("raise ValueError('existing authority; use external readmission')", provision)
|
|
client = historical('scripts/d09-lifecycle-client.py')
|
|
self.assertIn("PUBLIC = pathlib.Path('/etc/archipelago-d09/client.json')", client)
|
|
self.assertIn("selected release differs from external admission", client)
|
|
self.assertIn('binding = json.loads(read(PUBLIC))', client)
|
|
self.assertNotIn('install_package(', client)
|
|
|
|
def test_standalone_bundle_is_exact_source_only_and_never_overwrites(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
output = Path(directory) / 'owner.py'
|
|
command = ['python3', '-B', str(ROOT / 'scripts/package-public-update-worker.py'), '--output', str(output)]
|
|
subprocess.run(command, check=True, capture_output=True)
|
|
preparation = (ROOT / 'scripts/public_update_prepare.py').read_text().rsplit("\nif __name__ == '__main__':", 1)[0]
|
|
expected = (preparation + '\n' + (ROOT / 'scripts/public_update_runtime.py').read_text()
|
|
+ '\n' + (ROOT / 'scripts/public_update_adopt.py').read_text()
|
|
+ '\n' + (ROOT / 'scripts/public_update_worker.py').read_text())
|
|
self.assertEqual(output.read_text(), expected)
|
|
self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0)
|
|
self.assertEqual(output.read_text(), expected)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|