quay/buildtrigger/test/bitbucketmock.py

from datetime import datetime

from mock import Mock

from buildtrigger.bitbuckethandler import BitbucketBuildTrigger
from util.morecollections import AttrDict


def get_bitbucket_trigger(dockerfile_path=""):
    trigger_obj = AttrDict(dict(auth_token="foobar", id="sometrigger"))
    trigger = BitbucketBuildTrigger(
        trigger_obj,
        {
            "build_source": "foo/bar",
            "dockerfile_path": dockerfile_path,
            "nickname": "knownuser",
            "account_id": "foo",
        },
    )

    trigger._get_client = get_mock_bitbucket
    return trigger


def get_repo_path_contents(path, revision):
    data = {
        "files": [{"path": "Dockerfile"}],
    }

    return (True, data, None)


def get_raw_path_contents(path, revision):
    if path == "Dockerfile":
        return (True, "hello world", None)

    if path == "somesubdir/Dockerfile":
        return (True, "hi universe", None)

    return (False, None, None)


def get_branches_and_tags():
    data = {
        "branches": [{"name": "master"}, {"name": "otherbranch"}],
        "tags": [{"name": "sometag"}, {"name": "someothertag"}],
    }
    return (True, data, None)


def get_branches():
    return (True, {"master": {}, "otherbranch": {}}, None)


def get_tags():
    return (True, {"sometag": {}, "someothertag": {}}, None)


def get_branch(branch_name):
    if branch_name != "master":
        return (False, None, None)

    data = {
        "target": {
            "hash": "aaaaaaa",
        },
    }

    return (True, data, None)


def get_tag(tag_name):
    if tag_name != "sometag":
        return (False, None, None)

    data = {
        "target": {
            "hash": "aaaaaaa",
        },
    }

    return (True, data, None)


def get_changeset_mock(commit_sha):
    if commit_sha != "aaaaaaa":
        return (False, None, "Not found")

    data = {
        "node": "aaaaaaa",
        "message": "some message",
        "timestamp": "now",
        "raw_author": "foo@bar.com",
    }

    return (True, data, None)


def get_changesets():
    changesets_mock = Mock()
    changesets_mock.get = Mock(side_effect=get_changeset_mock)
    return changesets_mock


def get_deploykeys():
    deploykeys_mock = Mock()
    deploykeys_mock.create = Mock(return_value=(True, {"pk": "someprivatekey"}, None))
    deploykeys_mock.delete = Mock(return_value=(True, {}, None))
    return deploykeys_mock


def get_webhooks():
    webhooks_mock = Mock()
    webhooks_mock.create = Mock(return_value=(True, {"uuid": "someuuid"}, None))
    webhooks_mock.delete = Mock(return_value=(True, {}, None))
    return webhooks_mock


def get_repo_mock(name):
    if name != "bar":
        return None

    repo_mock = Mock()
    repo_mock.get_main_branch = Mock(return_value=(True, {"name": "master"}, None))
    repo_mock.get_path_contents = Mock(side_effect=get_repo_path_contents)
    repo_mock.get_raw_path_contents = Mock(side_effect=get_raw_path_contents)
    repo_mock.get_branches_and_tags = Mock(side_effect=get_branches_and_tags)
    repo_mock.get_branches = Mock(side_effect=get_branches)
    repo_mock.get_tags = Mock(side_effect=get_tags)
    repo_mock.get_branch = Mock(side_effect=get_branch)
    repo_mock.get_tag = Mock(side_effect=get_tag)

    repo_mock.changesets = Mock(side_effect=get_changesets)
    repo_mock.deploykeys = Mock(side_effect=get_deploykeys)
    repo_mock.webhooks = Mock(side_effect=get_webhooks)
    return repo_mock


def get_repositories_mock():
    repos_mock = Mock()
    repos_mock.get = Mock(side_effect=get_repo_mock)
    return repos_mock


def get_namespace_mock(namespace):
    namespace_mock = Mock()
    namespace_mock.repositories = Mock(side_effect=get_repositories_mock)
    return namespace_mock


def get_repo(namespace, name):
    return {
        "owner": namespace,
        "logo": "avatarurl",
        "slug": name,
        "description": "some %s repo" % (name),
        "utc_last_updated": str(datetime.utcfromtimestamp(0)),
        "read_only": namespace != "knownuser",
        "is_private": name == "somerepo",
    }


def get_visible_repos():
    repos = [
        get_repo("knownuser", "somerepo"),
        get_repo("someorg", "somerepo"),
        get_repo("someorg", "anotherrepo"),
    ]
    return (True, repos, None)


def get_authed_mock(token, secret):
    authed_mock = Mock()
    authed_mock.for_namespace = Mock(side_effect=get_namespace_mock)
    authed_mock.get_visible_repositories = Mock(side_effect=get_visible_repos)
    return authed_mock


def get_mock_bitbucket():
    bitbucket_mock = Mock()
    bitbucket_mock.get_authorized_client = Mock(side_effect=get_authed_mock)
    return bitbucket_mock