actions/octokit.py
1
0
Fork 0
mirror of https://github.com/khornberg/octokit.py synced 2026-05-10 00:00:25 +03:00
Code Issues Packages Projects Releases Wiki Activity

Refactor and restyle

Browse Source
This commit is contained in:
Kyle Hornberg
2020-04-13 16:29:41 -05:00
parent ddacfed1ff
commit cdd7c69730
11 changed files with 105 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+16
View File
@@ -0,0 +1,16 @@
[tool.black]
line-length = 120
target-version = ['py36']
[tool.isort]
force_single_line = true
line_length = 120
known_first_party = 'octokit.py'
default_section = 'THIRDPARTY'
forced_separate = 'test_octokit.py'
not_skip = '__init__.py'
skip = 'migrations'
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
use_parentheses = true
setup.cfg
-6
View File
@@ -29,9 +29,3 @@ default_section = THIRDPARTY
forced_separate = test_octokit.py
not_skip = __init__.py
skip = migrations
[yapf]
based_on_style = facebook
COLUMN_LIMIT = 120
BLANK_LINE_BEFORE_NESTED_CLASS_OR_DEF = true
ALLOW_SPLIT_BEFORE_DICT_VALUE = false
setup.py
+2 -6
View File
@@ -16,9 +16,7 @@ from setuptools import setup
def read(*names, **kwargs):
return io.open(
join(dirname(__file__), *names), encoding=kwargs.get("encoding", "utf8")
).read()
return io.open(join(dirname(__file__), *names), encoding=kwargs.get("encoding", "utf8")).read()
setup(
@@ -28,9 +26,7 @@ setup(
description="Python client for GitHub API",
long_description="%s\n%s"
% (
re.compile("^.. start-badges.*^.. end-badges", re.M | re.S).sub(
"", read("README.rst")
),
re.compile("^.. start-badges.*^.. end-badges", re.M | re.S).sub("", read("README.rst")),
re.sub(":[a-z]+:`~?(.*?)`", r"``\1``", read("CHANGELOG.rst")),
),
author="Kyle Hornberg",
src/octokit/__init__.py
+13 -39
View File
@@ -27,13 +27,7 @@ class Octokit(Base):
class_attributes = defaultdict(dict)
for path, path_object in definitions["paths"].items():
for method, method_object in path_object.items():
cls_name, method_id_name = self._get_names_from_operation_id(
method_object
)
method_name = utils.snake_case(str(method_object.get("summary")))
methods = self._get_class_methods(
method_id_name, method_name, method_object, method, path
)
cls_name, methods = self._get_class_methods(method_object, method, path)
class_attributes[cls_name].update(methods)
return class_attributes
@@ -72,39 +66,27 @@ class Octokit(Base):
if method_object.get("x-changes"):
for change in method_object["x-changes"]:
if change.get("type") == "operation":
before_cls, before_name = self._get_names_from_operation_id(
change.get("before")
)
after_cls, after_name = self._get_names_from_operation_id(
change.get("after")
)
before_cls, before_name = self._get_names_from_operation_id(change.get("before"))
after_cls, after_name = self._get_names_from_operation_id(change.get("after"))
if before_cls == after_cls and methods.get(after_name):
deprecated_methods.update({before_name: methods[after_name]})
return deprecated_methods
def _get_class_methods(
self, method_id_name, method_name, method_object, method, path
):
def _get_class_methods(self, method_object, method, path):
cls_name, method_id_name = self._get_names_from_operation_id(method_object)
method_name = utils.snake_case(str(method_object.get("summary")))
methods = {
method_id_name: self._create_method(
method_id_name, method_object, method, path
),
method_id_name: self._create_method(method_id_name, method_object, method, path),
method_name: self._create_method(method_name, method_object, method, path),
}
methods.update(self._get_deprecated_methods(methods, method_object))
return methods
return cls_name, methods
def _convert_to_object(self, item):
if isinstance(item, dict):
return type(
"ResponseData",
(object,),
{k: self._convert_to_object(v) for k, v in item.items()},
)
return type("ResponseData", (object,), {k: self._convert_to_object(v) for k, v in item.items()})
if isinstance(item, list):
return list(
(self._convert_to_object(value) for index, value in enumerate(item))
)
return list((self._convert_to_object(value) for index, value in enumerate(item)))
else:
return item
@@ -116,15 +98,9 @@ class Octokit(Base):
if matches:
for page, kind in matches:
setattr(obj, "{}_page".format(kind), int(page))
setattr(
obj, "pages", getattr(obj, "last_page", previous_page_requested)
)
setattr(obj, "pages", getattr(obj, "last_page", previous_page_requested))
setattr(obj, "has_pages", True)
setattr(
obj,
"current_page",
previous_page_requested or getattr(obj, "next_page") - 1,
)
setattr(obj, "current_page", previous_page_requested or getattr(obj, "next_page") - 1)
setattr(obj, "is_last_page", obj.pages == obj.current_page)
else:
setattr(obj, "has_pages", False)
@@ -135,7 +111,5 @@ class Octokit(Base):
yield response.json
if hasattr(response, "is_last_page"):
while not response.is_last_page:
response = self.set_pages(
obj(page=response.next_page, **kwargs), response.next_page
)
response = self.set_pages(obj(page=response.next_page, **kwargs), response.next_page)
yield response.json
src/octokit/base.py
+14 -45
View File
@@ -15,10 +15,7 @@ class Base(object):
base_url = "https://api.github.com"
def __init__(self):
self.headers = {
"accept": "application/vnd.github.v3+json",
"Content-Type": "application/json",
}
self.headers = {"accept": "application/vnd.github.v3+json", "Content-Type": "application/json"}
self._attribute_cache = defaultdict(dict)
def _get_headers(self, method_headers):
@@ -30,17 +27,13 @@ class Base(object):
return required_params
def validate(self, parameters, definition):
self.validate_required_parameters(
parameters, self.get_required_parameters(definition)
)
self.validate_required_parameters(parameters, self.get_required_parameters(definition))
schema = {}
if definition.get("requestBody"):
schema = definition["requestBody"]["content"]["application/json"]["schema"]
self.validate_schema(parameters, schema)
properties = schema.get("properties", {})
valid_parameters = [p["name"] for p in definition.get("parameters")] + list(
properties.keys()
)
valid_parameters = [p["name"] for p in definition.get("parameters")] + list(properties.keys())
self.validate_other_parameters(parameters, valid_parameters, properties)
return True
@@ -55,9 +48,7 @@ class Base(object):
self.validate_enum(parameter, value, properties)
def validate_enum(self, parameter, value, properties):
if properties.get(parameter).get("enum") and value not in properties.get(
parameter
).get("enum"):
if properties.get(parameter).get("enum") and value not in properties.get(parameter).get("enum"):
message = "{} is not a valid option for {}; must be one of {}".format(
value, parameter, properties.get(parameter).get("enum")
)
@@ -68,9 +59,7 @@ class Base(object):
self.validate_list(parameters, schema)
if isinstance(parameters, dict):
self.validate_dict(parameters, schema)
self.validate_required_parameters(
parameters, self.get_required_schema_properties(schema)
)
self.validate_required_parameters(parameters, self.get_required_schema_properties(schema))
self.type_match(parameters, schema.get("type"))
def validate_dict(self, parameters, schema):
@@ -99,11 +88,7 @@ class Base(object):
return schema.get("required")
def get_required_parameters(self, definition):
return [
p.get("name")
for p in definition.get("parameters")
if p.get("required") and p.get("in") in ["path"]
]
return [p.get("name") for p in definition.get("parameters") if p.get("required") and p.get("in") in ["path"]]
def validate_required_parameters(self, parameters, required_parameters):
for required_parameter in required_parameters or []:
@@ -145,9 +130,7 @@ class Base(object):
data = {}
if parameter.get("in") in ["query", "body"] and not kwargs.get(parameter_name):
if parameter.get("schema", parameter).get("default") is not None:
data[parameter_name] = self._get_parameter_for_type(
parameter.get("schema", parameter)
)
data[parameter_name] = self._get_parameter_for_type(parameter.get("schema", parameter))
return data
def _get_array_data(self, parameter_name, parameter, kwargs):
@@ -199,9 +182,7 @@ class Base(object):
def _setup_installation_authentication(self, kwargs):
assert kwargs["app_id"]
assert kwargs["private_key"]
self.token, self.expires_at = self._app_auth_get_token(
kwargs["app_id"], kwargs["private_key"]
)
self.token, self.expires_at = self._app_auth_get_token(kwargs["app_id"], kwargs["private_key"])
self.auth = kwargs["auth"]
self.headers["accept"] = "application/vnd.github.machine-man-preview+json"
@@ -219,12 +200,8 @@ class Base(object):
}
installation_url = "{}/app/installations".format(self.base_url)
installations = requests.get(installation_url, headers=headers).json()
self.installation_id = [
x.get("id") for x in installations if str(x.get("app_id")) == app_id
].pop()
installation_token_url = "{}/installations/{}/access_tokens".format(
self.base_url, self.installation_id
)
self.installation_id = [x.get("id") for x in installations if str(x.get("app_id")) == app_id].pop()
installation_token_url = "{}/installations/{}/access_tokens".format(self.base_url, self.installation_id)
response = requests.post(installation_token_url, headers=headers).json()
return response["token"], response["expires_at"]
@@ -241,15 +218,9 @@ class Base(object):
return {"auth": (self.username, self.password)}
if getattr(self, "auth", None) in ["app", "token", "installation"]:
_headers = {
"app": {
"Authorization": "Bearer {}".format(getattr(self, "jwt", None))
},
"token": {
"Authorization": "token {}".format(getattr(self, "token", None))
},
"installation": {
"Authorization": "token {}".format(getattr(self, "token", None))
},
"app": {"Authorization": "Bearer {}".format(getattr(self, "jwt", None))},
"token": {"Authorization": "token {}".format(getattr(self, "token", None))},
"installation": {"Authorization": "token {}".format(getattr(self, "token", None))},
}
headers = requests_kwargs["headers"]
headers.update(_headers.get(getattr(self, "auth", None)))
@@ -259,9 +230,7 @@ class Base(object):
def _get_parameters(self, definition, method):
p = {}
if definition.get("requestBody"):
schema = definition.get("requestBody")["content"]["application/json"][
"schema"
]
schema = definition.get("requestBody")["content"]["application/json"]["schema"]
if schema["type"] == "object":
body_parameters = {}
for k, v in schema["properties"].items():
src/octokit/utils.py
+1 -7
View File
@@ -5,13 +5,7 @@ import re
def snake_case(string):
# From https://gist.github.com/jaytaylor/3660565#gistcomment-2271689
return (
re.compile(r"(?!^)(?<!_)([A-Z])")
.sub(r"_\1", string)
.lower()
.replace(" ", "_")
.replace("-", "_")
)
return re.compile(r"(?!^)(?<!_)([A-Z])").sub(r"_\1", string).lower().replace(" ", "_").replace("-", "_")
def get_json_data(filename):
src/octokit/webhook.py
+2 -6
View File
@@ -9,9 +9,7 @@ from octokit_routes import webhook_names
def valid_signature(headers, payload, secret):
encoding = "utf-8"
algo, sig = headers.get("X-Hub-Signature").split("=")
digest = hmac.new(
secret.encode(encoding), payload.encode(encoding), getattr(hashlib, algo)
).hexdigest()
digest = hmac.new(secret.encode(encoding), payload.encode(encoding), getattr(hashlib, algo)).hexdigest()
return hmac.compare_digest(sig.encode(encoding), digest.encode(encoding))
@@ -40,9 +38,7 @@ def valid_headers(headers, events, verify_user_agent):
return True
def verify(
headers, payload, secret, events=None, verify_user_agent=False, return_app_id=False
):
def verify(headers, payload, secret, events=None, verify_user_agent=False, return_app_id=False):
if not valid_headers(headers, events, verify_user_agent):
return False
validity = valid_signature(headers, payload, secret)
tests/test_auth.py
+11 -34
View File
@@ -28,9 +28,9 @@ class TestAuth(object):
def test_basic_auth_used_if_set(self, mocker):
mocker.patch("requests.get")
Octokit(
auth="basic", username="myuser", password="mypassword"
).oauth_authorizations.get_authorization(authorization_id=100)
Octokit(auth="basic", username="myuser", password="mypassword").oauth_authorizations.get_authorization(
authorization_id=100
)
requests.get.assert_called_once_with(
"https://api.github.com/authorizations/100",
params={},
@@ -51,27 +51,16 @@ class TestAuth(object):
def test_token_auth_used_if_set(self, mocker):
mocker.patch("requests.get")
Octokit(auth="token", token="yak").oauth_authorizations.get_authorization(
authorization_id=100
)
Octokit(auth="token", token="yak").oauth_authorizations.get_authorization(authorization_id=100)
headers = dict(ChainMap(Octokit().headers, {"Authorization": "token yak"}))
requests.get.assert_called_once_with(
"https://api.github.com/authorizations/100", params={}, headers=headers
)
requests.get.assert_called_once_with("https://api.github.com/authorizations/100", params={}, headers=headers)
def test_can_set_installation_authentication(self, mocker):
Request = namedtuple("Request", ["json"])
get = mocker.patch("requests.get")
get.return_value = Request(
json=lambda: [{"id": 13, "app_id": 1}, {"id": 37, "app_id": 42}]
)
get.return_value = Request(json=lambda: [{"id": 13, "app_id": 1}, {"id": 37, "app_id": 42}])
post = mocker.patch("requests.post")
post.return_value = Request(
json=lambda: {
"token": "v1.1f699f1069f60",
"expires_at": "2016-07-11T22:14:10Z",
}
)
post.return_value = Request(json=lambda: {"token": "v1.1f699f1069f60", "expires_at": "2016-07-11T22:14:10Z"})
with open(os.path.join(os.path.dirname(__file__), "test.pem"), "r") as f:
private_key = f.read()
sut = Octokit(auth="installation", app_id="42", private_key=private_key)
@@ -92,16 +81,9 @@ class TestAuth(object):
def test_installation_token_is_used_if_set(self, mocker):
Request = namedtuple("Request", ["json"])
get = mocker.patch("requests.get")
get.return_value = Request(
json=lambda: [{"id": 13, "app_id": 1}, {"id": 37, "app_id": 42}]
)
get.return_value = Request(json=lambda: [{"id": 13, "app_id": 1}, {"id": 37, "app_id": 42}])
post = mocker.patch("requests.post")
post.return_value = Request(
json=lambda: {
"token": "v1.1f699f1069f60",
"expires_at": "2016-07-11T22:14:10Z",
}
)
post.return_value = Request(json=lambda: {"token": "v1.1f699f1069f60", "expires_at": "2016-07-11T22:14:10Z"})
with open(os.path.join(os.path.dirname(__file__), "test.pem"), "r") as f:
private_key = f.read()
sut = Octokit(auth="installation", app_id="42", private_key=private_key)
@@ -113,9 +95,7 @@ class TestAuth(object):
"Authorization": "token v1.1f699f1069f60",
"accept": "application/vnd.github.machine-man-preview+json",
}
requests.get.assert_called_once_with(
"https://api.github.com/authorizations/100", params={}, headers=headers
)
requests.get.assert_called_once_with("https://api.github.com/authorizations/100", params={}, headers=headers)
def test_cannot_set_app_authentication_with_out_required_data(self):
with pytest.raises(KeyError):
@@ -151,9 +131,6 @@ class TestAuth(object):
Octokit().users.list_followers_for_user(username="octokit")
requests.get.assert_called_once_with(
"https://api.github.com/users/octokit/followers",
headers={
"Content-Type": "application/json",
"accept": "application/vnd.github.v3+json",
},
headers={"Content-Type": "application/json", "accept": "application/vnd.github.v3+json"},
params={"page": 1, "per_page": 30},
)
tests/test_methods.py
+33 -114
View File
@@ -23,10 +23,7 @@ class TestClientMethods(object):
assert Octokit().oauth_authorizations.list_grants.__name__ == "list_grants"
def test_method_is_accessible_also_by_snake_case_name(self):
assert (
Octokit().oauth_authorizations.list_your_grants.__name__
== "list_your_grants"
)
assert Octokit().oauth_authorizations.list_your_grants.__name__ == "list_your_grants"
def test_method_calls_requests(self, mocker):
mocker.patch("requests.get")
@@ -51,9 +48,7 @@ class TestClientMethods(object):
mocker.patch("requests.get")
Octokit().oauth_authorizations.get_authorization(authorization_id=100)
requests.get.assert_called_once_with(
"https://api.github.com/authorizations/100",
params={},
headers=Octokit().headers,
"https://api.github.com/authorizations/100", params={}, headers=Octokit().headers
)
def test_request_has_body_parameters(self, mocker):
@@ -61,9 +56,7 @@ class TestClientMethods(object):
data = {"note": "remind me", "scopes": ["public_repo"]}
create = Octokit().oauth_authorizations.create_authorization(**data)
requests.post.assert_called_once_with(
"https://api.github.com/authorizations",
data=json.dumps(data, sort_keys=True),
headers=create.headers,
"https://api.github.com/authorizations", data=json.dumps(data, sort_keys=True), headers=create.headers
)
def test_must_include_required_body_parameters(self):
@@ -87,13 +80,7 @@ class TestClientMethods(object):
def test_must_include_required_array_sub_parameters_when_used(self, mocker):
mocker.patch("requests.get")
data = {
"owner": "owner",
"repo": "repo",
"name": "name",
"head_sha": "master",
"actions": [],
}
data = {"owner": "owner", "repo": "repo", "name": "name", "head_sha": "master", "actions": []}
with pytest.raises(errors.OctokitParameterError) as e:
Octokit().checks.create(**data)
assert "property is missing required items" == str(e.value)
@@ -101,9 +88,7 @@ class TestClientMethods(object):
with pytest.raises(errors.OctokitParameterError) as e:
Octokit().checks.create(**data)
assert "description is a required parameter" == str(e.value)
data.update(
{"actions": [{"label": "blah", "description": "x", "identifier": "a"}]}
)
data.update({"actions": [{"label": "blah", "description": "x", "identifier": "a"}]})
Octokit().checks.create(**data)
def test_schema_types_must_match(self, mocker):
@@ -117,17 +102,11 @@ class TestClientMethods(object):
}
with pytest.raises(errors.OctokitParameterError) as e:
Octokit().checks.create(**data)
assert (
f'dict type does not match the schema type of array for the data of {data["actions"]}'
== str(e.value)
)
assert f'dict type does not match the schema type of array for the data of {data["actions"]}' == str(e.value)
def test_use_default_parameter_values(self, mocker):
mocker.patch("requests.get")
headers = {
"Content-Type": "application/json",
"accept": "application/vnd.github.v3+json",
}
headers = {"Content-Type": "application/json", "accept": "application/vnd.github.v3+json"}
data = {
"visibility": "all",
"affiliation": "owner,collaborator,organization_member",
@@ -137,16 +116,11 @@ class TestClientMethods(object):
"page": 1,
}
Octokit().repos.list_for_authenticated_user()
requests.get.assert_called_once_with(
"https://api.github.com/user/repos", params=data, headers=headers
)
requests.get.assert_called_once_with("https://api.github.com/user/repos", params=data, headers=headers)
def test_deprecated_methods_are_available(self, mocker):
mocker.patch("requests.get")
headers = {
"Content-Type": "application/json",
"accept": "application/vnd.github.v3+json",
}
headers = {"Content-Type": "application/json", "accept": "application/vnd.github.v3+json"}
data = {
"visibility": "all",
"affiliation": "owner,collaborator,organization_member",
@@ -156,106 +130,62 @@ class TestClientMethods(object):
"page": 1,
}
Octokit().repos.list()
requests.get.assert_called_once_with(
"https://api.github.com/user/repos", params=data, headers=headers
)
requests.get.assert_called_once_with("https://api.github.com/user/repos", params=data, headers=headers)
def test_use_passed_value_instead_of_default_parameter_values(self, mocker):
mocker.patch("requests.get")
headers = {
"Content-Type": "application/json",
"accept": "application/vnd.github.v3+json",
}
headers = {"Content-Type": "application/json", "accept": "application/vnd.github.v3+json"}
data = {"sort": "updated", "per_page": 30, "page": 1}
Octokit().issues.list_comments_for_repo(
owner="testUser", repo="testRepo", **data
)
Octokit().issues.list_comments_for_repo(owner="testUser", repo="testRepo", **data)
requests.get.assert_called_once_with(
"https://api.github.com/repos/testUser/testRepo/issues/comments",
params=data,
headers=headers,
"https://api.github.com/repos/testUser/testRepo/issues/comments", params=data, headers=headers
)
def test_validate_enum_values(self):
with pytest.raises(errors.OctokitParameterError) as e:
Octokit().issues.update(
owner="testUser", repo="testRepo", issue_number=1, state="closeddddd"
)
assert (
"closeddddd is not a valid option for state; must be one of ['open', 'closed']"
== str(e.value)
)
Octokit().issues.update(owner="testUser", repo="testRepo", issue_number=1, state="closeddddd")
assert "closeddddd is not a valid option for state; must be one of ['open', 'closed']" == str(e.value)
def test_validate_boolean_values(self, mocker):
mocker.patch("requests.post")
Octokit().repos.create_deployment(
owner="testUser", repo="testRepo", ref="abc123"
)
Octokit().repos.create_deployment(owner="testUser", repo="testRepo", ref="abc123")
data = '{"auto_merge": true, "description": "", "environment": "production", "payload": "", "ref": "abc123", "task": "deploy", "transient_environment": false}' # noqa E501
headers = {
"Content-Type": "application/json",
"accept": "application/vnd.github.v3+json",
}
headers = {"Content-Type": "application/json", "accept": "application/vnd.github.v3+json"}
requests.post.assert_called_once_with(
"https://api.github.com/repos/testUser/testRepo/deployments",
data=data,
headers=headers,
"https://api.github.com/repos/testUser/testRepo/deployments", data=data, headers=headers
)
def test_non_default_params_not_in_the_url_for_get_requests_go_in_the_query_string(
self, mocker
):
def test_non_default_params_not_in_the_url_for_get_requests_go_in_the_query_string(self, mocker):
mocker.patch("requests.get")
params = {"page": 2, "per_page": 30}
Octokit().oauth_authorizations.list_grants(page=2)
requests.get.assert_called_once_with(
"https://api.github.com/applications/grants",
params=params,
headers=Octokit().headers,
"https://api.github.com/applications/grants", params=params, headers=Octokit().headers
)
def test_does_not_use_previous_values(self, mocker):
mocker.patch("requests.patch")
mocker.patch("requests.post")
headers = {
"accept": "application/vnd.github.v3+json",
"Content-Type": "application/json",
}
headers = {"accept": "application/vnd.github.v3+json", "Content-Type": "application/json"}
data = {"state": "closed"}
issue = Octokit().issues.update(
owner="testUser", repo="testRepo", issue_number=1, **data
)
issue = Octokit().issues.update(owner="testUser", repo="testRepo", issue_number=1, **data)
requests.patch.assert_called_with(
"https://api.github.com/repos/testUser/testRepo/issues/1",
data=json.dumps(data),
headers=headers,
)
issue.pulls.create(
owner="user", head="branch", base="master", title="Title", repo="testRepo"
"https://api.github.com/repos/testUser/testRepo/issues/1", data=json.dumps(data), headers=headers
)
issue.pulls.create(owner="user", head="branch", base="master", title="Title", repo="testRepo")
requests.post.assert_called_with(
"https://api.github.com/repos/user/testRepo/pulls",
data=json.dumps(
{"base": "master", "head": "branch", "title": "Title"}, sort_keys=True
),
headers={
"Content-Type": "application/json",
"accept": "application/vnd.github.v3+json",
},
data=json.dumps({"base": "master", "head": "branch", "title": "Title"}, sort_keys=True),
headers={"Content-Type": "application/json", "accept": "application/vnd.github.v3+json"},
)
def test_returned_object_is_not_self_but_a_copy_of_self(self, mocker):
mocker.patch("requests.patch")
headers = {
"accept": "application/vnd.github.v3+json",
"Content-Type": "application/json",
}
headers = {"accept": "application/vnd.github.v3+json", "Content-Type": "application/json"}
octokit = Octokit()
sut = octokit.issues.update(owner="testUser", repo="testRepo", issue_number=1)
requests.patch.assert_called_once_with(
"https://api.github.com/repos/testUser/testRepo/issues/1",
data="{}",
headers=headers,
"https://api.github.com/repos/testUser/testRepo/issues/1", data="{}", headers=headers
)
assert sut.__class__.__name__ == "Octokit"
assert sut != octokit
@@ -316,33 +246,22 @@ class TestClientMethods(object):
def test_can_pass_in_optional_headers(self, mocker):
mocker.patch("requests.get")
headers = {
"accept": "application/vnd.github.ant-man-preview+json",
"Content-Type": "application/json",
}
headers = {"accept": "application/vnd.github.ant-man-preview+json", "Content-Type": "application/json"}
Octokit().oauth_authorizations.get_authorization(
authorization_id=100,
headers={"accept": "application/vnd.github.ant-man-preview+json"},
)
requests.get.assert_called_once_with(
"https://api.github.com/authorizations/100", params={}, headers=headers
authorization_id=100, headers={"accept": "application/vnd.github.ant-man-preview+json"}
)
requests.get.assert_called_once_with("https://api.github.com/authorizations/100", params={}, headers=headers)
def test_dictionary_keys_are_validated(self, mocker):
mocker.patch("requests.put")
headers = {
"accept": "application/vnd.github.v3+json",
"Content-Type": "application/json",
}
headers = {"accept": "application/vnd.github.v3+json", "Content-Type": "application/json"}
data = {
"required_status_checks": {"strict": True, "contexts": []},
"required_pull_request_reviews": {"dismiss_stale_reviews": True},
"enforce_admins": True,
"restrictions": {"users": [], "teams": []},
}
Octokit().repos.update_branch_protection(
owner="user", repo="repo", branch="branch", **data
)
Octokit().repos.update_branch_protection(owner="user", repo="repo", branch="branch", **data)
requests.put.assert_called_with(
"https://api.github.com/repos/user/repo/branches/branch/protection",
data=json.dumps(data, sort_keys=True),
tests/test_validation.py
+8 -40
View File
@@ -29,23 +29,11 @@ class TestBase(object):
assert Base().validate(attrs, self.definition)
def test_validate_nested_request_body_properties(self):
attrs = {
"owner": "me",
"repo": "my_repo",
"name": "blah",
"head_sha": "master",
"output": {},
}
attrs = {"owner": "me", "repo": "my_repo", "name": "blah", "head_sha": "master", "output": {}}
with pytest.raises(errors.OctokitParameterError) as e:
Base().validate(attrs, self.definition)
assert "title is a required parameter" == str(e.value)
attrs = {
"owner": "me",
"repo": "my_repo",
"name": "blah",
"head_sha": "master",
"output": {"title": "here"},
}
attrs = {"owner": "me", "repo": "my_repo", "name": "blah", "head_sha": "master", "output": {"title": "here"}}
with pytest.raises(errors.OctokitParameterError) as e:
Base().validate(attrs, self.definition)
assert "summary is a required parameter" == str(e.value)
@@ -66,10 +54,7 @@ class TestBase(object):
"name": "accept",
"description": "This API is under preview and subject to change.",
"in": "header",
"schema": {
"type": "string",
"default": "application/vnd.github.antiope-preview+json",
},
"schema": {"type": "string", "default": "application/vnd.github.antiope-preview+json"},
"required": True,
},
{
@@ -97,10 +82,7 @@ class TestBase(object):
"type": "string",
"description": 'The name of the check. For example, "code-coverage".',
},
"head_sha": {
"type": "string",
"description": "The SHA of the commit.",
},
"head_sha": {"type": "string", "description": "The SHA of the commit."},
"details_url": {"type": "string"},
"external_id": {
"type": "string",
@@ -127,10 +109,7 @@ class TestBase(object):
"output": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "The title of the check run.",
},
"title": {"type": "string", "description": "The title of the check run."},
"summary": {"type": "string"},
"text": {"type": "string"},
"annotations": {
@@ -151,11 +130,7 @@ class TestBase(object):
"end_column": {"type": "integer"},
"annotation_level": {
"type": "string",
"enum": [
"notice",
"warning",
"failure",
],
"enum": ["notice", "warning", "failure"],
},
"message": {"type": "string"},
"title": {"type": "string"},
@@ -203,11 +178,7 @@ class TestBase(object):
"description": {"type": "string"},
"identifier": {"type": "string"},
},
"required": [
"label",
"description",
"identifier",
],
"required": ["label", "description", "identifier"],
},
},
},
@@ -226,10 +197,7 @@ class TestBase(object):
"name": "accept",
"description": "This API is under preview and subject to change.",
"in": "header",
"schema": {
"type": "string",
"default": "application/vnd.github.antiope-preview+json",
},
"schema": {"type": "string", "default": "application/vnd.github.antiope-preview+json"},
"required": True,
},
{
tests/test_webhook.py
+5 -24
View File
@@ -77,9 +77,7 @@ class TestWebhook(object):
payload = ""
secret = "secret"
events = ["push"]
assert webhook.verify(
headers, payload, secret, events=events, verify_user_agent=True
)
assert webhook.verify(headers, payload, secret, events=events, verify_user_agent=True)
def test_verifies_user_agent(self):
headers = {
@@ -91,12 +89,7 @@ class TestWebhook(object):
payload = ""
secret = "secret"
events = ["push"]
assert (
webhook.verify(
headers, payload, secret, events=events, verify_user_agent=True
)
is False
)
assert webhook.verify(headers, payload, secret, events=events, verify_user_agent=True) is False
def test_verify_ping_event(self):
headers = {
@@ -106,21 +99,11 @@ class TestWebhook(object):
"User-Agent": "GitHub-Hookshot/",
}
payload = json.dumps(
{
"hook": {
"type": "App",
"id": 11,
"active": True,
"events": ["pull_request"],
"app_id": 42,
}
}
{"hook": {"type": "App", "id": 11, "active": True, "events": ["pull_request"], "app_id": 42}}
)
payload = '{"hook": {"events": ["pull_request"], "app_id": 42, "id": 11, "active": true, "type": "App"}}'
secret = "secret"
app_id = webhook.verify(
headers, payload, secret, events=["*"], return_app_id=True
)
app_id = webhook.verify(headers, payload, secret, events=["*"], return_app_id=True)
assert app_id == 42
def test_can_request_app_id_be_returned_on_non_ping_events(self):
@@ -132,6 +115,4 @@ class TestWebhook(object):
}
payload = ""
secret = "secret"
assert webhook.verify(
headers, payload, secret, events=["*"], return_app_id=True
)
assert webhook.verify(headers, payload, secret, events=["*"], return_app_id=True)