build-push: workflow_dispatch input only to rebuild a single image leg

Gitea cannot re-run one job and a full run takes ~1h. Add a plan job that
turns the optional `only` input (e.g. cac:php85) into one filtered matrix per
image family; image jobs take their matrix from it and are skipped when empty.
Push events / empty input plan the full set as before. Unknown values fail
the plan job. Tests added to multiarch-workflow.test.py.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0185ukXjXyosSkunDsdWz3mb

scripts/tests/multiarch-workflow.test.py

@@ -59,7 +59,7 @@ class Wiring(unittest.TestCase):
             self.assertIn(j, self.jobs, j)
             if matrix:
                 self.assertIn('fail-fast: false', self.jobs[j], j)
-                self.assertIn('phpver: [82, 83, 84, 85]', 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:
@@ -169,5 +169,142 @@ class Wiring(unittest.TestCase):
             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)