143 lines
6.3 KiB
Python
143 lines
6.3 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import runpy
|
|
import shlex
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from test_backup import Controlled
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
helper = runpy.run_path(str(ROOT / 'scripts/deploy-update-remote.py'))
|
|
template = (ROOT / 'scripts/werkjournal-backend.service.in').read_text()
|
|
controls = runpy.run_path(str(ROOT / 'scripts/service-control-remote.py'),
|
|
init_globals=dict(helper, SERVICE_UNIT_TEMPLATE=template))
|
|
ssh = runpy.run_path(str(ROOT / 'scripts/ssh-command.py'))
|
|
|
|
|
|
class Service(Controlled):
|
|
url = 'https://werkjournal.javagil.de'
|
|
dropins = ''
|
|
fail_reload = False
|
|
|
|
def command(self, *args):
|
|
self.calls.append(args)
|
|
if '--property=DropInPaths' in args:
|
|
return self.dropins
|
|
if '--property=ActiveState' in args:
|
|
return 'active' if self.running else 'inactive'
|
|
if 'daemon-reload' in args and self.fail_reload:
|
|
raise RuntimeError('reload failed')
|
|
return ''
|
|
|
|
|
|
class RemoteBasicsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
self.d = Service(Path(directory.name))
|
|
java = self.d.home / 'opt/jdk25/bin/java'
|
|
java.parent.mkdir(parents=True)
|
|
java.write_text('fixture')
|
|
self.unit = self.d.home / '.config/systemd/user/werkjournal-backend.service'
|
|
self.unit.parent.mkdir(parents=True)
|
|
|
|
def test_setup_writes_private_unit_but_preserves_secrets_and_running_process(self):
|
|
secret = self.d.home / '.config/werkjournal/environment'
|
|
secret.parent.mkdir()
|
|
secret.write_text('PRIVATE=fixture')
|
|
self.unit.write_text('previous unit')
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(0, controls['control'](self.d, 'setup'))
|
|
self.assertEqual(template.format(home=str(self.d.home)), self.unit.read_text())
|
|
self.assertEqual(0o600, self.unit.stat().st_mode & 0o777)
|
|
self.assertEqual('previous unit', self.unit.with_name(self.unit.name + '.previous').read_text())
|
|
self.assertEqual('PRIVATE=fixture', secret.read_text())
|
|
self.assertTrue(self.d.running)
|
|
self.assertEqual([('systemctl', '--user', 'show', self.d.service, '--property=DropInPaths', '--value'),
|
|
('systemctl', '--user', 'daemon-reload')], self.d.calls)
|
|
|
|
def test_failed_reload_restores_previous_unit(self):
|
|
self.unit.write_text('previous unit')
|
|
self.d.fail_reload = True
|
|
with self.assertRaisesRegex(RuntimeError, 'reload failed'):
|
|
controls['setup_unit'](self.d)
|
|
self.assertEqual('previous unit', self.unit.read_text())
|
|
|
|
def test_rendered_unit_matches_the_shared_deployment_preflight(self):
|
|
controls['setup_unit'](self.d)
|
|
with patch.object(self.d, 'command', side_effect=lambda *args: 'no' if '--property=NeedDaemonReload' in args else ''):
|
|
helper['Deployment'].preflight(self.d)
|
|
|
|
def test_setup_refuses_symlinks_overrides_and_pending_restore(self):
|
|
self.unit.symlink_to('/not-an-owned-unit')
|
|
with self.assertRaisesRegex(RuntimeError, 'symlink'):
|
|
controls['setup_unit'](self.d)
|
|
self.unit.unlink()
|
|
self.d.dropins = '/custom/override.conf'
|
|
with self.assertRaisesRegex(RuntimeError, 'overrides'):
|
|
controls['setup_unit'](self.d)
|
|
self.d.dropins = ''
|
|
(self.d.app / 'pending-restore.json').write_text('{}')
|
|
with self.assertRaisesRegex(RuntimeError, 'restore'):
|
|
controls['control'](self.d, 'setup')
|
|
self.assertFalse(self.unit.exists())
|
|
|
|
def info(self, responder):
|
|
output = io.StringIO()
|
|
with patch.object(helper['urllib'].request, 'urlopen', side_effect=responder), contextlib.redirect_stdout(output):
|
|
status = controls['control'](self.d, 'info')
|
|
return status, json.loads(output.getvalue())
|
|
|
|
def good(self, url, **kwargs):
|
|
return io.BytesIO(json.dumps({'application': 'werkjournal', 'version': 'fixture'}).encode())
|
|
|
|
def test_info_checks_internal_and_public_version_without_logging_in(self):
|
|
urls = []
|
|
def respond(url, **kwargs):
|
|
urls.append(url)
|
|
return self.good(url)
|
|
status, info = self.info(respond)
|
|
self.assertEqual(0, status)
|
|
self.assertEqual(['http://127.0.0.1:18090/api/v1/info', self.d.url + '/api/v1/info'], urls)
|
|
self.assertTrue(info['releaseVerified'])
|
|
self.assertTrue(info['backendReady'])
|
|
self.assertTrue(info['publicReady'])
|
|
self.assertEqual('a' * 40, info['commit'])
|
|
|
|
def test_info_checks_only_internal_endpoint_during_maintenance(self):
|
|
self.d.flag.write_text('maintenance')
|
|
def respond(url, **kwargs):
|
|
self.assertTrue(url.startswith('http://127.0.0.1:18090/'))
|
|
return self.good(url)
|
|
status, info = self.info(respond)
|
|
self.assertEqual(0, status)
|
|
self.assertIsNone(info['publicReady'])
|
|
self.assertTrue(info['maintenance'])
|
|
|
|
def test_info_reports_mismatch_or_transition_as_failure(self):
|
|
status, info = self.info(lambda *args, **kwargs: io.BytesIO(b'{"version":"other"}'))
|
|
self.assertEqual(1, status)
|
|
self.assertFalse(info['backendReady'])
|
|
self.d.pending.write_text('{}')
|
|
status, info = self.info(self.good)
|
|
self.assertEqual(1, status)
|
|
self.assertTrue(info['transition'])
|
|
|
|
def test_ssh_shortcut_preserves_literal_arguments_and_supports_interactive_shell(self):
|
|
args = ['printf', '%s', '$(id); `whoami` and spaces']
|
|
command = ssh['command']('user@host', args)
|
|
self.assertEqual(args, shlex.split(command[-1]))
|
|
self.assertEqual(['ssh', '-o', 'BatchMode=yes', 'user@host'], ssh['command']('user@host', []))
|
|
|
|
def test_deploy_accepts_yes_flag_but_still_requires_verified_artifact(self):
|
|
result = subprocess.run([str(ROOT / 'tools/remote'), 'backend', 'deploy', '-y', '--artifact',
|
|
str(self.d.home / 'missing')], capture_output=True, text=True)
|
|
self.assertNotEqual(0, result.returncode)
|
|
self.assertNotIn('Usage:', result.stderr)
|
|
self.assertIn('manifest.json', result.stderr)
|