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(subdir=''):
  trigger_obj = AttrDict(dict(auth_token='foobar', id='sometrigger'))
  trigger = BitbucketBuildTrigger(trigger_obj, {
    'build_source': 'foo/bar',
    'subdir': subdir,
    'username': 'knownuser'
  })

  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