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('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_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))