# 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, '<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_v1": {"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_v1'
                dispatcher = list(m.msgmanager.dispatchers.values())[0]

                if sandboxed_worker_path is not None and worker_python_version == '2.7':
                    raise SkipTest('MessagePack protocol is not supported on python 2.7 worker')

                # 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)
