268 lines
12 KiB
Python
268 lines
12 KiB
Python
"""CI gate, selection, persistent configuration and recovery regression tests."""
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def module(name, filename):
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / '.gitea/workflows' / filename)
|
|
loaded = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(loaded)
|
|
return loaded
|
|
|
|
|
|
release_module = module('release_test', 'release.py')
|
|
planner = module('plan_test', 'deploy-plan.py')
|
|
compose_module = module('compose_test', 'compose-release.py')
|
|
controller = module('controller_test', 'deploy-controller.py')
|
|
|
|
|
|
def release(sha='a' * 40):
|
|
return {
|
|
'version': 1,
|
|
'sha': sha,
|
|
'images': {f'gcr.forust.xyz/forust/{name}': 'sha256:' + 'b' * 64 for name in release_module.IMAGES},
|
|
'inputs': {f'gcr.forust.xyz/forust/{name}': 'c' * 64 for name in release_module.IMAGES},
|
|
}
|
|
|
|
|
|
class ReleaseGateTests(unittest.TestCase):
|
|
def test_release_rejects_wrong_sha_missing_images_and_mutable_tags(self):
|
|
for mutation in ('sha', 'missing', 'tag'):
|
|
data = release()
|
|
if mutation == 'sha':
|
|
data['sha'] = 'd' * 40
|
|
elif mutation == 'missing':
|
|
data['images'].pop(next(iter(data['images'])))
|
|
else:
|
|
data['images'][next(iter(data['images']))] = 'prod'
|
|
with self.assertRaises(ValueError):
|
|
release_module.validate_release(data, 'a' * 40)
|
|
|
|
def test_gate_excludes_pr_wrong_branch_and_failed_runs(self):
|
|
api = object.__new__(release_module.Gitea)
|
|
api.repository = 'forust/homelab'
|
|
good = {
|
|
'id': 1,
|
|
'status': 'completed',
|
|
'conclusion': 'success',
|
|
'head_branch': 'main',
|
|
'event': 'push',
|
|
'head_sha': 'a' * 40,
|
|
'repository': {'full_name': api.repository},
|
|
}
|
|
entries = [
|
|
good,
|
|
{**good, 'event': 'pull_request'},
|
|
{**good, 'head_branch': 'dev'},
|
|
{**good, 'conclusion': 'failure'},
|
|
{**good, 'head_sha': 'b' * 40},
|
|
{**good, 'head_repository': {'full_name': 'attacker/fork'}},
|
|
]
|
|
with patch.object(api, 'pages', return_value=iter(entries)):
|
|
self.assertEqual(list(api.successful_runs('a' * 40)), [good])
|
|
|
|
def test_green_ci_with_skipped_build_is_rejected(self):
|
|
api = object.__new__(release_module.Gitea)
|
|
with (
|
|
patch.object(api, 'pages', return_value=iter([{'name': 'build', 'conclusion': 'skipped'}])),
|
|
self.assertRaisesRegex(ValueError, 'build job'),
|
|
):
|
|
api.release({'id': 1, 'head_sha': 'a' * 40})
|
|
|
|
def test_expired_or_ambiguous_artifacts_are_rejected(self):
|
|
api = object.__new__(release_module.Gitea)
|
|
api.base = 'https://example.test/api/v1/repos/a/b'
|
|
artifact = {'id': 1, 'name': 'release-' + 'a' * 40}
|
|
for artifacts in ([{**artifact, 'expired': True}], [artifact, artifact], []):
|
|
with (
|
|
patch.object(api, 'pages', return_value=iter([{'name': 'build', 'conclusion': 'success'}])),
|
|
patch.object(api, 'request', return_value={'artifacts': artifacts}),
|
|
self.assertRaisesRegex(ValueError, 'artifact'),
|
|
):
|
|
api.release({'id': 1, 'head_sha': 'a' * 40})
|
|
|
|
|
|
class SelectionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.scratch = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.scratch.cleanup)
|
|
self.repo = Path(self.scratch.name)
|
|
self.git('init', '-q')
|
|
self.git('config', 'user.email', 'test@example.test')
|
|
self.git('config', 'user.name', 'CI Test')
|
|
for service in ('one', 'two', 'postgres'):
|
|
directory = self.repo / service / 'k8s'
|
|
directory.mkdir(parents=True)
|
|
(directory / 'active').touch()
|
|
(directory / 'app.yaml').write_text('kind: Deployment\n')
|
|
(self.repo / '.gitea/workflows').mkdir(parents=True)
|
|
(self.repo / '.gitea/workflows/deploy-lib.sh').write_text('HELM_RELEASES=(\n)\n')
|
|
(self.repo / '.gitea/deploy-dependencies.json').write_text('{"postgres": ["one", "two"]}')
|
|
self.sha = self.commit()
|
|
self.initial = planner.make_plan(self.repo, self.repo, release(self.sha), None, 'full', [])
|
|
|
|
def git(self, *args):
|
|
return planner.output('git', '-C', str(self.repo), *args)
|
|
|
|
def commit(self):
|
|
self.git('add', '.')
|
|
self.git('commit', '-qm', 'Test state')
|
|
return self.git('rev-parse', 'HEAD')
|
|
|
|
def test_first_changed_deploy_requires_explicit_full(self):
|
|
with self.assertRaisesRegex(ValueError, 'full'):
|
|
planner.make_plan(self.repo, self.repo, release(self.sha), None, 'changed', [])
|
|
|
|
def test_only_changed_service_is_selected(self):
|
|
(self.repo / 'one/k8s/app.yaml').write_text('kind: StatefulSet\n')
|
|
result = planner.make_plan(self.repo, self.repo, release(self.commit()), self.initial, 'changed', [])
|
|
self.assertEqual(result['selected']['k8s'], ['one'])
|
|
self.assertEqual(result['helm'], [])
|
|
|
|
def test_failed_intermediate_deploy_does_not_lose_changes(self):
|
|
(self.repo / 'one/k8s/app.yaml').write_text('kind: StatefulSet\n')
|
|
self.commit() # This commit failed deploy: baseline must remain initial.
|
|
(self.repo / 'two/k8s/app.yaml').write_text('kind: StatefulSet\n')
|
|
result = planner.make_plan(self.repo, self.repo, release(self.commit()), self.initial, 'changed', [])
|
|
self.assertEqual(result['selected']['k8s'], ['one', 'two'])
|
|
|
|
def test_dependencies_and_removals_are_reported(self):
|
|
(self.repo / 'postgres/k8s/app.yaml').write_text('kind: StatefulSet\n')
|
|
(self.repo / 'two/k8s/active').unlink()
|
|
result = planner.make_plan(self.repo, self.repo, release(self.commit()), self.initial, 'changed', [])
|
|
self.assertEqual(result['selected']['k8s'], ['one', 'postgres'])
|
|
self.assertIn('two', result['removed'])
|
|
|
|
def test_local_configuration_change_selects_service(self):
|
|
(self.repo / 'one/.env').write_text('TEST_VALUE=changed\n')
|
|
result = planner.make_plan(self.repo, self.repo, release(self.sha), self.initial, 'changed', [])
|
|
self.assertEqual(result['selected']['k8s'], ['one'])
|
|
|
|
def test_redeploy_is_noop_and_full_includes_all(self):
|
|
result = planner.make_plan(self.repo, self.repo, release(self.sha), self.initial, 'changed', [])
|
|
self.assertEqual(result['selected']['k8s'], [])
|
|
result = planner.make_plan(self.repo, self.repo, release(self.sha), self.initial, 'full', [])
|
|
self.assertEqual(result['selected']['k8s'], ['one', 'postgres', 'two'])
|
|
|
|
|
|
class ComposeConfigurationTests(unittest.TestCase):
|
|
def test_pin_preserves_project_volumes_paths_and_previous_image(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
root = Path(scratch)
|
|
run = root / 'run'
|
|
source = run / 'source'
|
|
config_repo = root / 'persistent'
|
|
(source / 'headscale').mkdir(parents=True)
|
|
config_repo.mkdir()
|
|
(run / 'release.json').write_text(json.dumps(release()))
|
|
old = 'busybox@sha256:' + 'd' * 64
|
|
new = 'busybox@sha256:' + 'e' * 64
|
|
config = {
|
|
'name': 'headscale',
|
|
'services': {
|
|
'app': {
|
|
'image': 'busybox:latest',
|
|
'volumes': [
|
|
{'type': 'bind', 'source': str(config_repo / 'headscale/config.yaml'), 'target': '/config'},
|
|
{'type': 'volume', 'source': 'data', 'target': '/data'},
|
|
],
|
|
}
|
|
},
|
|
'volumes': {'data': {'name': 'headscale_data'}},
|
|
}
|
|
|
|
def fake_output(*args, **kwargs):
|
|
if args[:2] == ('docker', 'compose'):
|
|
self.assertEqual(kwargs['cwd'], config_repo)
|
|
self.assertIn(str(config_repo / 'headscale'), args)
|
|
return json.dumps(config)
|
|
if args[:2] == ('docker', 'ps'):
|
|
return 'container'
|
|
if args[:2] == ('docker', 'inspect'):
|
|
return 'sha256:' + 'f' * 64
|
|
return json.dumps([old])
|
|
|
|
with (
|
|
patch.dict(os.environ, {'CONFIG_REPO': str(config_repo), 'REPO': str(source), 'RUN_DIR': str(run)}),
|
|
patch.object(compose_module, 'output', side_effect=fake_output),
|
|
patch.object(compose_module, 'resolve', return_value=new),
|
|
):
|
|
compose_module.prepare(source / 'headscale/compose.yaml')
|
|
pinned = json.loads((run / 'compose/headscale.json').read_text())
|
|
before = json.loads((run / 'compose-before/headscale.json').read_text())
|
|
self.assertEqual(pinned['name'], 'headscale')
|
|
self.assertEqual(pinned['volumes'], config['volumes'])
|
|
self.assertEqual(pinned['services']['app']['volumes'], config['services']['app']['volumes'])
|
|
self.assertEqual(pinned['services']['app']['image'], new)
|
|
self.assertEqual(before['services']['app']['image'], old)
|
|
self.assertEqual((run / 'compose/headscale.json').stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_registry_index_and_single_image_descriptors(self):
|
|
for digest in ('a' * 64, 'b' * 64):
|
|
with patch.object(compose_module, 'output', return_value=json.dumps({'digest': 'sha256:' + digest})):
|
|
self.assertEqual(
|
|
compose_module.resolve('registry.test:5000/repo:latest'), f'registry.test:5000/repo@sha256:{digest}'
|
|
)
|
|
|
|
|
|
class ControllerTests(unittest.TestCase):
|
|
def test_completed_stage_cannot_apply_again(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
directory = Path(scratch)
|
|
controller.atomic_json(
|
|
directory / 'status.json', {'state': 'success', 'stages': {'apply-k8s': {'result': 'success'}}}
|
|
)
|
|
with patch.object(subprocess, 'run') as execute:
|
|
self.assertTrue(controller.stage(directory, 'apply-k8s', 60))
|
|
execute.assert_not_called()
|
|
|
|
def test_run_id_is_not_shell_or_path_input(self):
|
|
for invalid in ('../123', '-1', '1;touch bad', 'abc', '1/2'):
|
|
with self.assertRaises(ValueError):
|
|
controller.run_directory(invalid)
|
|
|
|
def test_exact_previous_revision_is_used_for_rollback(self):
|
|
with tempfile.TemporaryDirectory() as scratch:
|
|
root = Path(scratch)
|
|
(root / 'current').write_text(str(root))
|
|
(root / 'revisions.json').write_text(
|
|
json.dumps([{'kind': 'deployment', 'namespace': 'app', 'name': 'web', 'uid': 'same', 'revision': 7}])
|
|
)
|
|
(root / 'failed').write_text('deployment app web\n')
|
|
script = """set -euo pipefail
|
|
source "$LIB"
|
|
kubectl() {
|
|
case "$*" in
|
|
*metadata.annotations*) printf '{}' ;;
|
|
*metadata.uid*) printf same ;;
|
|
'rollout undo'*) printf '%s\\n' "$*" >>"$CALLS" ;;
|
|
'rollout status'*) return 0 ;;
|
|
*) return 1 ;;
|
|
esac
|
|
}
|
|
rollback_workloads "$FAILED"
|
|
"""
|
|
env = {
|
|
**os.environ,
|
|
'REPO': str(ROOT),
|
|
'LIB': str(ROOT / '.gitea/workflows/deploy-lib.sh'),
|
|
'DEPLOY_SNAPSHOT_DIR': str(root),
|
|
'CALLS': str(root / 'calls'),
|
|
'FAILED': str(root / 'failed'),
|
|
}
|
|
subprocess.run(['/usr/bin/bash', '-c', script], env=env, check=True) # noqa: S603
|
|
self.assertIn('--to-revision=7', (root / 'calls').read_text())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|