homelab/tests/test_image_matrix.py

"""Matrix release contracts and failure gates without a registry."""

import json
import os
import tempfile
from pathlib import Path
from unittest.mock import Mock, call, patch

from ci_test_case import IsolatedCITestCase
from test_cicd import release, release_module


def plan_data(changed):
    targets = []
    for name, (context, dockerfile) in release_module.IMAGES.items():
        targets.append(
            {
                'name': name,
                'image': f'gcr.forust.xyz/forust/{name}',
                'context': context,
                'dockerfile': dockerfile,
                'inputs': ('d' if name in changed else 'c') * 64,
                'reuse_digest': None if name in changed else 'sha256:' + 'b' * 64,
            }
        )
    return {'sha': 'a' * 40, 'targets': targets}


class MatrixTests(IsolatedCITestCase):
    def test_no_change_one_image_all_images_and_missing_baseline(self):
        for changed in (set(), {'error-pages'}, set(release_module.IMAGES)):
            with self.subTest(changed=changed), tempfile.TemporaryDirectory() as scratch:
                api = Mock()
                api.successful_runs.return_value = iter([{'id': 1}])
                api.release.return_value = release()
                expected = plan_data(changed)
                fingerprints = {t['dockerfile']: t['inputs'] for t in expected['targets']}
                output = Path(scratch) / 'plan.json'
                with (
                    patch.dict(os.environ, {'GITHUB_SHA': 'a' * 40, 'GITHUB_RUN_ID': '2'}),
                    patch.object(release_module, 'command', return_value='a' * 40),
                    patch.object(release_module, 'Gitea', return_value=api),
                    patch.object(
                        release_module, 'fingerprint', side_effect=lambda _c, f, mapping=fingerprints: mapping[f]
                    ),
                ):
                    release_module.prepare_images(output)
                self.assertEqual(json.loads(output.read_text()), expected)
        api.successful_runs.return_value = iter([])
        with (
            tempfile.TemporaryDirectory() as scratch,
            patch.dict(os.environ, {'GITHUB_SHA': 'a' * 40}),
            patch.object(release_module, 'command', return_value='a' * 40),
            patch.object(release_module, 'Gitea', return_value=api),
            patch.object(release_module, 'fingerprint', return_value='c' * 64),
        ):
            output = Path(scratch) / 'plan.json'
            release_module.prepare_images(output)
            self.assertTrue(all(t['reuse_digest'] is None for t in json.loads(output.read_text())['targets']))

    def test_incomplete_or_wrong_sha_fragments_cannot_publish_tags(self):
        for wrong_sha in (False, True):
            with self.subTest(wrong_sha=wrong_sha), tempfile.TemporaryDirectory() as scratch:
                root = Path(scratch)
                plan = root / 'plan.json'
                plan.write_text(json.dumps(plan_data(set())))
                for name in release_module.IMAGES:
                    if name == 'xdfnx-homepage' and not wrong_sha:
                        continue
                    folder = root / f'image-{name}'
                    folder.mkdir()
                    image = f'gcr.forust.xyz/forust/{name}'
                    folder.joinpath('image.json').write_text(
                        json.dumps(
                            {
                                'version': 1,
                                'sha': ('e' if wrong_sha else 'a') * 40,
                                'images': {image: 'sha256:' + 'b' * 64},
                                'inputs': {image: 'c' * 64},
                            }
                        )
                    )
                with (
                    patch.object(release_module, 'checked_plan', return_value=plan_data(set())),
                    patch.object(release_module.subprocess, 'run') as execute,
                    self.assertRaises((ValueError, FileNotFoundError)),
                ):
                    release_module.finalize_images(root / 'release.json', root, plan)
                execute.assert_not_called()
                self.assertFalse((root / 'release.json').exists())

    def test_manifest_only_release_pins_each_successful_digest(self):
        with tempfile.TemporaryDirectory() as scratch:
            root = Path(scratch)
            data = plan_data(set())
            for target in data['targets']:
                folder = root / f'image-{target["name"]}'
                folder.mkdir()
                image = target['image']
                folder.joinpath('image.json').write_text(
                    json.dumps(
                        {
                            'version': 1,
                            'sha': data['sha'],
                            'images': {image: target['reuse_digest']},
                            'inputs': {image: target['inputs']},
                        }
                    )
                )
            with (
                patch.object(release_module, 'checked_plan', return_value=data),
                patch.dict(os.environ, {'REGISTRY_USERNAME': 'test', 'REGISTRY_PASSWORD': 'placeholder'}),
                patch.object(release_module.subprocess, 'run'),
                patch.object(release_module, 'command') as execute,
            ):
                release_module.finalize_images(root / 'release.json', root, root / 'plan.json')
            self.assertEqual(
                [entry.args for entry in execute.call_args_list],
                [
                    call(
                        'docker',
                        'buildx',
                        'imagetools',
                        'create',
                        '--prefer-index=false',
                        '--tag',
                        f'{t["image"]}:sha-{data["sha"]}',
                        f'{t["image"]}@{t["reuse_digest"]}',
                    ).args
                    for t in data['targets']
                ],
            )
            self.assertEqual(json.loads((root / 'release.json').read_text()), release())

    def test_checkout_mismatch_cannot_build(self):
        with tempfile.TemporaryDirectory() as scratch:
            plan = Path(scratch) / 'plan.json'
            plan.write_text(json.dumps(plan_data(set())))
            with (
                patch.dict(os.environ, {'GITHUB_SHA': 'e' * 40}),
                patch.object(release_module, 'command', return_value='a' * 40),
                self.assertRaisesRegex(ValueError, 'source commit'),
            ):
                release_module.checked_plan(plan)