forust
95e4d8f875
ci / Compose (push) Skipped
ci / Workflows (push) Skipped
ci / Shell (push) Skipped
ci / Formatting (push) Skipped
ci / Python and tests (push) Skipped
ci / YAML (push) Skipped
ci / Dockerfiles (push) Skipped
ci / Kubernetes (push) Skipped
ci / Compose (pull_request) Successful in 11s
ci / Workflows (pull_request) Successful in 7s
ci / Shell (pull_request) Successful in 15s
ci / Formatting (pull_request) Successful in 15s
ci / Python and tests (pull_request) Successful in 5s
ci / YAML (pull_request) Successful in 7s
ci / Dockerfiles (pull_request) Successful in 5s
ci / Kubernetes (pull_request) Successful in 6s
ci / image-plan (pull_request) Skipped
ci / Image (${{ matrix.name }}) (pull_request) Skipped
ci / build (pull_request) Skipped
renovate-ci / validate-renovate (pull_request) Successful in 11s
286 lines
13 KiB
Python
286 lines
13 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_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)),
|
|
):
|
|
plan = root / 'plan.json'
|
|
release_module.prepare_images(plan)
|
|
for name in release_module.IMAGES:
|
|
release_module.build(root / f'{name}.json', name, plan)
|
|
self.assertEqual(built, ['errorpages/Dockerfile'])
|
|
self.assertTrue(all(not directory.exists() for directory in auth_directories))
|
|
self.assertEqual(json.loads((root / 'error-pages.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 FailureSummaryTests(unittest.TestCase):
|
|
def test_build_failure_keeps_progress_and_does_not_expose_exception_text(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
summary = Path(scratch) / 'summary.md'
|
|
|
|
def failed_build(_output, report, _name, _plan):
|
|
report.update(phase='Build or reuse xdfnx-homepage', current='xdfnx-homepage', built=['error-pages'])
|
|
report['images']['gcr.forust.xyz/forust/error-pages'] = 'sha256:' + 'b' * 64
|
|
raise RuntimeError('private value must not appear in the summary')
|
|
|
|
with (
|
|
patch.dict(os.environ, {'GITHUB_STEP_SUMMARY': str(summary), 'GITHUB_SHA': 'a' * 40}),
|
|
patch.object(release_module, 'build_images', side_effect=failed_build),
|
|
self.assertRaises(RuntimeError),
|
|
):
|
|
release_module.build(Path(scratch) / 'release.json', 'xdfnx-homepage', Path('plan.json'))
|
|
content = summary.read_text()
|
|
self.assertIn('**failure**', content)
|
|
self.assertIn('error-pages', content)
|
|
self.assertIn('xdfnx-homepage', content)
|
|
self.assertNotIn('private value', content)
|
|
|
|
def test_deploy_failure_reports_completed_apply_and_rollback_result(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
state = Path(scratch)
|
|
directory = state / 'runs/123-1'
|
|
snapshot = directory / 'snapshot/before'
|
|
snapshot.mkdir(parents=True)
|
|
(directory / 'snapshot/current').write_text(str(snapshot))
|
|
(snapshot / 'failed-workloads').write_text('deployment app api\nROLLED_BACK=1\nUNRECOVERED=0\n')
|
|
controller.atomic_json(
|
|
directory / 'request.json', {'release': release(), 'mode': 'changed', 'refresh_images': False}
|
|
)
|
|
controller.atomic_json(
|
|
directory / 'status.json',
|
|
{
|
|
'state': 'failure',
|
|
'stages': {
|
|
'apply-k8s': {'result': 'success', 'exit_code': 0},
|
|
'verify-k8s': {'result': 'failure', 'exit_code': 1},
|
|
},
|
|
},
|
|
)
|
|
controller.atomic_json(directory / 'plan.json', {'selected': {'k8s': ['app'], 'compose': []}})
|
|
(directory / 'apply-events.jsonl').write_text(
|
|
json.dumps({'action': 'kubectl', 'target': 'app/k8s/api.yaml', 'result': 'success'}) + '\n'
|
|
)
|
|
output = io.StringIO()
|
|
with patch.object(controller, 'STATE', state), patch('sys.stdout', output):
|
|
controller.summary('123-1')
|
|
content = output.getvalue()
|
|
self.assertIn('verify-k8s | failure | 1', content)
|
|
self.assertIn('app/k8s/api.yaml', content)
|
|
self.assertIn('Workloads restored: **1**', content)
|
|
self.assertIn('manual recovery: **0**', content)
|
|
self.assertIn('Compose requires manual recovery', content)
|
|
|
|
|
|
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()
|