test_mhd2_publisher.py (5135B)
1 # This file is part of TALER 2 # (C) 2026 Taler Systems SA 3 # 4 # TALER is free software; you can redistribute it and/or 5 # modify it under the terms of the GNU Affero General Public 6 # License as published by the Free Software Foundation; either 7 # version 3, or (at your option) any later version. 8 # 9 # TALER is distributed in the hope that it will be useful, 10 # but WITHOUT ANY WARRANTY; without even the implied warranty 11 # of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 12 # See the GNU General Public License for more details. 13 # 14 # You should have received a copy of the GNU General Public 15 # License along with TALER; see the file COPYING. If not, 16 # see <http://www.gnu.org/licenses/> 17 # 18 19 """MHD2 publication must wait for both architecture builds.""" 20 21 import ast 22 import contextlib 23 import io 24 import os 25 from pathlib import Path 26 import subprocess 27 import tempfile 28 from types import SimpleNamespace 29 import unittest 30 31 from buildbot.plugins import schedulers, util 32 from buildbot.steps.source.git import Git 33 from test_container_steps import container_add_step 34 35 ROOT = Path(__file__).resolve().parent.parent 36 37 38 def factory_config(): 39 tree = ast.parse((ROOT / "buildbot/master.cfg").read_text()) 40 nodes = [] 41 collecting = False 42 for node in tree.body: 43 if isinstance(node, ast.Assign): 44 names = [target.id for target in node.targets if isinstance(target, ast.Name)] 45 if "MHD2_WORKDIR" in names: 46 collecting = True 47 if "container_repos" in names: 48 break 49 if collecting: 50 nodes.append(node) 51 scope = dict(util=util, Git=Git, schedulers=schedulers, 52 container_add_step=container_add_step, BUILDERS=[], SCHEDULERS=[]) 53 with contextlib.redirect_stdout(io.StringIO()): 54 exec(compile(ast.Module(body=nodes, type_ignores=[]), "mhd2-config", "exec"), scope) 55 return scope 56 57 58 class PublisherTests(unittest.TestCase): 59 def test_both_builds_precede_upload_and_halt_on_failure(self): 60 config = factory_config() 61 steps = [step.buildStep() for step in config["mhd2_factory"].steps] 62 self.assertEqual([step.name for step in steps], 63 ["git", "build-amd64", "build-arm64", "upload"]) 64 for arch, step in zip(("amd64", "arm64"), steps[1:3]): 65 self.assertTrue(step.haltOnFailure) 66 build, run = step.commands 67 self.assertTrue(build.haltOnFailure) 68 self.assertTrue(run.haltOnFailure) 69 self.assertEqual(build.command[build.command.index("--arch") + 1], arch) 70 self.assertEqual(run.command[-1], "buildbot/libmicrohttpd2/build.sh") 71 self.assertTrue(steps[-1].haltOnFailure) 72 self.assertEqual(steps[-1].commands[0].command[-1], 73 "buildbot/libmicrohttpd2/upload.sh") 74 75 def test_changes_to_pin_overlay_and_jobs_trigger_build(self): 76 changed = factory_config()["mhd2_packaging_changed"] 77 for path in ("buildbot/master.cfg", "buildbot/libmicrohttpd2/build.sh", 78 "packaging/ng/buildscripts/debian-overlays/libmicrohttpd2/source.conf", 79 "packaging/ng/buildscripts/debian-overlays/libmicrohttpd2/debian/control"): 80 self.assertTrue(changed(SimpleNamespace(files=[path]))) 81 self.assertFalse(changed(SimpleNamespace(files=["packaging/ng/packages.toml"]))) 82 83 def test_upload_requires_both_manifests_and_ignores_stale_packages(self): 84 with tempfile.TemporaryDirectory() as tmp: 85 root = Path(tmp) 86 log = root / "uploaded" 87 (root / "rsync").write_text('#!/bin/sh\nprintf "%s\\n" "$@" > "$UPLOAD_LOG"\n') 88 (root / "rsync").chmod(0o755) 89 env = dict(os.environ, PATH=tmp + ":" + os.environ["PATH"], 90 UPLOAD_LOG=str(log), CI_COMMIT_REF="test", 91 MHD2_ARTIFACT_ROOT=tmp) 92 def upload(): 93 return subprocess.run([str(ROOT / "buildbot/libmicrohttpd2/upload.sh")], 94 env=env, capture_output=True) 95 self.assertNotEqual(upload().returncode, 0) 96 self.assertFalse(log.exists()) 97 for arch in ("amd64", "arm64"): 98 output = root / "libmicrohttpd2/test" / arch 99 output.mkdir(parents=True) 100 (output / "packages.list").write_text(f"current_{arch}.deb\n") 101 (output / f"current_{arch}.deb").touch() 102 (output / "stale.deb").touch() 103 if arch == "amd64": 104 self.assertNotEqual(upload().returncode, 0) 105 self.assertFalse(log.exists()) 106 self.assertEqual(upload().returncode, 0) 107 args = log.read_text() 108 self.assertIn("current_amd64.deb", args) 109 self.assertIn("current_arm64.deb", args) 110 self.assertNotIn("stale.deb", args) 111 log.unlink() 112 (output / "packages.list").write_text("../escape.deb\n") 113 self.assertNotEqual(upload().returncode, 0) 114 self.assertFalse(log.exists()) 115 116 117 if __name__ == "__main__": 118 unittest.main()