88 lines
4.5 KiB
Python
88 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Regression checks for isolated web-installer paths."""
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
INSTALLER = ROOT / 'scripts/install-web'
|
|
|
|
|
|
class WebInstallerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
source = json.loads((ROOT / 'config/web.json').read_text())
|
|
source['public_root'] = str(self.base / 'public')
|
|
source['storage']['root'] = str(self.base / 'private')
|
|
source['storage']['database'] = 'db/lab.sqlite3'
|
|
source['storage']['jobs'] = 'queued-jobs'
|
|
source['storage']['runtime'] = 'runtime-files'
|
|
source['storage']['catalog'] = 'catalog/parameters.json'
|
|
config_dir = self.base / 'alternate' / 'config'
|
|
config_dir.mkdir(parents=True)
|
|
self.config = config_dir / 'web.json'
|
|
self.config.write_text(json.dumps(source))
|
|
self.bin = self.base / 'bin'
|
|
self.bin.mkdir()
|
|
self.log = self.base / 'commands.log'
|
|
script = self.bin / 'command-stub'
|
|
script.write_text('#!/bin/sh\nprintf "%s\\n" "$(basename "$0") $*" >> "$INSTALLER_TEST_LOG"\nexit 97\n')
|
|
script.chmod(0o755)
|
|
self.commands = ['systemctl', 'install', 'rsync', 'chown', 'chmod', 'getent', 'useradd', 'setfacl']
|
|
for name in self.commands:
|
|
(self.bin / name).symlink_to(script)
|
|
self.env = os.environ.copy()
|
|
self.env.update(PATH=f"{self.bin}:{self.env['PATH']}", INSTALLER_TEST_LOG=str(self.log))
|
|
|
|
def run_installer(self, *args):
|
|
return subprocess.run([str(INSTALLER), *args], cwd=self.base / 'alternate', env={**self.env, 'EXOPLASIM_CONFIG': str(self.config)}, capture_output=True, text=True, timeout=15)
|
|
|
|
def test_dry_run_uses_alternate_config_from_other_working_directory_without_mutation(self):
|
|
before = {p.relative_to(self.base) for p in self.base.rglob('*')}
|
|
result = self.run_installer('--dry-run')
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn(str(self.base / 'public'), result.stdout)
|
|
self.assertIn(str(self.base / 'private'), result.stdout)
|
|
self.assertFalse(self.log.exists(), 'dry-run invoked a potentially mutating command')
|
|
self.assertEqual(before, {p.relative_to(self.base) for p in self.base.rglob('*')}, 'dry-run changed the fixture')
|
|
|
|
def test_active_jobs_block_runtime_changes_even_when_worker_is_inactive(self):
|
|
database = self.base / 'private' / 'db' / 'lab.sqlite3'
|
|
database.parent.mkdir(parents=True)
|
|
with sqlite3.connect(database) as db:
|
|
db.executescript((ROOT / 'web/server/schema.sql').read_text())
|
|
db.execute("INSERT INTO jobs(id,owner,name,spec,status,total_cycles,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)", ('a' * 32, 'local', 'fixture', '{}', 'queued', 1, 'now', 'now'))
|
|
for status in ('queued', 'running'):
|
|
with self.subTest(status=status):
|
|
with sqlite3.connect(database) as db:
|
|
db.execute('UPDATE jobs SET status=?', (status,))
|
|
result = self.run_installer()
|
|
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
|
|
self.assertIn('Refusing', result.stderr)
|
|
self.assertFalse(self.log.exists(), 'installer invoked external mutation before refusal')
|
|
with sqlite3.connect(database) as db:
|
|
self.assertEqual(db.execute('SELECT status FROM jobs').fetchone()[0], status)
|
|
|
|
def test_another_installation_blocks_runtime_changes(self):
|
|
result = subprocess.run([shutil.which('php'), '-r', 'require $argv[1]."/web/config/lab.php";echo lab_config()["runtime_lock"];', str(ROOT)], env={**self.env, 'EXOPLASIM_CONFIG': str(self.config)}, capture_output=True, text=True, timeout=5, check=True)
|
|
lock = Path(result.stdout)
|
|
lock.parent.mkdir(parents=True)
|
|
with lock.open('w') as stream:
|
|
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
result = self.run_installer()
|
|
self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
|
|
self.assertIn('installation or job claim is in progress', result.stderr)
|
|
self.assertFalse(self.log.exists(), 'a concurrent installation invoked external mutation')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|