"""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()