# This file is part of Buildbot. Buildbot is free software: you can # redistribute it and/or modify it under the terms of the GNU General Public # License as published by the Free Software Foundation, version 2. # # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. See the GNU General Public License for more # details. # # You should have received a copy of the GNU General Public License along with # this program; if not, write to the Free Software Foundation, Inc., 51 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. # # Copyright Buildbot Team Members import os import re import sys from io import StringIO from unittest.case import SkipTest import mock from twisted.internet import defer from twisted.internet import reactor from twisted.python.filepath import FilePath from twisted.trial import unittest from zope.interface import implementer from buildbot.config.master import MasterConfig from buildbot.data import resultspec from buildbot.interfaces import IConfigLoader from buildbot.master import BuildMaster from buildbot.plugins import worker from buildbot.process.properties import Interpolate from buildbot.process.results import SUCCESS from buildbot.process.results import statusToString from buildbot.test.reactor import TestReactorMixin from buildbot.test.util.misc import DebugIntegrationLogsMixin from buildbot.test.util.sandboxed_worker import SandboxedWorker from buildbot.worker.local import LocalWorker try: from buildbot_worker.bot import Worker except ImportError: Worker = None @implementer(IConfigLoader) class DictLoader: def __init__(self, config_dict): self.config_dict = config_dict def loadConfig(self): return MasterConfig.loadFromDict(self.config_dict, '') @defer.inlineCallbacks def getMaster(case, reactor, config_dict): """ Create a started ``BuildMaster`` with the given configuration. """ basedir = FilePath(case.mktemp()) basedir.createDirectory() config_dict['buildbotNetUsageData'] = None master = BuildMaster( basedir.path, reactor=reactor, config_loader=DictLoader(config_dict)) if 'db_url' not in config_dict: config_dict['db_url'] = 'sqlite://' # TODO: Allow BuildMaster to transparently upgrade the database, at least # for tests. master.config.db['db_url'] = config_dict['db_url'] yield master.db.setup(check_version=False) yield master.db.model.upgrade() master.db.setup = lambda: None yield master.startService() case.addCleanup(master.db.pool.shutdown) case.addCleanup(master.stopService) return master class RunFakeMasterTestCase(unittest.TestCase, TestReactorMixin, DebugIntegrationLogsMixin): def setUp(self): self.setup_test_reactor() self.setupDebugIntegrationLogs() def tearDown(self): self.assertFalse(self.master.running, "master is still running!") @defer.inlineCallbacks def setup_master(self, config_dict): self.master = yield getMaster(self, self.reactor, config_dict) @defer.inlineCallbacks def reconfig_master(self, config_dict=None): if config_dict is not None: self.master.config_loader.config_dict = config_dict yield self.master.doReconfig() @defer.inlineCallbacks def clean_master_shutdown(self, quick=False): yield self.master.botmaster.cleanShutdown(quickMode=quick, stopReactor=False) def createLocalWorker(self, name, **kwargs): workdir = FilePath(self.mktemp()) workdir.createDirectory() return LocalWorker(name, workdir.path, **kwargs) @defer.inlineCallbacks def assertBuildResults(self, build_id, result): dbdict = yield self.master.db.builds.getBuild(build_id) self.assertEqual(result, dbdict['results']) @defer.inlineCallbacks def assertStepStateString(self, step_id, state_string): datadict = yield self.master.data.get(('steps', step_id)) self.assertEqual(datadict['state_string'], state_string) @defer.inlineCallbacks def assertLogs(self, build_id, exp_logs): got_logs = {} data_logs = yield self.master.data.get(('builds', build_id, 'steps', 1, 'logs')) for log in data_logs: self.assertTrue(log['complete']) log_contents = yield self.master.data.get(('builds', build_id, 'steps', 1, 'logs', log['slug'], 'contents')) got_logs[log['name']] = log_contents['content'] self.assertEqual(got_logs, exp_logs) @defer.inlineCallbacks def create_build_request(self, builder_ids, properties=None): properties = properties.asDict() if properties is not None else None ret = yield self.master.data.updates.addBuildset( waited_for=False, builderids=builder_ids, sourcestamps=[ {'codebase': '', 'repository': '', 'branch': None, 'revision': None, 'project': ''}, ], properties=properties, ) return ret @defer.inlineCallbacks def do_test_build_by_name(self, builder_name): builder_id = yield self.master.data.updates.findBuilderId(builder_name) yield self.do_test_build(builder_id) @defer.inlineCallbacks def do_test_build(self, builder_id): # setup waiting for build to finish d_finished = defer.Deferred() def on_finished(_, __): if not d_finished.called: d_finished.callback(None) consumer = yield self.master.mq.startConsuming(on_finished, ('builds', None, 'finished')) # start the builder yield self.create_build_request([builder_id]) # and wait for build completion yield d_finished yield consumer.stopConsuming() class RunMasterBase(unittest.TestCase): proto = "null" if Worker is None: skip = "buildbot-worker package is not installed" @defer.inlineCallbacks def setupConfig(self, config_dict, startWorker=True, **worker_kwargs): """ Setup and start a master configured by the function configFunc defined in the test module. @type config_dict: dict @param configFunc: The BuildmasterConfig dictionary. """ # mock reactor.stop (which trial *really* doesn't # like test code to call!) stop = mock.create_autospec(reactor.stop) self.patch(reactor, 'stop', stop) if startWorker: if self.proto == 'pb': proto = {"pb": {"port": "tcp:0:interface=127.0.0.1"}} workerclass = worker.Worker if self.proto == 'msgpack': proto = {"msgpack_experimental_v7": {"port": 0}} workerclass = worker.Worker elif self.proto == 'null': proto = {"null": {}} workerclass = worker.LocalWorker config_dict['workers'] = [workerclass("local1", password=Interpolate("localpw"), missing_timeout=0)] config_dict['protocols'] = proto m = yield getMaster(self, reactor, config_dict) self.master = m self.assertFalse(stop.called, "startService tried to stop the reactor; check logs") if not startWorker: return if self.proto in ('pb', 'msgpack'): sandboxed_worker_path = os.environ.get("SANDBOXED_WORKER_PATH", None) worker_python_version = os.environ.get("WORKER_PYTHON", None) if self.proto == 'pb': protocol = 'pb' dispatcher = list(m.pbmanager.dispatchers.values())[0] else: protocol = 'msgpack_experimental_v7' dispatcher = list(m.msgmanager.dispatchers.values())[0] unsupported_python_versions = ['2.7', '3.4', '3.5'] if sandboxed_worker_path is not None and \ worker_python_version in unsupported_python_versions: raise SkipTest('MessagePack protocol requires worker python >= 3.6') # We currently don't handle connection closing cleanly. dispatcher.serverFactory.setProtocolOptions(closeHandshakeTimeout=0) workerPort = dispatcher.port.getHost().port # create a worker, and attach it to the master, it will be started, and stopped # along with the master worker_dir = FilePath(self.mktemp()) worker_dir.createDirectory() if sandboxed_worker_path is None: self.w = Worker( "127.0.0.1", workerPort, "local1", "localpw", worker_dir.path, False, protocol=protocol, **worker_kwargs) else: self.w = SandboxedWorker( "127.0.0.1", workerPort, "local1", "localpw", worker_dir.path, sandboxed_worker_path, protocol=protocol, **worker_kwargs) self.addCleanup(self.w.shutdownWorker) elif self.proto == 'null': self.w = None if self.w is not None: yield self.w.setServiceParent(m) @defer.inlineCallbacks def dump(): if not self._passed: dump = StringIO() print("FAILED! dumping build db for debug", file=dump) builds = yield self.master.data.get(("builds",)) for build in builds: yield self.printBuild(build, dump, withLogs=True) raise self.failureException(dump.getvalue()) self.addCleanup(dump) @defer.inlineCallbacks def doForceBuild(self, wantSteps=False, wantProperties=False, wantLogs=False, useChange=False, forceParams=None, triggerCallback=None): if forceParams is None: forceParams = {} # force a build, and wait until it is finished d = defer.Deferred() # in order to allow trigger based integration tests # we wait until the first started build is finished self.firstbsid = None def newCallback(_, data): if self.firstbsid is None: self.firstbsid = data['bsid'] newConsumer.stopConsuming() def finishedCallback(_, data): if self.firstbsid == data['bsid']: d.callback(data) newConsumer = yield self.master.mq.startConsuming( newCallback, ('buildsets', None, 'new')) finishedConsumer = yield self.master.mq.startConsuming( finishedCallback, ('buildsets', None, 'complete')) if triggerCallback is not None: yield triggerCallback() elif useChange is False: # use data api to force a build yield self.master.data.control("force", forceParams, ("forceschedulers", "force")) else: # use data api to force a build, via a new change yield self.master.data.updates.addChange(**useChange) # wait until we receive the build finished event buildset = yield d buildrequests = yield self.master.data.get( ('buildrequests',), filters=[resultspec.Filter('buildsetid', 'eq', [buildset['bsid']])]) buildrequest = buildrequests[-1] builds = yield self.master.data.get( ('builds',), filters=[resultspec.Filter('buildrequestid', 'eq', [buildrequest['buildrequestid']])]) # if the build has been retried, there will be several matching builds. # We return the last build build = builds[-1] finishedConsumer.stopConsuming() yield self.enrichBuild(build, wantSteps, wantProperties, wantLogs) return build @defer.inlineCallbacks def enrichBuild(self, build, wantSteps=False, wantProperties=False, wantLogs=False): # enrich the build result, with the step results if wantSteps: build["steps"] = yield self.master.data.get(("builds", build['buildid'], "steps")) # enrich the step result, with the logs results if wantLogs: build["steps"] = list(build["steps"]) for step in build["steps"]: step['logs'] = yield self.master.data.get(("steps", step['stepid'], "logs")) step["logs"] = list(step['logs']) for log in step["logs"]: log['contents'] = yield self.master.data.get(("logs", log['logid'], "contents")) if wantProperties: build["properties"] = yield self.master.data.get(("builds", build['buildid'], "properties")) @defer.inlineCallbacks def printBuild(self, build, out=sys.stdout, withLogs=False): # helper for debugging: print a build yield self.enrichBuild(build, wantSteps=True, wantProperties=True, wantLogs=True) print(f"*** BUILD {build['buildid']} *** ==> {build['state_string']} " f"({statusToString(build['results'])})", file=out) for step in build['steps']: print(f" *** STEP {step['name']} *** ==> {step['state_string']} " f"({statusToString(step['results'])})", file=out) for url in step['urls']: print(f" url:{url['name']} ({url['url']})", file=out) for log in step['logs']: print(f" log:{log['name']} ({log['num_lines']})", file=out) if step['results'] != SUCCESS or withLogs: self.printLog(log, out) def _match_patterns_consume(self, text, patterns, is_regex): for pattern in patterns[:]: if is_regex: if re.search(pattern, text): patterns.remove(pattern) else: if pattern in text: patterns.remove(pattern) return patterns @defer.inlineCallbacks def checkBuildStepLogExist(self, build, expectedLog, onlyStdout=False, regex=False): if isinstance(expectedLog, str): expectedLog = [expectedLog] if not isinstance(expectedLog, list): raise Exception('The expectedLog argument must be either string or a list of strings') yield self.enrichBuild(build, wantSteps=True, wantProperties=True, wantLogs=True) for step in build['steps']: for log in step['logs']: for line in log['contents']['content'].splitlines(): if onlyStdout and line[0] != 'o': continue expectedLog = self._match_patterns_consume(line, expectedLog, is_regex=regex) if expectedLog: print(f"{expectedLog} not found in logs") return len(expectedLog) == 0 def printLog(self, log, out): print(" " * 8 + f"*********** LOG: {log['name']} *********", file=out) if log['type'] == 's': for line in log['contents']['content'].splitlines(): linetype = line[0] line = line[1:] if linetype == 'h': # cyan line = "\x1b[36m" + line + "\x1b[0m" if linetype == 'e': # red line = "\x1b[31m" + line + "\x1b[0m" print(" " * 8 + line) else: print("" + log['contents']['content'], file=out) print(" " * 8 + "********************************", file=out)