175 lines
8.1 KiB
Python
175 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Focused refusal/snapshot checks against pinned fixture inputs.
|
|
|
|
Usage: python3 -B scripts/test-podsteadr-manifest-output.py PATH/fixture_manifests.py
|
|
Generator and companion bytes are verified; generation reads a private checked
|
|
configuration snapshot. Both input pins and repeated output hashes are printed.
|
|
External source is read-only. All writes are synthetic children of TMPDIR.
|
|
This is not a managed-app, key-binding, production-manifest or migration test.
|
|
"""
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
PIN = '0d61e839d4d2cb797c59130895a87bd50cb6e21327b930d1ed3aa27dc72ca13c'
|
|
COMPANION_PIN = '654b30dc66322fc21294911b23b31f38acc2e4b2aa2f0b3831cb89e9936624c4'
|
|
source = Path(sys.argv.pop(1))
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def load_fixture(source, companion=None):
|
|
if source.is_symlink() or not source.is_file():
|
|
raise ValueError('Explicit regular source file required')
|
|
raw = source.read_bytes()
|
|
if hashlib.sha256(raw).hexdigest() != PIN:
|
|
raise ValueError('Fixture generator differs from selected source pin')
|
|
companion = companion if companion is not None else source.with_name('mediamtx-fixture.yml')
|
|
if companion.is_symlink() or not companion.is_file():
|
|
raise ValueError('Explicit regular companion file required')
|
|
config = companion.read_bytes()
|
|
if hashlib.sha256(config).hexdigest() != COMPANION_PIN:
|
|
raise ValueError('Fixture companion differs from selected configuration pin')
|
|
# The generator must read the checked snapshot, not reopen the external path.
|
|
# This is controlled fixture preparation, not a hostile-filesystem sandbox.
|
|
with tempfile.TemporaryDirectory(prefix='podsteadr-checked-') as snapshot:
|
|
snapshot_source = Path(snapshot) / source.name
|
|
(Path(snapshot) / 'mediamtx-fixture.yml').write_bytes(config)
|
|
spec = importlib.util.spec_from_loader('fixture_manifests_checked', loader=None)
|
|
assert spec is not None
|
|
checked = importlib.util.module_from_spec(spec)
|
|
checked.__file__ = str(snapshot_source)
|
|
exec(compile(raw, str(source), 'exec'), checked.__dict__)
|
|
yield checked
|
|
|
|
|
|
class OutputRefusals(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.output = self.root / 'output'
|
|
self.output.mkdir()
|
|
|
|
def invoke(self, output=None, overrides=None):
|
|
values = {
|
|
'app-image': 'a' * 64, 'mtx-image': 'b' * 64,
|
|
'blossom-image': 'c' * 64, 'server-pubkey': 'd' * 64,
|
|
'data-root': '/var/lib/archipelago/fixture-podsteadr-output-test',
|
|
'output': str(output if output is not None else self.output),
|
|
}
|
|
values.update(overrides or {})
|
|
argv = ['fixture']
|
|
for name, value in values.items():
|
|
argv.extend(['--' + name, value])
|
|
with patch.object(sys, 'argv', argv), contextlib.redirect_stdout(io.StringIO()):
|
|
fixture.main()
|
|
|
|
def test_changed_companion_is_refused_before_generation(self):
|
|
companion = self.root / 'mediamtx-fixture.yml'
|
|
companion.write_bytes(b'unverified configuration')
|
|
with self.assertRaisesRegex(ValueError, 'companion'):
|
|
with load_fixture(source, companion):
|
|
self.fail('unverified companion accepted')
|
|
self.assertEqual(list(self.output.iterdir()), [])
|
|
|
|
def test_verified_companion_snapshot_survives_path_change(self):
|
|
companion = self.root / 'mediamtx-fixture.yml'
|
|
verified = source.with_name('mediamtx-fixture.yml').read_bytes()
|
|
companion.write_bytes(verified)
|
|
copied_source = self.root / 'fixture_manifests.py'
|
|
copied_source.write_bytes(source.read_bytes())
|
|
with load_fixture(copied_source, companion) as checked:
|
|
companion.write_bytes(b'unverified configuration')
|
|
with patch.dict(globals(), fixture=checked):
|
|
self.invoke()
|
|
import json
|
|
manifest = json.loads((self.output / 'podsteadr-mediamtx' / 'manifest.yml').read_text())
|
|
expected = verified.decode('utf-8').replace(
|
|
'http://127.0.0.1:8095/api/mediamtx/auth',
|
|
'http://podsteadr-app:8095/api/mediamtx/auth')
|
|
expected = '# Fixture network aliases; not an installed first-start contract.\n' + '\n'.join(
|
|
line for line in expected.splitlines() if not line.startswith('#')) + '\n'
|
|
self.assertEqual(manifest['app']['files'][0]['content'], expected)
|
|
|
|
def test_missing_or_symlink_companion_is_refused(self):
|
|
missing = self.root / 'missing.yml'
|
|
link = self.root / 'link.yml'
|
|
link.symlink_to(source.with_name('mediamtx-fixture.yml'))
|
|
for companion in (missing, link):
|
|
with self.subTest(companion=companion.name):
|
|
with self.assertRaisesRegex(ValueError, 'regular companion'):
|
|
with load_fixture(source, companion):
|
|
self.fail('nonregular companion accepted')
|
|
|
|
def test_repeated_generation_matches_exact_output_bytes(self):
|
|
import json
|
|
second = self.root / 'second'
|
|
second.mkdir()
|
|
self.invoke()
|
|
self.invoke(second)
|
|
first_files = {str(path.relative_to(self.output)): path.read_bytes()
|
|
for path in self.output.glob('*/manifest.yml')}
|
|
second_files = {str(path.relative_to(second)): path.read_bytes()
|
|
for path in second.glob('*/manifest.yml')}
|
|
self.assertEqual(set(first_files), {
|
|
'podsteadr/manifest.yml', 'podsteadr-mediamtx/manifest.yml',
|
|
'podsteadr-blossom/manifest.yml'})
|
|
self.assertEqual(first_files, second_files)
|
|
print(json.dumps({'generator_sha256': PIN, 'companion_sha256': COMPANION_PIN,
|
|
'fixture_output_sha256': {name: hashlib.sha256(raw).hexdigest()
|
|
for name, raw in sorted(first_files.items())}}, sort_keys=True))
|
|
|
|
def test_output_symlink_preserves_empty_target(self):
|
|
link = self.root / 'link'
|
|
link.symlink_to(self.output, target_is_directory=True)
|
|
with self.assertRaises(ValueError): self.invoke(link)
|
|
self.assertEqual(list(self.output.iterdir()), [])
|
|
self.assertTrue(link.is_symlink())
|
|
|
|
def test_non_directory_and_missing_output_are_not_replaced(self):
|
|
file = self.root / 'file'
|
|
file.write_bytes(b'preserve sentinel')
|
|
with self.assertRaises(ValueError): self.invoke(file)
|
|
self.assertEqual(file.read_bytes(), b'preserve sentinel')
|
|
missing = self.root / 'missing'
|
|
with self.assertRaises(ValueError): self.invoke(missing)
|
|
self.assertFalse(missing.exists())
|
|
|
|
def test_even_empty_existing_child_prevents_partial_writes(self):
|
|
child = self.output / 'podsteadr-mediamtx'
|
|
child.mkdir()
|
|
with self.assertRaises(ValueError): self.invoke()
|
|
self.assertEqual(list(self.output.iterdir()), [child])
|
|
self.assertEqual(list(child.iterdir()), [])
|
|
|
|
def test_invalid_inputs_refused_before_any_output(self):
|
|
for field, value in (
|
|
('app-image', 'sha256:' + 'a' * 63),
|
|
('mtx-image', 'b' * 64 + '\n'),
|
|
('blossom-image', 'C' * 64),
|
|
('server-pubkey', 'd' * 64 + '\n'),
|
|
('data-root', '/var/lib/archipelago/fixture-podsteadr-x/../operator'),
|
|
):
|
|
with self.subTest(field=field):
|
|
with self.assertRaises(ValueError): self.invoke(overrides={field: value})
|
|
self.assertEqual(list(self.output.iterdir()), [])
|
|
|
|
def test_dangling_output_symlink_is_not_followed(self):
|
|
target = self.root / 'missing-target'
|
|
link = self.root / 'dangling'
|
|
link.symlink_to(target, target_is_directory=True)
|
|
with self.assertRaises(ValueError): self.invoke(link)
|
|
self.assertTrue(link.is_symlink())
|
|
self.assertFalse(target.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
with load_fixture(source) as fixture:
|
|
unittest.main()
|