99 lines
3.9 KiB
Python
99 lines
3.9 KiB
Python
import contextlib
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import runpy
|
|
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'))
|
|
recovery = runpy.run_path(str(ROOT / 'scripts/admin-recovery-remote.py'), init_globals=helper)
|
|
prepare = recovery['prepare_recovery']
|
|
|
|
|
|
class Controlled:
|
|
def __init__(self, home):
|
|
self.home = home
|
|
self.app = home / 'opt/werkjournal'
|
|
self.site = home / 'doms/werkjournal.javagil.de/htdocs-ssl'
|
|
self.site.mkdir(parents=True)
|
|
self.app.mkdir(parents=True)
|
|
self.pending = self.app / 'pending-deployment.json'
|
|
self.flag = self.site / 'maintenance.flag'
|
|
self.running = True
|
|
self.fail_ready = False
|
|
|
|
def preflight(self):
|
|
pass
|
|
|
|
def maintenance(self):
|
|
self.flag.write_text('maintenance')
|
|
|
|
def stop(self):
|
|
self.running = False
|
|
|
|
def start(self):
|
|
self.running = True
|
|
|
|
def ready(self, expected):
|
|
if self.fail_ready:
|
|
raise RuntimeError('not ready')
|
|
|
|
|
|
class AdminRecoveryOperationTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.deployment = Controlled(Path(self.directory.name))
|
|
self.target = self.deployment.home / '.config/werkjournal/admin-recovery-code'
|
|
self.manifest = patch.dict(prepare.__globals__, manifest=lambda path: {'commit': 'test'})
|
|
self.manifest.start()
|
|
self.addCleanup(self.manifest.stop)
|
|
|
|
def run_java(self, *args, **kwargs):
|
|
self.assertFalse(self.deployment.running)
|
|
self.assertTrue(self.deployment.flag.exists())
|
|
self.assertEqual('--prepare-admin-recovery', args[0][-1])
|
|
return subprocess.CompletedProcess(args[0], 0, 'a' * 43)
|
|
|
|
def test_closed_database_secret_file_and_restart_without_secret_output(self):
|
|
output = io.StringIO()
|
|
with patch.object(subprocess, 'run', side_effect=self.run_java), contextlib.redirect_stdout(output):
|
|
prepare(self.deployment)
|
|
self.assertEqual('a' * 43 + '\n', self.target.read_text())
|
|
self.assertEqual(0o600, self.target.stat().st_mode & 0o777)
|
|
self.assertNotIn('a' * 43, output.getvalue())
|
|
self.assertTrue(self.deployment.running)
|
|
self.assertFalse(self.deployment.flag.exists())
|
|
|
|
def test_refused_recovery_restarts_existing_application(self):
|
|
with patch.object(subprocess, 'run', return_value=subprocess.CompletedProcess([], 1, '')):
|
|
with self.assertRaisesRegex(RuntimeError, 'refused'):
|
|
prepare(self.deployment)
|
|
self.assertTrue(self.deployment.running)
|
|
self.assertFalse(self.deployment.flag.exists())
|
|
self.assertFalse(self.target.exists())
|
|
|
|
def test_readiness_failure_keeps_maintenance_and_does_not_restore_old_data(self):
|
|
self.deployment.fail_ready = True
|
|
with patch.object(subprocess, 'run', side_effect=self.run_java):
|
|
with self.assertRaisesRegex(RuntimeError, 'not ready'):
|
|
prepare(self.deployment)
|
|
self.assertTrue(self.deployment.flag.exists())
|
|
self.assertTrue(self.target.exists())
|
|
|
|
def test_existing_maintenance_and_symlink_are_refused_before_stopping(self):
|
|
self.deployment.flag.write_text('another operation')
|
|
with self.assertRaisesRegex(RuntimeError, 'existing'):
|
|
prepare(self.deployment)
|
|
self.deployment.flag.unlink()
|
|
self.target.parent.mkdir(parents=True, mode=0o700)
|
|
self.target.with_name(self.target.name + '.next').symlink_to('/missing')
|
|
with self.assertRaisesRegex(RuntimeError, 'layout'):
|
|
prepare(self.deployment)
|
|
self.assertTrue(self.deployment.running)
|
|
self.assertFalse(self.deployment.flag.exists())
|