Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix Variable and Secret url bugs #2835

Merged
merged 12 commits into from
Dec 5, 2023
Merged
40 changes: 35 additions & 5 deletions github/PaginatedList.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
# #
################################################################################
from typing import Any, Dict, Generic, Iterator, List, Optional, Type, TypeVar, Union
from typing import Any, Callable, Dict, Generic, Iterator, List, Optional, Type, TypeVar, Union
from urllib.parse import parse_qs

from github.GithubObject import GithubObject
Expand Down Expand Up @@ -140,6 +140,7 @@ def __init__(
list_item: str = "items",
firstData: Optional[Any] = None,
firstHeaders: Optional[Dict[str, Union[str, int]]] = None,
attributesTransformer: Optional[Callable[[Any], Any]] = None,
AndrewJDawes marked this conversation as resolved.
Show resolved Hide resolved
):
self.__requester = requester
self.__contentClass = contentClass
Expand All @@ -153,12 +154,18 @@ def __init__(
self.__nextParams["per_page"] = self.__requester.per_page
self._reversed = False
self.__totalCount: Optional[int] = None
self._attributesTransformer = attributesTransformer

first_page = []
if firstData is not None and firstHeaders is not None:
first_page = self._getPage(firstData, firstHeaders)
super().__init__(first_page)

def _transformAttribute(self, attribute: Any) -> Any:
if self._attributesTransformer is None:
return attribute
return self._attributesTransformer(attribute)
AndrewJDawes marked this conversation as resolved.
Show resolved Hide resolved

@property
def totalCount(self) -> int:
if not self.__totalCount:
Expand Down Expand Up @@ -202,6 +209,7 @@ def reversed(self) -> "PaginatedList[T]":
self.__firstParams,
self.__headers,
self.__list_item,
attributesTransformer=self._attributesTransformer,
)
r.__reverse()
return r
Expand Down Expand Up @@ -232,13 +240,11 @@ def _getPage(self, data: Any, headers: Dict[str, Any]) -> List[T]:
elif "next" in links:
self.__nextUrl = links["next"]
self.__nextParams = None

if self.__list_item in data:
self.__totalCount = data.get("total_count")
data = data[self.__list_item]

content = [
self.__contentClass(self.__requester, headers, element, completed=False)
self.__contentClass(self.__requester, headers, self._transformAttribute(element), completed=False)
for element in data
if element is not None
]
Expand Down Expand Up @@ -270,5 +276,29 @@ def get_page(self, page: int) -> List[T]:
if self.__list_item in data:
self.__totalCount = data.get("total_count")
data = data[self.__list_item]
return [
self.__contentClass(self.__requester, headers, self._transformAttribute(element), completed=False)
for element in data
]

return [self.__contentClass(self.__requester, headers, element, completed=False) for element in data]
@classmethod
def attributes_transformer_override_from_dictionary(
cls, transformer: Callable[[Dict[str, Any]], Dict[str, Any]], overrides: dict
AndrewJDawes marked this conversation as resolved.
Show resolved Hide resolved
) -> Callable[[Dict[str, Any]], Dict[str, Any]]:
AndrewJDawes marked this conversation as resolved.
Show resolved Hide resolved
def attributes_transformer(attributes: Dict[str, Any]) -> Dict[str, Any]:
AndrewJDawes marked this conversation as resolved.
Show resolved Hide resolved
# Recursively merge overrides with attributes, overriding attributes with overrides
attributes = cls.merge_dicts(attributes, overrides)
return transformer(attributes)

return attributes_transformer

@classmethod
def merge_dicts(cls, d1: Dict[str, Any], d2: Dict[str, Any]) -> Dict[str, Any]:
# clone d1
d1 = d1.copy()
for k, v in d2.items():
if isinstance(v, dict):
d1[k] = cls.merge_dicts(d1.get(k, {}), v)
else:
d1[k] = v
return d1
8 changes: 7 additions & 1 deletion github/Repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -1694,6 +1694,9 @@ def get_secrets(self) -> PaginatedList[github.Secret.Secret]:
self._requester,
f"{self.url}/actions/secrets",
None,
attributesTransformer=PaginatedList.attributes_transformer_override_from_dictionary(
lambda x: x, {"secrets_url": f"{self.url}/actions/secrets"}
),
list_item="secrets",
)

Expand Down Expand Up @@ -1726,7 +1729,7 @@ def create_variable(self, variable_name: str, value: str) -> github.Variable.Var
attributes={
"name": variable_name,
"value": value,
"url": self.url,
"url": f"{self.url}/actions/variables/{variable_name}",
},
completed=False,
)
Expand All @@ -1741,6 +1744,9 @@ def get_variables(self) -> PaginatedList[github.Variable.Variable]:
self._requester,
f"{self.url}/actions/variables",
None,
attributesTransformer=PaginatedList.attributes_transformer_override_from_dictionary(
lambda x: x, {"variables_url": f"{self.url}/actions/variables"}
),
list_item="variables",
)

Expand Down
13 changes: 13 additions & 0 deletions github/Secret.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ def _initAttributes(self) -> None:
self._name: Attribute[str] = NotSet
self._created_at: Attribute[datetime] = NotSet
self._updated_at: Attribute[datetime] = NotSet
self._secrets_url: Attribute[str] = NotSet
self._url: Attribute[str] = NotSet

def __repr__(self) -> str:
Expand Down Expand Up @@ -64,11 +65,21 @@ def updated_at(self) -> datetime:
self._completeIfNotSet(self._updated_at)
return self._updated_at.value

@property
def secrets_url(self) -> str:
"""
:type: string
"""
return self._secrets_url.value

@property
def url(self) -> str:
"""
:type: string
"""
# Construct url from secrets_url and name, if self._url. is not set
if self._url is NotSet:
self._url = self._makeStringAttribute(self.secrets_url + "/" + self.name)
return self._url.value

def delete(self) -> None:
Expand All @@ -85,5 +96,7 @@ def _useAttributes(self, attributes: Dict[str, Any]) -> None:
self._created_at = self._makeDatetimeAttribute(attributes["created_at"])
if "updated_at" in attributes:
self._updated_at = self._makeDatetimeAttribute(attributes["updated_at"])
if "secrets_url" in attributes:
self._secrets_url = self._makeStringAttribute(attributes["secrets_url"])
if "url" in attributes:
self._url = self._makeStringAttribute(attributes["url"])
17 changes: 15 additions & 2 deletions github/Variable.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ def _initAttributes(self) -> None:
self._value: Attribute[str] = NotSet
self._created_at: Attribute[datetime] = NotSet
self._updated_at: Attribute[datetime] = NotSet
self._variables_url: Attribute[str] = NotSet
self._url: Attribute[str] = NotSet

def __repr__(self) -> str:
Expand Down Expand Up @@ -75,11 +76,21 @@ def updated_at(self) -> datetime:
self._completeIfNotSet(self._updated_at)
return self._updated_at.value

@property
def variables_url(self) -> str:
"""
:type: string
"""
return self._variables_url.value

@property
def url(self) -> str:
"""
:type: string
"""
# Construct url from variables_url and name, if self._url. is not set
if self._url is NotSet:
self._url = self._makeStringAttribute(self.variables_url + "/" + self.name)
return self._url.value

def edit(self, value: str) -> bool:
Expand All @@ -96,7 +107,7 @@ def edit(self, value: str) -> bool:
}
status, _, _ = self._requester.requestJson(
"PATCH",
f"{self.url}/actions/variables/{self.name}",
self.url,
input=patch_parameters,
)
return status == 204
Expand All @@ -106,7 +117,7 @@ def delete(self) -> None:
:calls: `DELETE {variable_url} <https://docs.github.com/en/rest/actions/variables>`_
:rtype: None
"""
self._requester.requestJsonAndCheck("DELETE", f"{self.url}/actions/variables/{self.name}")
self._requester.requestJsonAndCheck("DELETE", self.url)

def _useAttributes(self, attributes: dict[str, Any]) -> None:
if "name" in attributes:
Expand All @@ -117,5 +128,7 @@ def _useAttributes(self, attributes: dict[str, Any]) -> None:
self._created_at = self._makeDatetimeAttribute(attributes["created_at"])
if "updated_at" in attributes:
self._updated_at = self._makeDatetimeAttribute(attributes["updated_at"])
if "variables_url" in attributes:
self._variables_url = self._makeStringAttribute(attributes["variables_url"])
if "url" in attributes:
self._url = self._makeStringAttribute(attributes["url"])
18 changes: 17 additions & 1 deletion tests/PaginatedList.py
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,8 @@ def testInterruptedIteration(self):
def testInterruptedIterationInSlice(self):
# No asserts, but checks that only three pages are fetched
count = 0
for element in self.list[:100]: # pragma no branch (exits only by break)
# pragma no branch (exits only by break)
for element in self.list[:100]:
count += 1
if count == 75:
break
Expand Down Expand Up @@ -317,3 +318,18 @@ def testCustomPerPageWithGetPage(self):

def testNoFirstPage(self):
self.assertFalse(next(iter(self.list), None))

def testMergeDicts(self):
self.assertDictEqual(
PaginatedListImpl.merge_dicts(
{"a": 1, "b": 2, "c": 3},
{"c": 4, "d": 5, "e": 6},
),
{"a": 1, "b": 2, "c": 4, "d": 5, "e": 6},
)

def testAttributesTransformerOverrideFromDictionary(self):
input_dict = {"a": 1, "b": 2, "c": 3}
overrides_dict = {"c": 4, "d": 5, "e": 6}
transformer = PaginatedListImpl.attributes_transformer_override_from_dictionary(lambda x: x, overrides_dict)
self.assertDictEqual(transformer(input_dict), {"a": 1, "b": 2, "c": 4, "d": 5, "e": 6})
8 changes: 8 additions & 0 deletions tests/Pickle.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import unittest

import github
from github.PaginatedList import PaginatedList
from github.Repository import Repository

REPO_NAME = "PyGithub/PyGithub"
Expand All @@ -24,3 +25,10 @@ def testPickleRepository(self):
self.assertIsNotNone(repo2._requester._Requester__connection_lock)
self.assertIsNone(repo2._requester._Requester__connection)
self.assertEqual(len(repo2._requester._Requester__custom_connections), 0)

def testPicklePaginatedList(self):
gh = github.Github()
repo = gh.get_repo(REPO_NAME, lazy=True)
branches = repo.get_branches()
branches2 = pickle.loads(pickle.dumps(branches))
self.assertIsInstance(branches2, PaginatedList)
97 changes: 97 additions & 0 deletions tests/ReplayData/Repository.testRepoSecrets.txt

Large diffs are not rendered by default.

75 changes: 75 additions & 0 deletions tests/ReplayData/Repository.testRepoVariables.txt

Large diffs are not rendered by default.

41 changes: 41 additions & 0 deletions tests/Repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -455,6 +455,28 @@ def testCreateSecret(self, encrypt):
secret = self.repo.create_secret("secret-name", "secret-value")
self.assertIsNotNone(secret)

@mock.patch("github.PublicKey.encrypt")
def testRepoSecrets(self, encrypt):
# encrypt returns a non-deterministic value, we need to mock it so the replay data matches
encrypt.return_value = "M+5Fm/BqTfB90h3nC7F3BoZuu3nXs+/KtpXwxm9gG211tbRo0F5UiN0OIfYT83CKcx9oKES9Va4E96/b"
# GitHub will always capitalize the secret name
secrets = (("SECRET_NAME_ONE", "secret-value-one"), ("SECRET_NAME_TWO", "secret-value-two"))
repo = self.g.get_repo("AndrewJDawes/PyGithub")
for matched_repo_secret in secrets:
repo.create_secret(matched_repo_secret[0], matched_repo_secret[1])
repo.update()
repo_secrets = repo.get_secrets()
matched_repo_secrets = []
for matched_repo_secret in secrets:
for repo_secret in repo_secrets:
# GitHub will always capitalize the secret name, may be best to uppercase test data for comparison
if repo_secret.name == matched_repo_secret[0].upper():
matched_repo_secrets.append(repo_secret)
break
self.assertEqual(len(matched_repo_secrets), len(secrets))
for matched_repo_secret in matched_repo_secrets:
matched_repo_secret.delete()

def testCodeScanAlerts(self):
codescan_alerts = self.repo.get_codescan_alerts()
self.assertListKeyEqual(
Expand Down Expand Up @@ -1853,6 +1875,25 @@ def testRepoVariable(self):
self.assertTrue(variable.edit("variable-value123"))
variable.delete()

def testRepoVariables(self):
# GitHub will always capitalize the variable name
variables = (("VARIABLE_NAME_ONE", "variable-value-one"), ("VARIABLE_NAME_TWO", "variable-value-two"))
repo = self.g.get_repo("AndrewJDawes/PyGithub")
for variable in variables:
repo.create_variable(variable[0], variable[1])
repo.update()
repo_variables = repo.get_variables()
matched_repo_variables = []
for variable in variables:
for repo_variable in repo_variables:
# GitHub will always capitalize the variable name, may be best to uppercase test data for comparison
if repo_variable.name == variable[0].upper() and repo_variable.value == variable[1]:
matched_repo_variables.append(repo_variable)
break
self.assertEqual(len(matched_repo_variables), len(variables))
for matched_repo_variable in matched_repo_variables:
matched_repo_variable.delete()


class LazyRepository(Framework.TestCase):
def setUp(self):
Expand Down