cloud-apache-container/scripts/tests/multiarch-workflow.test.py

#!/usr/bin/env python3
"""Static wiring checks for the multi-arch publish jobs in .gitea/workflows/build-push.yaml
(Build-and-Push = cac, Build-FPM-Images = cac-fpm, Build-Shared-httpd = shared-httpd).

No YAML library (the Shell-Checks runner has none): jobs/steps are split on their indentation.
Env WORKFLOW overrides the file under test (used to show the tests fail on the old workflow).
Run: python3 -I scripts/tests/multiarch-workflow.test.py
"""
import os, re, subprocess, tempfile, unittest, stat

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.normpath(os.path.join(HERE, '..', '..'))
WF = os.environ.get('WORKFLOW') or os.path.join(ROOT, '.gitea', 'workflows', 'build-push.yaml')
with open(WF) as _f:
    TEXT = _f.read()
REG = 'repo.anhonesthost.net/cloud-hosting-platform'

SLICE = {  # job -> (image repo, dockerfile arg or None, matrix?)
    'Build-and-Push': ('cac', None, True),
    'Build-FPM-Images': ('cac-fpm', './Dockerfile.fpm', True),
    'Build-Shared-httpd': ('shared-httpd', './Dockerfile.shared-httpd', False),
}
OUT_OF_SLICE = ['Build-LiteSpeed-Images', 'Build-LSPHP-Images', 'Build-Shared-OLS']


def jobs():
    m = re.search(r'^jobs:\n', TEXT, re.M)
    body = TEXT[m.end():]
    parts = re.split(r'^  ([A-Za-z0-9_-]+):\n', body, flags=re.M)
    return {parts[i]: parts[i + 1] for i in range(1, len(parts), 2)}


def steps(job):
    s = re.split(r'^      - name: ', job, flags=re.M)[1:]
    out = []
    for blk in s:
        name = blk.split('\n', 1)[0].strip()
        out.append((name, blk))
    return out


def step(job, prefix):
    hits = [(n, b) for n, b in steps(job) if n.startswith(prefix)]
    assert len(hits) == 1, (prefix, [n for n, _ in hits])
    return hits[0][1]


class Wiring(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.jobs = jobs()

    def test_triggers_unchanged(self):
        head = TEXT.split('jobs:')[0]
        self.assertRegex(head, r'on:\n  push:\n    branches:\n      - trunk\n  workflow_dispatch:')

    def test_slice_jobs_exist_and_keep_matrix_and_no_fail_fast(self):
        for j, (_, _, matrix) in SLICE.items():
            self.assertIn(j, self.jobs, j)
            if matrix:
                self.assertIn('fail-fast: false', self.jobs[j], j)
                self.assertRegex(self.jobs[j], r'phpver: \$\{\{ fromJSON\(needs\.plan\.outputs\.\w+\) \}\}', j)

    def test_job_timeout_bounds_the_arm64_step(self):
        for j in SLICE:
            job = self.jobs[j]
            jt = int(re.search(r'^    timeout-minutes: (\d+)$', job, re.M).group(1))
            at = int(re.search(r'timeout-minutes: (\d+)', step(job, 'Build arm64')).group(1))
            self.assertGreaterEqual(at, 20, j)            # T3: arm64 under qemu took 7m47s
            self.assertGreater(jt, at + 20, j)            # room for setup + amd64 + assemble
            self.assertLessEqual(jt, 120, j)

    def test_step_order_and_ids(self):
        for j in SLICE:
            names = [n for n, _ in steps(self.jobs[j])]
            a = [i for i, n in enumerate(names) if n.startswith('Build amd64')]
            r = [i for i, n in enumerate(names) if n.startswith('Build arm64')]
            s = [i for i, n in enumerate(names) if n.startswith('Assemble')]
            self.assertEqual((len(a), len(r), len(s)), (1, 1, 1), j)
            self.assertLess(a[0], r[0], j)
            self.assertLess(r[0], s[0], j)
            self.assertIn('id: amd64', step(self.jobs[j], 'Build amd64'))
            self.assertIn('id: arm64', step(self.jobs[j], 'Build arm64'))
            self.assertLess(names.index('Login to Gitea'), a[0], j)

    def test_arm64_non_fatal_amd64_fatal(self):
        for j in SLICE:
            self.assertIn('continue-on-error: true', step(self.jobs[j], 'Build arm64'), j)
            self.assertNotIn('continue-on-error', step(self.jobs[j], 'Build amd64'), j)
            self.assertNotIn('continue-on-error', step(self.jobs[j], 'Assemble'), j)

    def test_platforms_and_qemu(self):
        for j in SLICE:
            self.assertIn('platforms: linux/amd64\n', step(self.jobs[j], 'Build amd64'), j)
            self.assertIn('platforms: linux/arm64\n', step(self.jobs[j], 'Build arm64'), j)
            self.assertRegex(step(self.jobs[j], 'Set up QEMU'), r'platforms: arm64', j)

    def test_both_push_by_digest_untagged_to_gitea_only(self):
        for j, (repo, _, _) in SLICE.items():
            for which in ('Build amd64', 'Build arm64'):
                st = step(self.jobs[j], which)
                self.assertIn('type=image,name=%s/%s,push-by-digest=true,name-canonical=true,push=true' % (REG, repo), st, j)
                self.assertNotRegex(st, r'^\s+tags:', j)
                self.assertNotRegex(st, r'^\s+push: true', j)
            self.assertNotIn('ghcr.io', self.jobs[j], j)   # CAC has never pushed to GHCR; do not widen

    def test_same_build_args_and_dockerfile_for_both_platforms(self):
        for j, (_, dockerfile, _) in SLICE.items():
            a = step(self.jobs[j], 'Build amd64'); r = step(self.jobs[j], 'Build arm64')
            ba = lambda s: re.search(r'build-args: \|\n((?:            .*\n)+)', s).group(1)
            self.assertEqual(ba(a), ba(r), j)
            for need in (('PKG_CACHE=${{ steps.pkgcache.outputs.url }}', 'BASE_IMAGE=${{ steps.base.outputs.image }}')):
                for n in need:
                    self.assertIn(n, ba(a), j)
            if SLICE[j][2]:
                for n in ('PHPVER=${{ matrix.phpver }}', 'CACHE_BUST=${{ steps.bust.outputs.value }}',
                          'EXPECT_PHP_MIN=${{ steps.phpmin.outputs.min }}'):
                    self.assertIn(n, ba(a), j)
            if dockerfile:
                self.assertIn('file: %s\n' % dockerfile, a, j)
                self.assertIn('file: %s\n' % dockerfile, r, j)

    def test_assemble_uses_this_runs_outputs_and_repo(self):
        for j, (repo, _, _) in SLICE.items():
            st = step(self.jobs[j], 'Assemble')
            self.assertIn('${{ steps.amd64.outputs.digest }}', st, j)
            self.assertIn('${{ steps.arm64.outputs.digest }}', st, j)
            self.assertIn('${{ steps.arm64.outcome }}', st, j)
            self.assertIn('scripts/assemble-multiarch-index.sh', st, j)
            self.assertIn('--repo %s/%s' % (REG, repo), st, j)
            self.assertEqual(st.count('--repo '), 1, j)

    def test_tag_scheme_unchanged(self):
        """Run the assemble step's shell with a recording stub: exact tags per version."""
        for j, (repo, _, matrix) in SLICE.items():
            st = step(self.jobs[j], 'Assemble')
            script = re.search(r'run: \|\n((?:          .*\n?)+)', st).group(1)
            script = '\n'.join(l[10:] for l in script.split('\n'))
            for ver in (['82', '83', '84', '85'] if matrix else ['']):
                with tempfile.TemporaryDirectory() as d:
                    os.makedirs(os.path.join(d, 'scripts'))
                    stub = os.path.join(d, 'scripts', 'assemble-multiarch-index.sh')
                    with open(stub, 'w') as f:
                        f.write('#!/usr/bin/env bash\nprintf "%s\\n" "$@" > "$LOG"\n')
                    os.chmod(stub, 0o755)
                    log = os.path.join(d, 'log')
                    env = dict(os.environ, LOG=log, AMD64_DIGEST='sha256:' + 'a' * 64,
                               ARM64_DIGEST='sha256:' + 'b' * 64, ARM64_OUTCOME='success', PHPVER=ver)
                    subprocess.run(['bash', '-c', script], cwd=d, env=env, check=True)
                    with open(log) as f:
                        args = f.read().split('\n')[:-1]
                    tags = [args[i + 1] for i, x in enumerate(args) if x == '--tag']
                    want = ['latest'] if not matrix else (['php' + ver] + (['latest'] if ver == '85' else []))
                    self.assertEqual(tags, want, (j, ver))
                    self.assertEqual(args[args.index('--repo') + 1], '%s/%s' % (REG, repo))
                    self.assertEqual(args[args.index('--arm64-outcome') + 1], 'success')

    def test_arm64_never_added_outside_the_first_slice(self):
        for j in OUT_OF_SLICE:
            self.assertIn(j, self.jobs, j)
            job = self.jobs[j]
            self.assertIn('platforms: linux/amd64', job, j)
            self.assertNotIn('linux/arm64', job, j)
            self.assertNotIn('push-by-digest', job, j)
            self.assertNotIn('assemble-multiarch-index', job, j)

    def test_only_the_assemble_script_creates_tags_in_slice(self):
        for j in SLICE:
            self.assertNotIn('imagetools create', self.jobs[j], j)   # only via the tested script


# ---- single-leg dispatch (workflow_dispatch input `only`, plan job) ----------------------
PLAN = os.path.join(ROOT, '.gitea', 'scripts', 'plan-legs.sh')
FULL = '[82,83,84,85]'
# image job -> (plan output, matrix?)
LEGS = {
    'Build-and-Push': ('cac', True),
    'Build-FPM-Images': ('cac_fpm', True),
    'Build-LiteSpeed-Images': ('cac_litespeed', True),
    'Build-LSPHP-Images': ('cac_lsphp', True),
    'Build-Shared-httpd': ('shared_httpd', False),
    'Build-Shared-OLS': ('shared_ols', False),
}
EMPTY = {'cac': '[]', 'cac_fpm': '[]', 'cac_litespeed': '[]', 'cac_lsphp': '[]',
         'shared_httpd': 'false', 'shared_ols': 'false'}
EVERYTHING = {'cac': FULL, 'cac_fpm': FULL, 'cac_litespeed': FULL, 'cac_lsphp': FULL,
              'shared_httpd': 'true', 'shared_ols': 'true'}


def plan(only):
    """Run plan-legs.sh like the runner does. Returns (exit code, outputs dict, stderr)."""
    with tempfile.TemporaryDirectory() as d:
        out = os.path.join(d, 'out')
        open(out, 'w').close()
        env = dict(os.environ, GITHUB_OUTPUT=out)
        if only is None:
            env.pop('ONLY', None)       # push event: the input does not exist
        else:
            env['ONLY'] = only
        r = subprocess.run(['bash', PLAN], env=env, capture_output=True, text=True, timeout=30)
        with open(out) as f:
            kv = dict(l.rstrip('\n').split('=', 1) for l in f if '=' in l)
        return r.returncode, kv, r.stderr


class SingleLegDispatch(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.jobs = jobs()

    def test_trigger_still_trunk_push_only_and_dispatch_has_string_input(self):
        head = TEXT.split('jobs:')[0]
        self.assertEqual(re.findall(r'^    branches:\n((?:      - .*\n)+)', head, re.M), ['      - trunk\n'])
        m = re.search(r'^  workflow_dispatch:\n    inputs:\n      only:\n((?:        .*\n)+)', head, re.M)
        self.assertIsNotNone(m, 'workflow_dispatch needs an `only` input')
        self.assertIn("default: ''", m.group(1))
        self.assertIn('required: false', m.group(1))
        self.assertIn('type: string', m.group(1))

    def test_plan_job_wiring(self):
        plan_job = self.jobs['plan']
        # input goes through env, never interpolated into the shell script
        self.assertRegex(plan_job, r'env:\n          ONLY: \$\{\{ github\.event\.inputs\.only \}\}')
        run = re.search(r'run: (.*)\n', step(plan_job, 'Plan legs')).group(1)
        self.assertNotIn('${{', run)
        self.assertIn('.gitea/scripts/plan-legs.sh', run)
        for _, (out, _) in LEGS.items():
            self.assertIn('      %s: ${{ steps.plan.outputs.%s }}' % (out, out), plan_job, out)
        self.assertIn('id: plan', step(plan_job, 'Plan legs'))

    def test_every_image_job_depends_on_plan_with_its_own_output(self):
        for j, (out, matrix) in LEGS.items():
            job = self.jobs[j]
            self.assertRegex(job, r'(?m)^    needs: plan$', j)
            want = ("if: needs.plan.outputs.%s != '[]'" if matrix else "if: needs.plan.outputs.%s == 'true'") % out
            self.assertRegex(job, r'(?m)^    ' + re.escape(want) + r'$', j)
            if matrix:
                self.assertIn('phpver: ${{ fromJSON(needs.plan.outputs.%s) }}' % out, job, j)
            self.assertNotRegex(job, r'if:.*matrix\.', j)   # job-level if cannot see matrix
            self.assertNotRegex(job, r'phpver: \[', j)      # no literal list left behind
        outs = [o for o, _ in LEGS.values()]
        self.assertEqual(len(outs), len(set(outs)))

    def test_every_job_but_shell_checks_and_plan_is_gated(self):
        for j in self.jobs:
            if j in ('Shell-Checks', 'plan'):
                continue
            self.assertIn(j, LEGS, 'new job %s must be gated by the plan job' % j)
        sc = self.jobs['Shell-Checks']
        self.assertNotRegex(sc, r'^    (needs|if):', 'Shell-Checks must stay ungated')
        self.assertIn('.gitea/scripts/plan-legs.sh', sc)               # shellchecked
        self.assertIn('multiarch-workflow.test.py', sc)                # this suite still runs

    def test_plan_outputs_match_what_the_script_emits(self):
        rc, kv, _ = plan('')
        self.assertEqual(rc, 0)
        self.assertEqual(sorted(kv), sorted(o for o, _ in LEGS.values()))

    def test_push_and_empty_input_plan_everything(self):
        for only in (None, '', '   ', '\t'):
            rc, kv, err = plan(only)
            self.assertEqual((rc, kv), (0, EVERYTHING), (only, err))

    def test_full_matrix_equals_the_old_literal(self):
        # the pre-change matrix was `phpver: [82, 83, 84, 85]`
        rc, kv, _ = plan(None)
        self.assertEqual([int(x) for x in kv['cac'].strip('[]').split(',')], [82, 83, 84, 85])

    def test_single_leg(self):
        rc, kv, _ = plan('cac:php85')
        self.assertEqual((rc, kv), (0, dict(EMPTY, cac='[85]')))
        rc, kv, _ = plan('cac-fpm:php84')
        self.assertEqual((rc, kv), (0, dict(EMPTY, cac_fpm='[84]')))
        rc, kv, _ = plan('cac-litespeed:php82')
        self.assertEqual((rc, kv), (0, dict(EMPTY, cac_litespeed='[82]')))
        rc, kv, _ = plan('cac-lsphp:php83')
        self.assertEqual((rc, kv), (0, dict(EMPTY, cac_lsphp='[83]')))
        rc, kv, _ = plan('shared-httpd')
        self.assertEqual((rc, kv), (0, dict(EMPTY, shared_httpd='true')))
        rc, kv, _ = plan('shared-ols')
        self.assertEqual((rc, kv), (0, dict(EMPTY, shared_ols='true')))

    def test_cac_does_not_pull_in_cac_fpm(self):
        rc, kv, _ = plan('cac:php85')
        self.assertEqual(kv['cac_fpm'], '[]')
        self.assertEqual(kv['cac_litespeed'], '[]')

    def test_family_name_means_every_version(self):
        rc, kv, _ = plan('cac-fpm')
        self.assertEqual((rc, kv), (0, dict(EMPTY, cac_fpm=FULL)))

    def test_comma_list_spaces_duplicates_and_ordering(self):
        rc, kv, _ = plan(' cac:php85 , cac:php82,cac:php85, shared-httpd ,')
        self.assertEqual((rc, kv), (0, dict(EMPTY, cac='[82,85]', shared_httpd='true')))
        rc, kv, _ = plan('cac:php85,cac-fpm:php84')
        self.assertEqual((rc, kv), (0, dict(EMPTY, cac='[85]', cac_fpm='[84]')))

    def test_unknown_values_fail_loudly_and_emit_nothing(self):
        bad = ['foo', 'cac:php99', 'cac:85', 'cac:php8', 'CAC:php85', 'cac:', 'cac:php85:x', 'php85',
               'shared-httpd:php85', 'shared-ols:latest', ',', ' , ', 'cac:php85,nope',
               'cac:php85;rm -rf x', '$(id)', 'all', 'latest']
        for only in bad:
            rc, kv, err = plan(only)
            self.assertEqual(rc, 2, only)
            self.assertEqual(kv, {}, only)             # nothing half-planned
            self.assertIn('plan-legs:', err, only)


if __name__ == '__main__':
    unittest.main(verbosity=2)