113 lines
5.4 KiB
Python
113 lines
5.4 KiB
Python
"""Verify deployment decisions without connecting to a server or Docker daemon."""
|
|
import gzip
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1]
|
|
SHA = "a" * 40
|
|
OLD = "sundynix/veyra:" + "b" * 40
|
|
|
|
class ReleaseTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="veyra-release-")
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
(self.root / "configs").mkdir()
|
|
(self.root / "configs/.env").write_text("TEST_SETTING=unchanged\n")
|
|
self.release = self.root / "releases" / SHA
|
|
self.release.mkdir(parents=True)
|
|
archive = gzip.compress(b"fake image")
|
|
(self.release / "image.tar.gz").write_bytes(archive)
|
|
digest = hashlib.sha256(archive).hexdigest()
|
|
(self.release / "image.tar.gz.sha256").write_text(digest + " image.tar.gz\n")
|
|
(self.release / "docker-compose.yml").write_text("new compose\n")
|
|
(self.root / "docker-compose.yml").write_text("old compose\n")
|
|
(self.root / ".current-image").write_text(OLD + "\n")
|
|
tools = self.root / "tools"
|
|
tools.mkdir()
|
|
mock = tools / "docker"
|
|
mock.write_text(r"""#!/usr/bin/env python3
|
|
import os, sys
|
|
from pathlib import Path
|
|
args = sys.argv[1:]
|
|
with open(os.environ['MOCK_LOG'], 'a') as log:
|
|
log.write(os.environ.get('VEYRA_IMAGE', '') + ' ' + ' '.join(args) + '\n')
|
|
if args and args[0] == 'load':
|
|
sys.stdin.buffer.read()
|
|
if '-command' in args and args[-1] == 'db-migrate' and os.environ.get('FAIL_MIGRATION'):
|
|
sys.exit(1)
|
|
if 'up' in args and os.environ.get('FAIL_READINESS') and os.environ.get('VEYRA_IMAGE') != os.environ['OLD_IMAGE']:
|
|
sys.exit(1)
|
|
""")
|
|
mock.chmod(0o755)
|
|
# macOS has no flock; emulate it for this single-process decision test.
|
|
if not __import__('shutil').which('flock'):
|
|
(tools / 'flock').write_text('#!/bin/sh\nexit 0\n')
|
|
(tools / 'flock').chmod(0o755)
|
|
if not __import__('shutil').which('sha256sum'):
|
|
checksum = tools / 'sha256sum'
|
|
checksum.write_text("""#!/usr/bin/env python3
|
|
import hashlib, sys
|
|
from pathlib import Path
|
|
for line in Path(sys.argv[2]).read_text().splitlines():
|
|
digest, name = line.split(' ', 1)
|
|
if hashlib.sha256(Path(name).read_bytes()).hexdigest() != digest:
|
|
sys.exit(1)
|
|
""")
|
|
checksum.chmod(0o755)
|
|
self.env = dict(os.environ, PATH=str(tools) + os.pathsep + os.environ['PATH'],
|
|
MOCK_LOG=str(self.root / 'commands'), OLD_IMAGE=OLD)
|
|
|
|
def run_release(self, **flags):
|
|
return subprocess.run(['bash', str(SCRIPTS / 'release.sh'), str(self.root), SHA],
|
|
env=dict(self.env, **flags), capture_output=True, text=True)
|
|
|
|
def test_success_promotes_exact_commit(self):
|
|
result = self.run_release()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual((self.root / '.current-image').read_text().strip(), 'sundynix/veyra:' + SHA)
|
|
self.assertEqual((self.root / 'configs/.env').read_text(), 'TEST_SETTING=unchanged\n')
|
|
self.assertFalse((self.release / 'image.tar.gz').exists())
|
|
calls = (self.root / 'commands').read_text()
|
|
self.assertLess(calls.index('db-migrate'), calls.index(' up '))
|
|
|
|
def test_failed_readiness_restores_previous_application(self):
|
|
self.assertNotEqual(self.run_release(FAIL_READINESS='1').returncode, 0)
|
|
self.assertEqual((self.root / '.current-image').read_text().strip(), OLD)
|
|
self.assertEqual((self.root / 'docker-compose.yml').read_text(), 'old compose\n')
|
|
self.assertIn(OLD + ' compose', (self.root / 'commands').read_text())
|
|
|
|
def test_failed_migration_does_not_replace_running_application(self):
|
|
self.assertNotEqual(self.run_release(FAIL_MIGRATION='1').returncode, 0)
|
|
self.assertNotIn(' up ', (self.root / 'commands').read_text())
|
|
self.assertEqual((self.root / 'docker-compose.yml').read_text(), 'old compose\n')
|
|
|
|
def test_invalid_archive_never_loads_image(self):
|
|
(self.release / 'image.tar.gz').write_bytes(b'corrupt')
|
|
self.assertNotEqual(self.run_release().returncode, 0)
|
|
self.assertFalse((self.root / 'commands').exists())
|
|
|
|
def test_missing_configuration_fails_before_image_load(self):
|
|
(self.root / 'configs/.env').unlink()
|
|
self.assertNotEqual(self.run_release().returncode, 0)
|
|
self.assertFalse((self.root / 'commands').exists())
|
|
|
|
def test_bootstrap_preserves_existing_configuration(self):
|
|
for _ in range(2):
|
|
result = subprocess.run(['bash', str(SCRIPTS / 'bootstrap-config.sh'), str(self.root)],
|
|
input='TEST_SETTING=new\n', capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual((self.root / 'configs/.env').read_text(), 'TEST_SETTING=unchanged\n')
|
|
(self.root / 'configs/.env').unlink()
|
|
result = subprocess.run(['bash', str(SCRIPTS / 'bootstrap-config.sh'), str(self.root)],
|
|
input='TEST_SETTING=new\n', capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual((self.root / 'configs/.env').stat().st_mode & 0o777, 0o600)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|