khodowany
208 lines
12 KiB
Python
208 lines
12 KiB
Python
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
MOCK = '''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
from pathlib import Path
|
|
root = Path(os.environ['INSTALL_FIXTURE'])
|
|
command, args = Path(sys.argv[0]).name, sys.argv[1:]
|
|
with (root / 'commands').open('a') as output:
|
|
output.write(json.dumps([command, *args]) + '\\n')
|
|
if command == 'id':
|
|
print('0')
|
|
elif command == 'dpkg-query':
|
|
print('install ok installed')
|
|
elif command == 'docker':
|
|
if args[:2] == ['image', 'inspect']:
|
|
if os.environ.get('FIRST_INSTALL') and not (root / 'loaded').exists():
|
|
sys.exit(1)
|
|
print(os.environ['FIXTURE_IMAGE'] if (root / 'loaded').exists() else 'sha256:old')
|
|
elif args[:2] == ['image', 'ls']:
|
|
if not os.environ.get('FIRST_INSTALL'):
|
|
print('sha256:older')
|
|
elif args[:2] == ['container', 'inspect'] and os.environ.get('FIRST_INSTALL'):
|
|
sys.exit(1)
|
|
elif args[0] == 'load':
|
|
(root / 'loaded').touch()
|
|
elif args[0] == 'run' and os.environ.get('FAIL_PREFLIGHT'):
|
|
sys.exit(1)
|
|
elif args[0] == 'exec' and os.environ.get('FAIL_HEALTH'):
|
|
sys.exit(1)
|
|
elif args[0] == 'ps' and os.environ.get('SHARED_IMAGE') in args:
|
|
print('another-container')
|
|
elif command == 'systemctl' and args[0] == 'cat' and os.environ.get('FIRST_INSTALL'):
|
|
sys.exit(1)
|
|
elif command == 'nmcli' and args[:4] == ['-t', '-f', 'NAME', 'connection']:
|
|
print('product_vamp_control')
|
|
print('product-vamp-control')
|
|
print('unrelated-network')
|
|
'''
|
|
|
|
class InstallZeroTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = pathlib.Path(self.temporary.name)
|
|
self.bundle = self.root / 'bundle'
|
|
self.zero = self.bundle / 'zero'
|
|
self.zero.mkdir(parents=True)
|
|
(self.zero / 'config').mkdir()
|
|
(self.zero / 'systemd').mkdir()
|
|
self.image_tag = 'product_vamp_zero:dev'
|
|
self.write_archive(self.image_tag)
|
|
shutil.copyfile(ROOT / 'package/config/vamp.example.json', self.zero / 'config/vamp.example.json')
|
|
for name in ('start_zero.sh', 'start_ap.sh'):
|
|
shutil.copyfile(ROOT / 'package/zero' / name, self.zero / name)
|
|
for name in ('product_vamp_zero.service', 'product_vamp_ap.service'):
|
|
shutil.copyfile(ROOT / 'package/zero' / name, self.zero / 'systemd' / name)
|
|
script = (ROOT / 'package/zero/install_zero.sh').read_text()
|
|
for prefix in ('/tmp/', '/etc/', '/var/', '/run/', '/usr/local/', '/usr/share/'):
|
|
script = script.replace(prefix, str(self.root) + prefix)
|
|
(self.zero / 'install_zero.sh').write_text(script)
|
|
for path in ('etc/systemd/system', 'usr/local/bin', 'tmp', 'bin'):
|
|
(self.root / path).mkdir(parents=True, exist_ok=True)
|
|
for command in ('id', 'dpkg-query', 'apt-get', 'systemctl', 'docker', 'nmcli', 'sleep', 'chvt', 'setterm'):
|
|
path = self.root / 'bin' / command
|
|
path.write_text(MOCK)
|
|
path.chmod(0o755)
|
|
self.environment = {**os.environ, 'INSTALL_FIXTURE': str(self.root), 'FIXTURE_IMAGE': self.image_id, 'PATH': str(self.root / 'bin') + os.pathsep + os.environ['PATH']}
|
|
def write_archive(self, tag):
|
|
configuration = b'{"architecture":"arm64","os":"linux"}'
|
|
self.image_id = 'sha256:' + hashlib.sha256(configuration).hexdigest()
|
|
manifest = json.dumps([{'Config': 'image.json', 'RepoTags': [tag]}]).encode()
|
|
with tarfile.open(self.zero / 'product_vamp_zero_linux_arm64.docker.tar', 'w') as archive:
|
|
for name, data in (('manifest.json', manifest), ('image.json', configuration)):
|
|
info = tarfile.TarInfo(name)
|
|
info.size = len(data)
|
|
archive.addfile(info, io.BytesIO(data))
|
|
(self.bundle / 'manifest.json').write_text(json.dumps({'zero': {'image_tag': tag}}))
|
|
def install(self, **environment):
|
|
return subprocess.run(['bash', str(self.zero / 'install_zero.sh')], env={**self.environment, **environment}, text=True, capture_output=True)
|
|
def commands(self):
|
|
return [json.loads(line) for line in (self.root / 'commands').read_text().splitlines()]
|
|
def stale_state(self):
|
|
for name in ('etc/product_vamp/vamp.json', 'etc/product_vamp/obsolete', 'var/lib/product_vamp/vamp.db',
|
|
'var/lib/product_vamp/vamp.db-wal', 'var/lib/product_vamp/vamp.db-shm', 'run/product_vamp/heartbeat.json',
|
|
'usr/share/product_vamp/songs/removed.json', 'usr/share/product_vamp/practice/0001/electric.json', 'etc/systemd/system/product_vamp_zero.service.d/override.conf',
|
|
'run/systemd/system/product_vamp_ap.service.d/override.conf'):
|
|
path = self.root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text('previous installation')
|
|
unrelated = self.root / 'var/lib/unrelated/keep'
|
|
unrelated.parent.mkdir(parents=True)
|
|
unrelated.write_text('keep')
|
|
def test_first_install_and_reinstall_reset_only_owned_state(self):
|
|
first = self.install(FIRST_INSTALL='1')
|
|
self.assertEqual(first.returncode, 0, first.stderr)
|
|
self.stale_state()
|
|
second = self.install()
|
|
self.assertEqual(second.returncode, 0, second.stderr)
|
|
self.assertEqual((self.root / 'etc/product_vamp/vamp.json').read_bytes(), (self.zero / 'config/vamp.example.json').read_bytes())
|
|
self.assertEqual((self.root / 'etc/product_vamp/image_id').read_text().strip(), self.image_id)
|
|
self.assertFalse((self.root / 'etc/product_vamp/obsolete').exists())
|
|
self.assertFalse((self.root / 'var/lib/product_vamp').exists())
|
|
self.assertFalse((self.root / 'usr/share/product_vamp').exists())
|
|
self.assertFalse((self.root / 'etc/systemd/system/product_vamp_zero.service.d').exists())
|
|
self.assertFalse((self.root / 'run/systemd/system/product_vamp_ap.service.d').exists())
|
|
self.assertEqual(list((self.root / 'run/product_vamp').iterdir()), [])
|
|
self.assertEqual((self.root / 'var/lib/unrelated/keep').read_text(), 'keep')
|
|
self.assertEqual(list((self.root / 'tmp').iterdir()), [])
|
|
commands = self.commands()
|
|
self.assertIn(['nmcli', 'connection', 'delete', 'product_vamp_control'], commands)
|
|
self.assertNotIn(['nmcli', 'connection', 'delete', 'unrelated-network'], commands)
|
|
self.assertFalse(any('prune' in command for command in commands))
|
|
probe = next(index for index, command in enumerate(commands) if command[:2] == ['docker', 'run'])
|
|
stop = commands.index(['systemctl', 'stop', 'product_vamp_zero.service'])
|
|
self.assertLess(probe, stop)
|
|
def test_failed_preflight_keeps_old_state_and_restores_image_tag(self):
|
|
self.stale_state()
|
|
result = self.install(FAIL_PREFLIGHT='1')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual((self.root / 'var/lib/product_vamp/vamp.db').read_text(), 'previous installation')
|
|
self.assertEqual((self.root / 'etc/product_vamp/vamp.json').read_text(), 'previous installation')
|
|
self.assertNotIn(['systemctl', 'stop', 'product_vamp_zero.service'], self.commands())
|
|
self.assertIn(['docker', 'tag', 'sha256:old', 'product_vamp_zero:dev'], self.commands())
|
|
def test_legacy_paths_are_removed_only_after_successful_preflight(self):
|
|
legacy = ('etc/product-vamp/vamp.json', 'var/lib/product-vamp/vamp.db', 'run/product-vamp/heartbeat.json', 'usr/share/product-vamp/catalog.db', 'usr/local/bin/product-vamp-start-zero', 'etc/systemd/system/product-vamp-zero.service', 'etc/systemd/system/product-vamp-zero.service.d/override.conf')
|
|
for name in legacy:
|
|
path = self.root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text('previous installation')
|
|
failed = self.install(FAIL_PREFLIGHT='1')
|
|
self.assertNotEqual(failed.returncode, 0)
|
|
self.assertTrue(all((self.root / name).is_file() for name in legacy))
|
|
installed = self.install()
|
|
self.assertEqual(installed.returncode, 0, installed.stderr)
|
|
self.assertFalse(any((self.root / name).exists() for name in legacy))
|
|
self.assertIn(['systemctl', 'disable', 'product-vamp-zero.service'], self.commands())
|
|
self.assertIn(['nmcli', 'connection', 'delete', 'product-vamp-control'], self.commands())
|
|
def test_legacy_symlink_is_rejected_before_cleanup(self):
|
|
external = self.root / 'external'
|
|
external.mkdir()
|
|
(external / 'keep').write_text('keep')
|
|
(self.root / 'etc/product-vamp').symlink_to(external, target_is_directory=True)
|
|
result = self.install()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual((external / 'keep').read_text(), 'keep')
|
|
self.assertFalse(any(command[:2] == ['docker', 'load'] for command in self.commands()))
|
|
def test_failed_health_leaves_services_stopped_without_success(self):
|
|
result = self.install(FAIL_HEALTH='1')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertNotIn('Zero installed.', result.stdout)
|
|
self.assertIn(['systemctl', 'stop', 'product_vamp_zero.service', 'product_vamp_ap.service'], self.commands())
|
|
self.assertIn(['docker', 'rm', '-f', 'product_vamp_zero'], self.commands())
|
|
def test_symlink_target_aborts_without_deleting_external_data(self):
|
|
external = self.root / 'external'
|
|
external.mkdir()
|
|
(external / 'keep').write_text('keep')
|
|
(self.root / 'etc/product_vamp').symlink_to(external, target_is_directory=True)
|
|
result = self.install()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('unsafe installation target', result.stderr)
|
|
self.assertEqual((external / 'keep').read_text(), 'keep')
|
|
self.assertNotIn(['systemctl', 'stop', 'product_vamp_zero.service'], self.commands())
|
|
def test_image_in_use_by_another_container_is_not_removed(self):
|
|
result = self.install(SHARED_IMAGE='ancestor=sha256:old')
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertNotIn(['docker', 'image', 'rm', 'sha256:old'], self.commands())
|
|
self.assertIn(['docker', 'image', 'rm', 'sha256:older'], self.commands())
|
|
def test_custom_bundle_tag_selects_and_pins_the_incoming_image(self):
|
|
self.write_archive('product_vamp_zero:release-test')
|
|
result = self.install()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn(['docker', 'image', 'inspect', '--format', '{{.Id}}', 'product_vamp_zero:release-test'], self.commands())
|
|
self.assertEqual((self.root / 'etc/product_vamp/image_id').read_text().strip(), self.image_id)
|
|
def test_wrong_archive_never_resets_the_old_installation(self):
|
|
self.stale_state()
|
|
(self.bundle / 'manifest.json').write_text(json.dumps({'zero': {'image_tag': 'product_vamp_zero:missing'}}))
|
|
result = self.install()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual((self.root / 'var/lib/product_vamp/vamp.db').read_text(), 'previous installation')
|
|
self.assertFalse(any(command[:2] == ['docker', 'load'] for command in self.commands()))
|
|
def test_loaded_image_must_match_archive_digest(self):
|
|
self.stale_state()
|
|
result = self.install(FIXTURE_IMAGE='sha256:unexpected')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('does not match', result.stderr)
|
|
self.assertEqual((self.root / 'var/lib/product_vamp/vamp.db').read_text(), 'previous installation')
|
|
self.assertIn(['docker', 'tag', 'sha256:old', 'product_vamp_zero:dev'], self.commands())
|
|
def test_startup_uses_pinned_image_without_mounting_legacy_database(self):
|
|
result = self.install()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
environment = {**self.environment, 'PRODUCT_VAMP_CONFIG_DIR': str(self.root / 'etc/product_vamp'),
|
|
'PRODUCT_VAMP_RUN_DIR': str(self.root / 'run/product_vamp'), 'PRODUCT_VAMP_DISPLAY_TTY': str(self.root / 'tty')}
|
|
started = subprocess.run(['bash', str(ROOT / 'package/zero/start_zero.sh')], env=environment, text=True, capture_output=True)
|
|
self.assertEqual(started.returncode, 0, started.stderr)
|
|
command = [item for item in self.commands() if item[:2] == ['docker', 'run']][-1]
|
|
self.assertIn(self.image_id, command)
|
|
self.assertFalse(any('/var/lib/product_vamp' in argument for argument in command))
|