238 lines
11 KiB
Python
238 lines
11 KiB
Python
"""Release publication and controller recovery tests without a live server."""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from test_cicd import ROOT, controller, release, release_module
|
|
|
|
|
|
class ArtifactTests(unittest.TestCase):
|
|
def test_archive_rejects_nested_or_extra_files(self):
|
|
api = object.__new__(release_module.Gitea)
|
|
api.base = 'https://example.test/api/v1/repos/a/b'
|
|
for names in (['../release.json'], ['release.json', 'credentials']):
|
|
blob = io.BytesIO()
|
|
with zipfile.ZipFile(blob, 'w') as archive:
|
|
for name in names:
|
|
archive.writestr(name, json.dumps(release()))
|
|
replies = [{'artifacts': [{'id': 1, 'name': 'release-' + 'a' * 40}]}, blob.getvalue()]
|
|
with (
|
|
patch.object(api, 'pages', return_value=iter([{'name': 'build', 'conclusion': 'success'}])),
|
|
patch.object(api, 'request', side_effect=replies),
|
|
self.assertRaisesRegex(ValueError, 'archive'),
|
|
):
|
|
api.release({'id': 1, 'head_sha': 'a' * 40})
|
|
|
|
def test_quoted_and_single_platform_images_render_from_checked_release(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
path = Path(scratch) / 'release.json'
|
|
path.write_text(json.dumps(release()))
|
|
result = io.StringIO()
|
|
image = 'gcr.forust.xyz/forust/error-pages'
|
|
with patch.dict(os.environ, {'RELEASE_FILE': str(path), 'DEPLOY_SHA': 'a' * 40}):
|
|
release_module.render(io.StringIO(f' image: "{image}:prod" # note\n'), result)
|
|
self.assertEqual(result.getvalue(), f' image: "{image}@sha256:{"b" * 64}" # note\n')
|
|
|
|
def test_edu_release_digest_is_preserved(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
path = Path(scratch) / 'release.json'
|
|
path.write_text(json.dumps(release()))
|
|
image = 'gcr.forust.xyz/forust/session-keeper'
|
|
line = f'image: {image}@sha256:{"e" * 64}\n'
|
|
result = io.StringIO()
|
|
with patch.dict(os.environ, {'RELEASE_FILE': str(path), 'DEPLOY_SHA': 'a' * 40}):
|
|
release_module.render(io.StringIO(line), result)
|
|
self.assertEqual(result.getvalue(), line)
|
|
with self.assertRaises(ValueError):
|
|
release_module.render(io.StringIO(f'image: {image}:prod\n'), io.StringIO())
|
|
|
|
def test_unknown_image_cannot_emit_partial_manifest(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
path = Path(scratch) / 'release.json'
|
|
path.write_text(json.dumps(release()))
|
|
result = io.StringIO()
|
|
with (
|
|
patch.dict(os.environ, {'RELEASE_FILE': str(path), 'DEPLOY_SHA': 'a' * 40}),
|
|
self.assertRaisesRegex(ValueError, 'missing'),
|
|
):
|
|
release_module.render(
|
|
io.StringIO('kind: Deployment\nimage: gcr.forust.xyz/forust/unknown:prod\n'), result
|
|
)
|
|
self.assertEqual(result.getvalue(), '')
|
|
|
|
def test_only_changed_image_is_built_and_credentials_are_removed(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
root = Path(scratch)
|
|
built = []
|
|
auth_directories = []
|
|
old = release()
|
|
|
|
def fake_command(*args, **kwargs):
|
|
if args[:2] == ('git', 'rev-parse'):
|
|
return 'e' * 40
|
|
if args[:3] == ('docker', 'buildx', 'build'):
|
|
built.append(args[args.index('--file') + 1])
|
|
metadata = Path(args[args.index('--metadata-file') + 1])
|
|
metadata.write_text(json.dumps({'containerimage.digest': 'sha256:' + 'f' * 64}))
|
|
auth_directories.append(Path(kwargs['env']['DOCKER_CONFIG']))
|
|
return ''
|
|
|
|
api = Mock()
|
|
api.successful_runs.return_value = iter([{'id': 1}])
|
|
api.release.return_value = old
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{
|
|
'GITHUB_SHA': 'e' * 40,
|
|
'GITHUB_RUN_ID': '2',
|
|
'REGISTRY_USERNAME': 'test',
|
|
'REGISTRY_PASSWORD': 'placeholder',
|
|
},
|
|
),
|
|
patch.object(release_module.Path, 'home', return_value=root),
|
|
patch.object(release_module, 'Gitea', return_value=api),
|
|
patch.object(
|
|
release_module,
|
|
'fingerprint',
|
|
side_effect=lambda context, _file: ('d' if context == 'errorpages' else 'c') * 64,
|
|
),
|
|
patch.object(release_module, 'command', side_effect=fake_command),
|
|
patch.object(subprocess, 'run', return_value=subprocess.CompletedProcess([], 0)),
|
|
):
|
|
release_module.build(root / 'release.json')
|
|
self.assertEqual(built, ['errorpages/Dockerfile'])
|
|
self.assertTrue(all(not directory.exists() for directory in auth_directories))
|
|
self.assertEqual(json.loads((root / 'release.json').read_text())['sha'], 'e' * 40)
|
|
|
|
|
|
class DurableRunTests(unittest.TestCase):
|
|
def test_duplicate_start_only_reattaches(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
state = Path(scratch)
|
|
directory = state / 'runs/123-1'
|
|
directory.mkdir(parents=True)
|
|
request = {'release': release(), 'mode': 'full', 'refresh_images': False}
|
|
controller.atomic_json(directory / 'request.json', request)
|
|
controller.atomic_json(directory / 'status.json', {'state': 'running', 'stages': {}})
|
|
with (
|
|
patch.object(controller, 'STATE', state),
|
|
patch.object(controller.sys, 'stdin', io.TextIOWrapper(io.BytesIO(json.dumps(request).encode()))),
|
|
patch.object(controller, 'command') as execute,
|
|
):
|
|
controller.start('123-1')
|
|
execute.assert_not_called()
|
|
|
|
def test_failed_apply_still_verifies_and_does_not_advance_baseline(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
state = Path(scratch)
|
|
directory = state / 'runs/123-1'
|
|
directory.mkdir(parents=True)
|
|
controller.atomic_json(
|
|
directory / 'request.json', {'release': release(), 'mode': 'full', 'refresh_images': False}
|
|
)
|
|
controller.atomic_json(directory / 'status.json', {'state': 'queued', 'stages': {}})
|
|
called = []
|
|
|
|
def fake_stage(folder, name, _budget):
|
|
called.append(name)
|
|
status = json.loads((folder / 'status.json').read_text())
|
|
status['stages'][name] = {'result': 'failure' if name == 'apply-k8s' else 'success'}
|
|
controller.atomic_json(folder / 'status.json', status)
|
|
return name != 'apply-k8s'
|
|
|
|
with (
|
|
patch.object(controller, 'STATE', state),
|
|
patch.object(controller, 'make_plan', return_value={'selected': {}, 'helm': [], 'removed': []}),
|
|
patch.object(controller, 'stage', side_effect=fake_stage),
|
|
patch.object(controller, 'command', return_value='1'),
|
|
self.assertRaises(RuntimeError),
|
|
):
|
|
controller.execute('123-1')
|
|
self.assertIn('verify-k8s', called)
|
|
self.assertIn('smoke', called)
|
|
self.assertNotIn('apply-compose', called)
|
|
self.assertFalse((state / 'last-success.json').exists())
|
|
self.assertEqual(json.loads((directory / 'status.json').read_text())['state'], 'failure')
|
|
|
|
def test_recovery_finishes_baseline_after_all_stages_completed(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
state = Path(scratch)
|
|
directory = state / 'runs/123-1'
|
|
directory.mkdir(parents=True)
|
|
names = ('doctor', 'validate', 'apply-k8s', 'apply-compose', 'verify-k8s', 'smoke')
|
|
controller.atomic_json(
|
|
directory / 'status.json',
|
|
{'state': 'running', 'stages': {name: {'result': 'success'} for name in names}},
|
|
)
|
|
controller.atomic_json(directory / 'plan.json', {'sha': 'a' * 40})
|
|
with patch.object(controller, 'STATE', state), patch.object(controller, 'stage') as execute:
|
|
controller.recover(directory)
|
|
execute.assert_not_called()
|
|
self.assertEqual(json.loads((state / 'last-success.json').read_text())['run_id'], '123-1')
|
|
self.assertEqual(json.loads((directory / 'status.json').read_text())['state'], 'success')
|
|
|
|
def test_manual_recovery_retries_checks_without_repeating_apply(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
state = Path(scratch)
|
|
directory = state / 'runs/123-1'
|
|
(directory / 'snapshot').mkdir(parents=True)
|
|
(directory / 'snapshot/current').write_text('snapshot')
|
|
controller.atomic_json(
|
|
directory / 'status.json',
|
|
{
|
|
'state': 'failure',
|
|
'stages': {
|
|
'apply-k8s': {'result': 'failure'},
|
|
'verify-k8s': {'result': 'failure'},
|
|
'smoke': {'result': 'failure'},
|
|
},
|
|
},
|
|
)
|
|
called = []
|
|
|
|
def checks(folder, name, _budget):
|
|
status = json.loads((folder / 'status.json').read_text())
|
|
self.assertNotIn(name, status['stages'])
|
|
called.append(name)
|
|
status['stages'][name] = {'result': 'success'}
|
|
controller.atomic_json(folder / 'status.json', status)
|
|
return True
|
|
|
|
with patch.object(controller, 'STATE', state), patch.object(controller, 'stage', side_effect=checks):
|
|
controller.recover(directory, retry=True)
|
|
self.assertEqual(called, ['verify-k8s', 'smoke'])
|
|
self.assertFalse((state / 'last-success.json').exists())
|
|
self.assertEqual(json.loads((directory / 'status.json').read_text())['state'], 'failure')
|
|
|
|
|
|
class InstallerTests(unittest.TestCase):
|
|
def test_version_comparison_is_exact_without_network_or_host_packages(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
root = Path(scratch)
|
|
binary = root / 'bin/fake'
|
|
binary.parent.mkdir()
|
|
binary.write_text('#!/bin/sh\necho fake-v1.7.70\n')
|
|
binary.chmod(0o755)
|
|
script = """set -euo pipefail
|
|
source "$LIB"
|
|
if at_version fake 1.7.7; then exit 1; fi
|
|
at_version fake 1.7.70
|
|
"""
|
|
subprocess.run( # noqa: S603
|
|
['/usr/bin/bash', '-c', script],
|
|
check=True,
|
|
env={**os.environ, 'TOOLS_DIR': str(root), 'LIB': str(ROOT / '.gitea/workflows/install-ci-tools.sh')},
|
|
)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|