blob: 6fb2d6672a4f4f7127a935a2811a2cfa36f0d37c [file] [log] [blame]
#!/usr/bin/env python
# Copyright 2018 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import re
class FakeUser(object):
def __init__(self, login):
self.login = login
self.name = "Github User"
self.email = "github.user@example.com"
class FakeBranch(object):
def __init__(self, branch='master'):
self.name = branch
class FakeStatus(object):
def __init__(self, state, url, description, context, user):
self._state = state
self._url = url
self._description = description
self._context = context
self._user = user
def as_dict(self):
return {
'state': self._state,
'url': self._url,
'description': self._description,
'context': self._context,
'creator': {
'login': self._user
}
}
class FakeCommit(object):
def __init__(self):
self._statuses = []
def set_status(self, state, url, description, context, user):
status = FakeStatus(
state, url, description, context, user)
# always insert a status to the front of the list, to represent
# the last status provided for a commit.
self._statuses.insert(0, status)
def statuses(self):
return self._statuses
class FakeRepository(object):
def __init__(self):
self._branches = [FakeBranch()]
self._commits = {}
def branches(self, protected=False):
if protected:
# simulate there is no protected branch
return []
return self._branches
def create_status(self, sha, state, url, description, context,
user='zuul'):
# Since we're bypassing github API, which would require a user, we
# default the user as 'zuul' here.
commit = self._commits.get(sha, None)
if commit is None:
commit = FakeCommit()
self._commits[sha] = commit
commit.set_status(state, url, description, context, user)
def commit(self, sha):
commit = self._commits.get(sha, None)
if commit is None:
commit = FakeCommit()
self._commits[sha] = commit
return commit
class FakeLabel(object):
def __init__(self, name):
self.name = name
class FakeIssue(object):
def __init__(self, fake_pull_request):
self._fake_pull_request = fake_pull_request
def pull_request(self):
return FakePull(self._fake_pull_request)
def labels(self):
return [FakeLabel(l)
for l in self._fake_pull_request.labels]
class FakeFile(object):
def __init__(self, filename):
self.filename = filename
class FakePull(object):
def __init__(self, fake_pull_request):
self._fake_pull_request = fake_pull_request
def issue(self):
return FakeIssue(self._fake_pull_request)
def files(self):
return [FakeFile(fn)
for fn in self._fake_pull_request.files]
def as_dict(self):
pr = self._fake_pull_request
connection = pr.github
data = {
'number': pr.number,
'title': pr.subject,
'url': 'https://%s/%s/pull/%s' % (
connection.server, pr.project, pr.number
),
'updated_at': pr.updated_at,
'base': {
'repo': {
'full_name': pr.project
},
'ref': pr.branch,
},
'mergeable': True,
'state': pr.state,
'head': {
'sha': pr.head_sha,
'repo': {
'full_name': pr.project
}
},
'merged': pr.is_merged,
'body': pr.body
}
return data
class FakeIssueSearchResult(object):
def __init__(self, issue):
self.issue = issue
class FakeGithub(object):
def __init__(self, pull_requests):
self._pull_requests = pull_requests
self._repos = {}
def user(self, login):
return FakeUser(login)
def repository(self, owner, proj):
return self._repos.get((owner, proj), None)
def repo_from_project(self, project):
# This is a convenience method for the tests.
owner, proj = project.split('/')
return self.repository(owner, proj)
def addProject(self, project):
owner, proj = project.name.split('/')
self._repos[(owner, proj)] = FakeRepository()
def pull_request(self, owner, project, number):
fake_pr = self._pull_requests[number]
return FakePull(fake_pr)
def search_issues(self, query):
def tokenize(s):
return re.findall(r'[\w]+', s)
parts = tokenize(query)
terms = set()
results = []
for part in parts:
kv = part.split(':', 1)
if len(kv) == 2:
if kv[0] in set('type', 'is', 'in'):
# We only perform one search now and these aren't
# important; we can honor these terms later if
# necessary.
continue
terms.add(part)
for pr in self._pull_requests.values():
if not pr.body:
body = set()
else:
body = set(tokenize(pr.body))
if terms.intersection(body):
issue = FakeIssue(pr)
results.append(FakeIssueSearchResult(issue))
return results