124 lines
5.5 KiB
Python
124 lines
5.5 KiB
Python
import contextlib
|
|
import fcntl
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import runpy
|
|
import shlex
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
helper = runpy.run_path(str(ROOT / 'scripts/deploy-update-remote.py'))
|
|
controls = runpy.run_path(str(ROOT / 'scripts/service-control-remote.py'), init_globals=helper)
|
|
control = controls['control']
|
|
wrapper = runpy.run_path(str(ROOT / 'scripts/service-control.py'))
|
|
|
|
|
|
class Controlled:
|
|
service = 'werkjournal-backend.service'
|
|
|
|
def __init__(self, home):
|
|
self.app = home / 'opt/werkjournal'
|
|
self.app.mkdir(parents=True)
|
|
self.pending = self.app / 'pending-deployment.json'
|
|
self.flag = home / 'maintenance.flag'
|
|
self.calls = []
|
|
self.expected = {'commit': 'verified'}
|
|
(self.app / 'deployed.json').write_text(json.dumps(self.expected))
|
|
self.fail_ready = False
|
|
|
|
def preflight(self, check_data=True):
|
|
self.calls.append(('preflight', check_data))
|
|
|
|
def stop(self):
|
|
self.calls.append(('stop',))
|
|
|
|
def start(self):
|
|
self.calls.append(('start',))
|
|
|
|
def ready(self, expected):
|
|
self.calls.append(('ready', expected))
|
|
if self.fail_ready:
|
|
raise RuntimeError('not ready')
|
|
|
|
def command(self, *args):
|
|
self.calls.append(args)
|
|
|
|
|
|
class ServiceControlTest(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
self.deployment = Controlled(Path(directory.name))
|
|
patcher = patch.dict(control.__globals__, manifest=lambda path: self.deployment.expected)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
self.output = contextlib.redirect_stdout(io.StringIO())
|
|
self.output.__enter__()
|
|
self.addCleanup(self.output.__exit__, None, None, None)
|
|
|
|
def test_restart_only_verified_release_and_preserves_maintenance(self):
|
|
self.deployment.flag.write_text('Operator maintenance')
|
|
self.assertEqual(0, control(self.deployment, 'restart'))
|
|
self.assertEqual([('preflight', True), ('stop',), ('start',), ('ready', {'commit': 'verified'})], self.deployment.calls)
|
|
self.assertEqual('Operator maintenance', self.deployment.flag.read_text())
|
|
self.assertEqual({'commit': 'verified'}, json.loads((self.deployment.app / 'deployed.json').read_text()))
|
|
|
|
def test_unverified_release_and_pending_deployment_cannot_be_started(self):
|
|
(self.deployment.app / 'deployed.json').write_text('{"commit":"other"}')
|
|
with self.assertRaisesRegex(RuntimeError, 'differs'):
|
|
control(self.deployment, 'start')
|
|
self.assertEqual([('preflight', True)], self.deployment.calls)
|
|
self.deployment.calls.clear()
|
|
self.deployment.pending.write_text('{}')
|
|
with self.assertRaisesRegex(RuntimeError, 'pending'):
|
|
control(self.deployment, 'stop')
|
|
self.assertEqual([], self.deployment.calls)
|
|
|
|
def test_concurrent_deployment_lock_blocks_service_mutations(self):
|
|
with (self.deployment.app / '.deploy.lock').open('a') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
for action in controls['WRITE_ACTIONS']:
|
|
with self.assertRaises(BlockingIOError):
|
|
control(self.deployment, action)
|
|
self.assertEqual([], self.deployment.calls)
|
|
|
|
def test_enable_and_disable_do_not_start_or_stop_the_service(self):
|
|
for action in ('enable', 'disable'):
|
|
self.deployment.calls.clear()
|
|
control(self.deployment, action)
|
|
self.assertEqual([('preflight', False), ('systemctl', '--user', action, self.deployment.service)], self.deployment.calls)
|
|
|
|
def test_failed_readiness_is_not_success_and_preserves_maintenance(self):
|
|
self.deployment.fail_ready = True
|
|
self.deployment.flag.write_text('Keep closed')
|
|
with self.assertRaisesRegex(RuntimeError, 'not ready'):
|
|
control(self.deployment, 'start')
|
|
self.assertEqual('Keep closed', self.deployment.flag.read_text())
|
|
|
|
def test_read_only_actions_work_during_deployment_and_propagate_exit_code(self):
|
|
self.deployment.pending.write_text('{}')
|
|
with patch.object(subprocess, 'call', return_value=3) as call:
|
|
self.assertEqual(3, control(self.deployment, 'status'))
|
|
call.assert_called_once_with(['systemctl', '--user', 'status', self.deployment.service, '--no-pager'])
|
|
args = ['--since', '1 hour ago', '--grep', '$(touch /tmp/should-not-exist);`id`']
|
|
with patch.object(subprocess, 'call', return_value=0) as call:
|
|
control(self.deployment, 'log', args)
|
|
call.assert_called_once_with(['journalctl', '--user', '-u', self.deployment.service, '-n', '100', *args, '--no-pager'])
|
|
self.assertEqual([], self.deployment.calls)
|
|
|
|
def test_ssh_wrapper_round_trips_arguments_without_shell_expansion(self):
|
|
args = ['--since', '1 hour ago', '--grep', "$(touch /tmp/no); 'quoted' `id`"]
|
|
command = wrapper['command']('user@host', 'log', args)
|
|
self.assertEqual(['ssh', '-o', 'BatchMode=yes', 'user@host'], command[:4])
|
|
words = shlex.split(command[4])
|
|
self.assertEqual(['log', *args], words[3:])
|
|
self.assertEqual(['python3', '-c'], words[:2])
|
|
with self.assertRaises(ValueError):
|
|
wrapper['command']('user@host', 'restart', ['unwanted'])
|
|
with self.assertRaises(ValueError):
|
|
control(self.deployment, 'unknown')
|