test_container_steps.py (8193B)
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 """Run with python3 -m unittest discover -s buildbot -p 'test_*.py'.""" 20 21 import ast 22 from pathlib import Path 23 from types import SimpleNamespace 24 from unittest.mock import Mock, patch 25 26 from buildbot.plugins import steps, util 27 from buildbot.process import results 28 from twisted.internet import defer 29 from twisted.trial import unittest 30 31 32 # Load the helper without executing the master's workers, services and jobs. 33 config_path = Path(__file__).with_name("master.cfg") 34 config_ast = ast.parse(config_path.read_text(), filename=str(config_path)) 35 helper_ast = next(node for node in config_ast.body 36 if isinstance(node, ast.FunctionDef) 37 and node.name == "container_add_step") 38 namespace = { 39 "steps": steps, 40 "util": util, 41 "os": SimpleNamespace(path=SimpleNamespace(isdir=lambda _: False)), 42 "pwd": SimpleNamespace(getpwnam=lambda _: SimpleNamespace(pw_uid=1000)), 43 } 44 exec(compile(ast.Module(body=[helper_ast], type_ignores=[]), 45 str(config_path), "exec"), namespace) 46 container_add_step = namespace["container_add_step"] 47 48 49 class ContainerStepTests(unittest.TestCase): 50 def make_step(self, halt, warn, build=False): 51 with patch("builtins.print"): 52 template = container_add_step( 53 halt, warn, build, "localhost/test", None, 54 "/workdir", "test", "test-job") 55 step = template.get_step_factory().buildStep() 56 # Rendering the environment values is unrelated to result propagation. 57 for arg in step.commands: 58 arg.command = [str(value) for value in arg.command] 59 return step 60 61 def run_sequence(self, step, command_results): 62 step.makeRemoteShellCommand = Mock(side_effect=[ 63 SimpleNamespace(results=lambda result=result: result) 64 for result in command_results 65 ]) 66 step.runCommand = Mock(return_value=defer.succeed(None)) 67 result = self.successResultOf(step.runShellSequence(step.commands)) 68 build_result, halt = results.computeResultAndTermination( 69 step, result, results.SUCCESS) 70 return result, build_result, halt 71 72 def test_advisory_failure_warns_build_and_continues(self): 73 for build in (False, True): 74 with self.subTest(build=build): 75 step = self.make_step(False, True, build) 76 outcomes = ([results.SUCCESS] if build else []) + [results.FAILURE] 77 self.assertEqual(self.run_sequence(step, outcomes), 78 (results.WARNINGS, results.WARNINGS, False)) 79 80 def test_halting_failure_is_fatal_even_when_warnings_requested(self): 81 for warn in (False, True): 82 with self.subTest(warn=warn): 83 step = self.make_step(True, warn) 84 self.assertEqual(self.run_sequence(step, [results.FAILURE]), 85 (results.FAILURE, results.FAILURE, True)) 86 87 def test_non_halting_failure_without_warnings_still_fails_build(self): 88 step = self.make_step(False, False) 89 self.assertEqual(self.run_sequence(step, [results.FAILURE]), 90 (results.FAILURE, results.FAILURE, False)) 91 92 def test_container_failure_is_fatal_and_skips_run_command(self): 93 for halt in (False, True): 94 with self.subTest(halt=halt): 95 step = self.make_step(halt, True, build=True) 96 self.assertEqual(self.run_sequence(step, [results.FAILURE]), 97 (results.FAILURE, results.FAILURE, halt)) 98 self.assertEqual(step.runCommand.call_count, 1) 99 100 def test_success_stays_success(self): 101 for halt in (False, True): 102 for warn in (False, True): 103 with self.subTest(halt=halt, warn=warn): 104 step = self.make_step(halt, warn, build=True) 105 self.assertEqual(self.run_sequence( 106 step, [results.SUCCESS, results.SUCCESS]), 107 (results.SUCCESS, results.SUCCESS, False)) 108 109 110 class ContainerCheckoutTests(unittest.TestCase): 111 def checkout(self, repo): 112 # Evaluate the actual factory prefix, without registering schedulers or 113 # workers. Keep real Buildbot steps so their checkout options are checked. 114 loop = next(node for node in config_ast.body 115 if isinstance(node, ast.For) 116 and isinstance(node.iter, ast.Name) 117 and node.iter.id == "container_repos") 118 prefix = [] 119 for node in loop.body: 120 if (isinstance(node, ast.Expr) and isinstance(node.value, ast.Call) 121 and node.value.args and isinstance(node.value.args[0], ast.Call) 122 and isinstance(node.value.args[0].func, ast.Name) 123 and node.value.args[0].func.id == "GenerateStagesCommand"): 124 break 125 prefix.append(node) 126 scope = {"repo": "git.taler.net/" + repo, "util": util, 127 "ShellCommand": steps.ShellCommand, "Git": steps.Git} 128 exec(compile(ast.Module(body=prefix, type_ignores=[]), 129 str(config_path), "exec"), scope) 130 factory = scope["container_factory"] 131 return (scope["CONTAINER_WORKDIR"], 132 [step.buildStep() for step in factory.steps]) 133 134 def test_debian_publishers_cache_full_history_and_tags(self): 135 publishers = ("gnunet", "exchange", "merchant", "donau", "challenger", 136 "anastasis", "sync", "libeufin", "taler-rust", 137 "depolymerization", "taler-mailbox", "taldir", 138 "taler-typescript-core") 139 for repo in publishers: 140 with self.subTest(repo=repo): 141 _, (_, checkout) = self.checkout(repo) 142 self.assertEqual(checkout.mode, "full") 143 self.assertEqual(checkout.method, "fresh") 144 self.assertFalse(checkout.shallow) 145 self.assertTrue(checkout.tags) 146 self.assertTrue(checkout.submodules) 147 _, (_, android) = self.checkout("taler-android") 148 self.assertEqual(android.method, "clobber") 149 self.assertTrue(android.shallow) 150 151 def test_workspace_migrates_shallow_checkout_once(self): 152 import os 153 import subprocess 154 import tempfile 155 156 original, (workspace, _) = self.checkout("gnunet") 157 with tempfile.TemporaryDirectory() as temp: 158 root = Path(temp) 159 checkout = root / "checkout" 160 bindir = root / "bin" 161 bindir.mkdir() 162 # Skip only the container chmod. Exercise the actual removal and 163 # recreation commands against a disposable checkout directory. 164 podman = bindir / "podman" 165 podman.write_text("#!/bin/sh\nexit 0\n") 166 podman.chmod(0o755) 167 env = dict(os.environ, PATH=f"{bindir}:{os.environ['PATH']}") 168 command = workspace.command.replace(original, str(checkout)) 169 (checkout / ".git").mkdir(parents=True) 170 (checkout / ".git/shallow").touch() 171 marker = checkout / "old-checkout" 172 marker.touch() 173 subprocess.run(["sh", "-ec", command], check=True, env=env) 174 self.assertTrue(checkout.is_dir()) 175 self.assertFalse(marker.exists()) 176 (checkout / ".git").mkdir() 177 marker.touch() 178 subprocess.run(["sh", "-ec", command], check=True, env=env) 179 self.assertTrue(marker.exists())