87 lines
4.9 KiB
Python
87 lines
4.9 KiB
Python
"""The one-time deployment's publication and rollback boundaries, without SSH."""
|
|
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import runpy
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import urllib.error
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[2] / 'scripts/deploy-guest-preview-remote.py'
|
|
OLD = 'f7464cf232c347d59bb0025ce5eec450d239503c'
|
|
SHA = '6aaa1639aca222b759f1755033cc4916beab2f0db7ddd8093dd153c14f40e73d'
|
|
NEW = 'a' * 40
|
|
|
|
class PreviewDeploymentTest(unittest.TestCase):
|
|
def scenario(self, fail=False, existing=False):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
home = Path(directory); app = home / 'opt/werkjournal'; site = home / 'doms/werkjournal.javagil.de/htdocs-ssl'
|
|
old = app / 'releases' / OLD; old.mkdir(parents=True); site.mkdir(parents=True)
|
|
(old / 'werkjournal.jar').write_bytes(b'old')
|
|
(app / 'current').symlink_to(old.relative_to(app))
|
|
(app / 'deployed.json').write_text(json.dumps({'commit': OLD}))
|
|
if existing:
|
|
(app / 'data').mkdir(); (app / 'data/actual.mv.db').write_bytes(b'private')
|
|
checksum = hashlib.sha256(b'new').hexdigest()
|
|
manifest = {'schemaVersion': 1, 'runtime': 'java-25', 'commit': NEW, 'sha256': checksum, 'files': {'werkjournal.jar': checksum}}
|
|
data = io.BytesIO()
|
|
with tarfile.open(fileobj=data, mode='w') as archive:
|
|
for name, content in [('manifest.json', json.dumps(manifest).encode()), ('werkjournal.jar', b'new')]:
|
|
entry = tarfile.TarInfo(name); entry.size = len(content); archive.addfile(entry, io.BytesIO(content))
|
|
data.seek(0)
|
|
commands = []
|
|
def process(args, **kwargs):
|
|
commands.append(args)
|
|
if 'stop' in args:
|
|
self.assertTrue((site / 'maintenance.flag').exists())
|
|
if 'start' in args and (app / 'current').resolve().name == NEW:
|
|
(app / 'data').mkdir(); (app / 'data/werkjournal.mv.db').write_bytes(b'candidate data')
|
|
return type('Result', (), {'stdout': 'active\n'})()
|
|
def request(url, **kwargs):
|
|
if url.startswith('https:'):
|
|
self.assertTrue((site / 'maintenance.flag').exists())
|
|
raise urllib.error.HTTPError(url, 503, '', {}, io.BytesIO(b'Wartungsmodus'))
|
|
if 'v-r=init' in url:
|
|
if fail: raise RuntimeError('Injected Flow failure')
|
|
result = io.BytesIO(b'{"appConfig": {}}'); result.headers = {'Content-Type': 'application/json'}
|
|
return result
|
|
return io.BytesIO(b'{"application":"werkjournal"}')
|
|
real_digest = hashlib.file_digest
|
|
def digest(stream, algorithm):
|
|
if Path(stream.name) == old / 'werkjournal.jar':
|
|
return type('Digest', (), {'hexdigest': lambda self: SHA})()
|
|
return real_digest(stream, algorithm)
|
|
with patch.object(Path, 'home', return_value=home), patch('subprocess.run', side_effect=process), \
|
|
patch('urllib.request.urlopen', side_effect=request), patch('hashlib.file_digest', side_effect=digest), \
|
|
patch('sys.stdin', type('Input', (), {'buffer': data})()), contextlib.redirect_stdout(io.StringIO()):
|
|
if existing:
|
|
with self.assertRaises(SystemExit): runpy.run_path(str(SCRIPT))
|
|
elif fail:
|
|
with self.assertRaisesRegex(RuntimeError, 'Injected'): runpy.run_path(str(SCRIPT))
|
|
else:
|
|
runpy.run_path(str(SCRIPT))
|
|
self.assertFalse((site / 'maintenance.flag').exists())
|
|
if existing:
|
|
self.assertEqual([], commands)
|
|
self.assertEqual(b'private', (app / 'data/actual.mv.db').read_bytes())
|
|
elif fail:
|
|
self.assertEqual(OLD, (app / 'current').resolve().name)
|
|
self.assertEqual(OLD, json.loads((app / 'deployed.json').read_text())['commit'])
|
|
self.assertFalse((app / 'data').exists())
|
|
self.assertTrue((app / 'releases' / NEW / 'failed-initial-data/werkjournal.mv.db').exists())
|
|
else:
|
|
self.assertEqual(NEW, (app / 'current').resolve().name)
|
|
self.assertEqual(OLD, (app / 'previous').resolve().name)
|
|
self.assertEqual(NEW, json.loads((app / 'deployed.json').read_text())['commit'])
|
|
self.assertTrue((app / 'data/werkjournal.mv.db').exists())
|
|
|
|
def test_success_publishes_after_checks(self): self.scenario()
|
|
def test_flow_failure_restores_old_release_and_retains_failed_data(self): self.scenario(fail=True)
|
|
def test_existing_data_refuses_without_stopping_service(self): self.scenario(existing=True)
|
|
|
|
if __name__ == '__main__': unittest.main()
|