239 lines
6.6 KiB

# Copyright 2019 BMW Group
# 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.
from graphene import Boolean, Field, Int, List, ObjectType, String
class FakePageInfo(ObjectType):
end_cursor = String()
has_next_page = Boolean()
def resolve_end_cursor(parent, info):
return 'testcursor'
def resolve_has_next_page(parent, info):
return False
class FakeMatchingRef(ObjectType):
name = String()
def resolve_name(parent, info):
return parent
class FakeMatchingRefs(ObjectType):
nodes = List(FakeMatchingRef)
def resolve_nodes(parent, info):
# To simplify tests just return the pattern and a bogus ref that should
# not disturb zuul.
return [parent.pattern, 'bogus-ref']
class FakeBranchProtectionRule(ObjectType):
pattern = String()
requiredStatusCheckContexts = List(String)
requiresApprovingReviews = Boolean()
requiresCodeOwnerReviews = Boolean()
matchingRefs = Field(FakeMatchingRefs, first=Int())
def resolve_pattern(parent, info):
return parent.pattern
def resolve_requiredStatusCheckContexts(parent, info):
return parent.required_contexts
def resolve_requiresApprovingReviews(parent, info):
return parent.require_reviews
def resolve_requiresCodeOwnerReviews(parent, info):
return parent.require_codeowners_review
def resolve_matchingRefs(parent, info, first=None):
return parent
class FakeBranchProtectionRules(ObjectType):
nodes = List(FakeBranchProtectionRule)
def resolve_nodes(parent, info):
return parent.values()
class FakeActor(ObjectType):
login = String()
class FakeStatusContext(ObjectType):
state = String()
context = String()
creator = Field(FakeActor)
def resolve_state(parent, info):
state = parent.state.upper()
return state
def resolve_context(parent, info):
return parent.context
def resolve_creator(parent, info):
return parent.creator
class FakeStatus(ObjectType):
contexts = List(FakeStatusContext)
def resolve_contexts(parent, info):
return parent
class FakeCheckRun(ObjectType):
name = String()
conclusion = String()
def resolve_name(parent, info):
return parent.name
def resolve_conclusion(parent, info):
if parent.conclusion:
return parent.conclusion.upper()
return None
class FakeCheckRuns(ObjectType):
nodes = List(FakeCheckRun)
def resolve_nodes(parent, info):
return parent
class FakeApp(ObjectType):
slug = String()
name = String()
class FakeCheckSuite(ObjectType):
app = Field(FakeApp)
checkRuns = Field(FakeCheckRuns, first=Int())
def resolve_app(parent, info):
if not parent:
return None
return parent[0].app
def resolve_checkRuns(parent, info, first=None):
# We only want to return the latest result for a check run per app.
# Since the check runs are ordered from latest to oldest result we
# need to traverse the list in reverse order.
check_runs_by_name = {
"{}:{}".format(cr.app, cr.name): cr for cr in reversed(parent)
return check_runs_by_name.values()
class FakeCheckSuites(ObjectType):
nodes = List(FakeCheckSuite)
def resolve_nodes(parent, info):
# Note: we only use a single check suite in the tests so return a
# single item to keep it simple.
return [parent]
class FakeCommit(ObjectType):
class Meta:
# Graphql object type that defaults to the class name, but we require
# 'Commit'.
name = 'Commit'
status = Field(FakeStatus)
checkSuites = Field(FakeCheckSuites, first=Int())
def resolve_status(parent, info):
seen = set()
result = []
for status in parent._statuses:
if status.context not in seen:
# Github returns None if there are no results
return result or None
def resolve_checkSuites(parent, info, first=None):
# Tests only utilize one check suite so return all runs for that.
return parent._check_runs
class FakePullRequest(ObjectType):
isDraft = Boolean()
reviewDecision = String()
mergeable = String()
def resolve_isDraft(parent, info):
return parent.draft
def resolve_mergeable(parent, info):
return "MERGEABLE" if parent.mergeable else "CONFLICTING"
def resolve_reviewDecision(parent, info):
if hasattr(info.context, 'version') and info.context.version:
if info.context.version < (2, 21, 0):
raise Exception('Field unsupported')
# Check branch protection rules if reviews are required
org, project = parent.project.split('/')
repo = info.context._data.repos[(org, project)]
rule = repo._branch_protection_rules.get(parent.branch)
if not rule or not rule.require_reviews:
# Github returns None if there is no review required
return None
approvals = [r for r in parent.reviews
if r.data['state'] == 'APPROVED']
if approvals:
return 'APPROVED'
class FakeRepository(ObjectType):
name = String()
branchProtectionRules = Field(FakeBranchProtectionRules, first=Int())
pullRequest = Field(FakePullRequest, number=Int(required=True))
object = Field(FakeCommit, expression=String(required=True))
def resolve_name(parent, info):
org, name = parent.name.split('/')
return name
def resolve_branchProtectionRules(parent, info, first):
return parent._branch_protection_rules
def resolve_pullRequest(parent, info, number):
return parent.data.pull_requests.get(number)
def resolve_object(parent, info, expression):
return parent._commits.get(expression)
class FakeGithubQuery(ObjectType):
repository = Field(FakeRepository, owner=String(required=True),
def resolve_repository(root, info, owner, name):
return info.context._data.repos.get((owner, name))