Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
gh = FakeGH(getiter={"/orgs/python/teams": teams}, getitem=getitem)
assert await util.is_core_dev(gh, "mariatta")
assert gh.getiter_url == "/orgs/python/teams"
teams = [{"name": "python core", "id": 42}]
getitem = {
"/teams/42/memberships/miss-islington": gidgethub.BadRequest(
status_code=http.HTTPStatus(404)
)
}
gh = FakeGH(getiter={"/orgs/python/teams": teams}, getitem=getitem)
assert not await util.is_core_dev(gh, "miss-islington")
teams = [{"name": "python core", "id": 42}]
getitem = {
"/teams/42/memberships/miss-islington": gidgethub.BadRequest(
status_code=http.HTTPStatus(400)
)
}
gh = FakeGH(getiter={"/orgs/python/teams": teams}, getitem=getitem)
with pytest.raises(gidgethub.BadRequest):
await util.is_core_dev(gh, "miss-islington")
"state": "success",
"statuses": [
{
"state": "success",
"description": "Issue report skipped",
"context": "bedevere/issue-number",
},
{
"state": "success",
"description": "The Travis CI build passed",
"target_url": "https://travis-ci.org/python/cpython/builds/340259685?utm_source=github_status&utm_medium=notification",
"context": "continuous-integration/travis-ci/pr",
},
],
},
"/teams/42/memberships/miss-islington": gidgethub.BadRequest(
status_code=http.HTTPStatus(404)
),
}
getiter = {
"/repos/python/cpython/pulls/5547/commits": [{"sha": sha}],
"/orgs/python/teams": [{"name": "python core", "id": 42}],
}
gh = FakeGH(getitem=getitem, getiter=getiter)
await status_change.router.dispatch(event, gh)
assert (
gh.delete_url == f'{data["pull_request"]["issue_url"]}/labels/{AUTOMERGE_LABEL}'
)
assert not hasattr(gh, "post_data") # does not leave a comment
assert gh.getiter_url == "https://api.github.com/orgs/python/teams"
teams = [{"name": "Python core", "id": 42}]
getitem = {
"https://api.github.com/teams/42/memberships/andrea": gidgethub.BadRequest(
status_code=http.HTTPStatus(404)
)
}
gh = FakeGH(
getiter={"https://api.github.com/orgs/python/teams": teams}, getitem=getitem
)
assert not await util.is_core_dev(gh, "andrea")
teams = [{"name": "Python core", "id": 42}]
getitem = {
"https://api.github.com/teams/42/memberships/andrea": gidgethub.BadRequest(
status_code=http.HTTPStatus(400)
)
}
gh = FakeGH(
getiter={"https://api.github.com/orgs/python/teams": teams}, getitem=getitem
)
with pytest.raises(gidgethub.BadRequest):
await util.is_core_dev(gh, "andrea")
"labels": [],
"labels_url": "https://api.github.com/labels/42",
"url": "https://api.github.com/issue/42",
"pull_request": {"url": "https://api.github.com/pr/42"},
"comments_url": "https://api.github.com/comments/42",
},
"comment": {
"user": {"login": "andreamcinnes"},
"body": awaiting.BORING_TRIGGER_PHRASE,
},
}
event = sansio.Event(data, event="issue_comment", delivery_id="12345")
items = {
"https://api.github.com/teams/6/memberships/brettcannon": True,
"https://api.github.com/teams/6/memberships/gvanrossum": True,
"https://api.github.com/teams/6/memberships/not-core-dev": gidgethub.BadRequest(
status_code=http.HTTPStatus(404)
),
}
iterators = {
"https://api.github.com/orgs/python/teams": [{"name": "python core", "id": 6}],
"https://api.github.com/pr/42/reviews": [
{"user": {"login": "brettcannon"}, "state": "approved"},
{"user": {"login": "gvanrossum"}, "state": "changes_requested"},
{"user": {"login": "not-core-dev"}, "state": "approved"},
],
}
gh = FakeGH(getitem=items, getiter=iterators)
await awaiting.router.dispatch(event, gh)
assert len(gh.post_) == 3
labeling, comment, review_request = gh.post_
assert labeling[0] == "https://api.github.com/labels/42"
async def wrapper(request, *, github_app, webhook_secret=None):
if request.method not in _allowed_methods:
raise web.HTTPMethodNotAllowed(
method=request.method,
allowed_methods=_allowed_methods,
) from BadRequest(HTTPStatus.METHOD_NOT_ALLOWED)
return await wrapped_function(
request,
github_app=github_app,
webhook_secret=webhook_secret,
)
will be performed unconditionally. Any failure in validation
(including not providing a secret) will lead to ValidationFailure being
raised.
"""
if "x-hub-signature" in headers:
if secret is None:
raise ValidationFailure("secret not provided")
validate_event(body, signature=headers["x-hub-signature"],
secret=secret)
elif secret is not None:
raise ValidationFailure("signature is missing")
try:
data = _decode_body(headers["content-type"], body, strict=True)
except (KeyError, ValueError) as exc:
raise BadRequest(http.HTTPStatus(415),
"expected a content-type of "
"'application/json' or "
"'application/x-www-form-urlencoded'") from exc
return cls(data, event=headers["x-github-event"],
delivery_id=headers["x-github-delivery"])
async def _get_github_client(self, http_session: ClientSession):
"""Return a GitHub API client for the target org."""
github_app = self._get_github_app(http_session)
try:
github_app_installations = await github_app.get_installations()
except gidgethub.BadRequest:
error_msg = 'Invalid GitHub App credentials'
logger.error(error_msg)
raise LookupError(error_msg)
target_github_app_installation = next( # find the one
(
i for n, i in github_app_installations.items()
if i._metadata.account['login'] == self.github_org_name
),
None,
)
return target_github_app_installation.api_client
async def is_core_dev(gh, username):
"""Check if the user is a CPython core developer."""
org_teams = "/orgs/python/teams"
team_name = "python core"
async for team in gh.getiter(org_teams):
if team["name"].lower() == team_name:
break
else:
raise ValueError(f"{team_name!r} not found at {org_teams!r}")
# The 'teams' object only provides a URL to a deprecated endpoint,
# so manually construct the URL to the non-deprecated team membership
# endpoint.
membership_url = f"/teams/{team['id']}/memberships/{username}"
try:
await gh.getitem(membership_url)
except gidgethub.BadRequest as exc:
if exc.status_code == 404:
return False
raise
else:
return True
logger.info(
'Expected failure: should retry' if is_404
else 'Unexpected failure: should fail loudly'
)
logger.error(
'Error: %r; status: %r; args: %s.',
gh_err_resp,
gh_err_resp.status_code,
gh_err_resp.args,
)
return not is_404
retry_on_not_found = backoff.on_exception( # pylint: disable=invalid-name
backoff.expo, gidgethub.BadRequest,
max_tries=3, max_time=15, jitter=backoff.full_jitter,
giveup=_is_not_404_response,
)
def provision_http_session(async_method):
"""Inject aiohttp client session into method keyword args."""
async def async_method_wrapper(self, *args, **kwargs):
async with ClientSession() as http_session:
kwargs['http_session'] = http_session
return await async_method(self, *args, **kwargs)
return async_method_wrapper
@dataclass(frozen=True)
class GitHubOrgClient: