From 12f710a14d083e380beca8b614e89d280471da5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Ricks?= Date: Mon, 23 Oct 2023 11:46:25 +0200 Subject: [PATCH] GitHub Users API (#909) * Sort GitHub API properties alphabetically * Add: Add most important parts of the GitHub users REST API We want to be able to get information about the users and their login information. --- pontos/github/api/api.py | 20 ++- pontos/github/api/users.py | 289 +++++++++++++++++++++++++++++++ pontos/github/models/user.py | 67 +++++++ tests/github/api/test_users.py | 264 ++++++++++++++++++++++++++++ tests/github/models/test_user.py | 67 +++++++ 5 files changed, 701 insertions(+), 6 deletions(-) create mode 100644 pontos/github/api/users.py create mode 100644 pontos/github/models/user.py create mode 100644 tests/github/api/test_users.py create mode 100644 tests/github/models/test_user.py diff --git a/pontos/github/api/api.py b/pontos/github/api/api.py index 4298492e..d8dd7ebc 100644 --- a/pontos/github/api/api.py +++ b/pontos/github/api/api.py @@ -40,6 +40,7 @@ from pontos.github.api.secret_scanning import GitHubAsyncRESTSecretScanning from pontos.github.api.tags import GitHubAsyncRESTTags from pontos.github.api.teams import GitHubAsyncRESTTeams +from pontos.github.api.users import GitHubAsyncRESTUsers from pontos.github.api.workflows import GitHubAsyncRESTWorkflows from pontos.helper import deprecated @@ -95,6 +96,13 @@ def branches(self) -> GitHubAsyncRESTBranches: """ return GitHubAsyncRESTBranches(self._client) + @property + def code_scanning(self) -> GitHubAsyncRESTCodeScanning: + """ + Code scanning related API + """ + return GitHubAsyncRESTCodeScanning(self._client) + @property def contents(self) -> GitHubAsyncRESTContent: """ @@ -168,11 +176,11 @@ def secret_scanning(self) -> GitHubAsyncRESTSecretScanning: return GitHubAsyncRESTSecretScanning(self._client) @property - def code_scanning(self) -> GitHubAsyncRESTCodeScanning: + def search(self) -> GitHubAsyncRESTSearch: """ - Code scanning related API + Search related API """ - return GitHubAsyncRESTCodeScanning(self._client) + return GitHubAsyncRESTSearch(self._client) @property def teams(self) -> GitHubAsyncRESTTeams: @@ -189,11 +197,11 @@ def tags(self) -> GitHubAsyncRESTTags: return GitHubAsyncRESTTags(self._client) @property - def search(self) -> GitHubAsyncRESTSearch: + def users(self) -> GitHubAsyncRESTUsers: """ - Search related API + Users related API """ - return GitHubAsyncRESTSearch(self._client) + return GitHubAsyncRESTUsers(self._client) async def __aenter__(self) -> "GitHubAsyncRESTApi": await self._client.__aenter__() diff --git a/pontos/github/api/users.py b/pontos/github/api/users.py new file mode 100644 index 00000000..474bce2a --- /dev/null +++ b/pontos/github/api/users.py @@ -0,0 +1,289 @@ +# SPDX-FileCopyrightText: 2023 Greenbone AG +# +# SPDX-License-Identifier: GPL-3.0-or-later + +from typing import AsyncIterator, Union + +from pontos.github.api.client import GitHubAsyncREST +from pontos.github.api.helper import JSON_OBJECT +from pontos.github.models.base import User +from pontos.github.models.user import ( + EmailInformation, + SSHPublicKey, + SSHPublicKeyExtended, +) + + +class GitHubAsyncRESTUsers(GitHubAsyncREST): + async def users(self) -> AsyncIterator[User]: + """ + + https://docs.github.com/en/rest/users/users#list-users + + Args: + username: The handle for the GitHub user account + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + An async iterator yielding user information + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + async for user in api.users.users(): + print(user) + """ + api = "/users" + params = {"per_page": "100"} + async for response in self._client.get_all(api, params=params): + response.raise_for_status() + + for user in response.json(): + yield User.from_dict(user) + + async def user(self, username: str) -> User: + """ + Provide publicly available information about someone with a GitHub + account + + https://docs.github.com/en/rest/users/users#get-a-user + + Args: + username: The handle for the GitHub user account + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + Information about the user + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + user = await api.users.user("foo") + print(user) + """ + api = f"/users/{username}" + response = await self._client.get(api) + response.raise_for_status() + return User.from_dict(response.json()) + + async def current_user(self) -> User: + """ + Get the current authenticated user + + https://docs.github.com/en/rest/users/users#get-the-authenticated-user + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + Information about the user + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + user = await api.users.current_user() + print(user) + """ + api = "/user" + response = await self._client.get(api) + response.raise_for_status() + return User.from_dict(response.json()) + + async def user_keys(self, username: str) -> AsyncIterator[SSHPublicKey]: + """ + List the verified public SSH keys for a user + + https://docs.github.com/en/rest/users/keys#list-public-keys-for-a-user + + Args: + username: The handle for the GitHub user account + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + An async iterator yielding ssh key information + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + async for key in api.users.user_keys("foo"): + print(key) + """ + api = f"/users/{username}/keys" + params = {"per_page": "100"} + async for response in self._client.get_all(api, params=params): + response.raise_for_status() + + for key in response.json(): + yield SSHPublicKey.from_dict(key) + + async def keys(self) -> AsyncIterator[SSHPublicKey]: + """ + List the public SSH keys for the authenticated user's GitHub account + + https://docs.github.com/en/rest/users/keys#list-public-ssh-keys-for-the-authenticated-user + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + An async iterator yielding ssh key information + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + async for key in api.users.keys(): + print(key) + """ + api = "/user/keys" + params = {"per_page": "100"} + async for response in self._client.get_all(api, params=params): + response.raise_for_status() + + for key in response.json(): + yield SSHPublicKey.from_dict(key) + + async def emails(self) -> AsyncIterator[EmailInformation]: + """ + List all email addresses of the currently logged in user + + https://docs.github.com/en/rest/users/emails#list-email-addresses-for-the-authenticated-user + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + An async iterator yielding email information + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + async for email in api.users.emails(): + print(email) + """ + api = "/user/emails" + params = {"per_page": "100"} + async for response in self._client.get_all(api, params=params): + response.raise_for_status() + + for email in response.json(): + yield EmailInformation.from_dict(email) + + async def key(self, key_id: Union[str, int]) -> SSHPublicKeyExtended: + """ + View extended details for a single public SSH key + + https://docs.github.com/en/rest/users/keys#get-a-public-ssh-key-for-the-authenticated-user + + Args: + key_id: The unique identifier of the key + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + Extended information about the SSH key + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + key = await api.users.key(123) + print(key) + """ + api = f"/user/keys/{key_id}" + response = await self._client.get(api) + response.raise_for_status() + return SSHPublicKeyExtended.from_dict(response.json()) + + async def delete_key(self, key_id: Union[str, int]) -> None: + """ + Removes a public SSH key from the authenticated user's GitHub account + + https://docs.github.com/en/rest/users/keys#delete-a-public-ssh-key-for-the-authenticated-user + + Args: + key_id: The unique identifier of the key + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + await api.users.delete_key(123) + """ + api = f"/user/keys/{key_id}" + response = await self._client.delete(api) + response.raise_for_status() + + async def create_key(self, title: str, key: str) -> SSHPublicKeyExtended: + """ + Adds a public SSH key to the authenticated user's GitHub account + + https://docs.github.com/en/rest/users/keys#create-a-public-ssh-key-for-the-authenticated-user + + Args: + title: A descriptive name for the new key + key: The public SSH key to add to your GitHub account + + Raises: + HTTPStatusError: A httpx.HTTPStatusError is raised if the request + failed. + + Returns: + Extended information about the SSH key + + Example: + .. code-block:: python + + from pontos.github.api import GitHubAsyncRESTApi + + async with GitHubAsyncRESTApi(token) as api: + key = await api.users.create_key( + "My SSH Public Key", + "2Sg8iYjAxxmI2LvUXpJjkYrMxURPc8r+dB7TJyvv1234" + ) + print(key) + """ + api = "/user/keys" + data: JSON_OBJECT = {"key": key, "title": title} + response = await self._client.post(api, data=data) + response.raise_for_status() + return SSHPublicKeyExtended.from_dict(response.json()) diff --git a/pontos/github/models/user.py b/pontos/github/models/user.py new file mode 100644 index 00000000..adc2c0ce --- /dev/null +++ b/pontos/github/models/user.py @@ -0,0 +1,67 @@ +# SPDX-FileCopyrightText: 2023 Greenbone AG +# +# SPDX-License-Identifier: GPL-3.0-or-later + +from dataclasses import dataclass +from datetime import datetime +from typing import Optional + +from pontos.github.models.base import GitHubModel + + +@dataclass +class SSHPublicKey(GitHubModel): + """ + A public SSH key of a user + + Attributes: + id: ID of the SSH key + key: SSH Key + """ + + id: int + key: str + + +@dataclass +class SSHPublicKeyExtended(GitHubModel): + """ + Extended details of public SSH key of a user + + Attributes: + id: ID of the SSH key + key: SSH Key + url: + title: + created_at + verified: + read_only: + """ + + id: int + key: str + url: str + title: str + created_at: datetime + verified: bool + read_only: bool + + +@dataclass +class EmailInformation(GitHubModel): + """ + Information about an email address stored in GitHub + + Attributes: + email: The email address + primary: True if it is the primary email address of the user + verified: True if the email address is verified + visibility: public, private + """ + + email: str + primary: bool + verified: bool + # visibility should be an enum but the schema didn't define the possible + # values. therefore be safe and just use a str + visibility: Optional[str] = None diff --git a/tests/github/api/test_users.py b/tests/github/api/test_users.py new file mode 100644 index 00000000..4e391dcd --- /dev/null +++ b/tests/github/api/test_users.py @@ -0,0 +1,264 @@ +# SPDX-FileCopyrightText: 2023 Greenbone AG +# +# SPDX-License-Identifier: GPL-3.0-or-later + +# ruff: noqa:E501 + +from pontos.github.api.users import GitHubAsyncRESTUsers +from tests import AsyncIteratorMock, aiter, anext +from tests.github.api import GitHubAsyncRESTTestCase, create_response + + +class GitHubAsyncRESTUsersTestCase(GitHubAsyncRESTTestCase): + api_cls = GitHubAsyncRESTUsers + + async def test_users(self): + response = create_response() + response.json.return_value = [ + { + "login": "octocat", + "id": 1, + "node_id": "MDQ6VXNlcjE=", + "avatar_url": "https://github.com/images/error/octocat_happy.gif", + "gravatar_id": "", + "url": "https://api.github.com/users/octocat", + "html_url": "https://github.com/octocat", + "followers_url": "https://api.github.com/users/octocat/followers", + "following_url": "https://api.github.com/users/octocat/following{/other_user}", + "gists_url": "https://api.github.com/users/octocat/gists{/gist_id}", + "starred_url": "https://api.github.com/users/octocat/starred{/owner}{/repo}", + "subscriptions_url": "https://api.github.com/users/octocat/subscriptions", + "organizations_url": "https://api.github.com/users/octocat/orgs", + "repos_url": "https://api.github.com/users/octocat/repos", + "events_url": "https://api.github.com/users/octocat/events{/privacy}", + "received_events_url": "https://api.github.com/users/octocat/received_events", + "type": "User", + "site_admin": False, + } + ] + + self.client.get_all.return_value = AsyncIteratorMock([response]) + + async_it = aiter(self.api.users()) + user = await anext(async_it) + self.assertEqual(user.id, 1) + + with self.assertRaises(StopAsyncIteration): + await anext(async_it) + + self.client.get_all.assert_called_once_with( + "/users", + params={"per_page": "100"}, + ) + + async def test_user(self): + response = create_response() + response.json.return_value = { + "login": "octocat", + "id": 1, + "node_id": "MDQ6VXNlcjE=", + "avatar_url": "https://github.com/images/error/octocat_happy.gif", + "gravatar_id": "", + "url": "https://api.github.com/users/octocat", + "html_url": "https://github.com/octocat", + "followers_url": "https://api.github.com/users/octocat/followers", + "following_url": "https://api.github.com/users/octocat/following{/other_user}", + "gists_url": "https://api.github.com/users/octocat/gists{/gist_id}", + "starred_url": "https://api.github.com/users/octocat/starred{/owner}{/repo}", + "subscriptions_url": "https://api.github.com/users/octocat/subscriptions", + "organizations_url": "https://api.github.com/users/octocat/orgs", + "repos_url": "https://api.github.com/users/octocat/repos", + "events_url": "https://api.github.com/users/octocat/events{/privacy}", + "received_events_url": "https://api.github.com/users/octocat/received_events", + "type": "User", + "site_admin": False, + "name": "monalisa octocat", + "company": "GitHub", + "blog": "https://github.com/blog", + "location": "San Francisco", + "email": "octocat@github.com", + "hireable": False, + "bio": "There once was...", + "twitter_username": "monatheoctocat", + "public_repos": 2, + "public_gists": 1, + "followers": 20, + "following": 0, + "created_at": "2008-01-14T04:33:35Z", + "updated_at": "2008-01-14T04:33:35Z", + } + self.client.get.return_value = response + + user = await self.api.user( + "octocat", + ) + + self.client.get.assert_awaited_once_with( + "/users/octocat", + ) + + self.assertEqual(user.id, 1) + + async def test_current_user(self): + response = create_response() + response.json.return_value = { + "login": "octocat", + "id": 1, + "node_id": "MDQ6VXNlcjE=", + "avatar_url": "https://github.com/images/error/octocat_happy.gif", + "gravatar_id": "", + "url": "https://api.github.com/users/octocat", + "html_url": "https://github.com/octocat", + "followers_url": "https://api.github.com/users/octocat/followers", + "following_url": "https://api.github.com/users/octocat/following{/other_user}", + "gists_url": "https://api.github.com/users/octocat/gists{/gist_id}", + "starred_url": "https://api.github.com/users/octocat/starred{/owner}{/repo}", + "subscriptions_url": "https://api.github.com/users/octocat/subscriptions", + "organizations_url": "https://api.github.com/users/octocat/orgs", + "repos_url": "https://api.github.com/users/octocat/repos", + "events_url": "https://api.github.com/users/octocat/events{/privacy}", + "received_events_url": "https://api.github.com/users/octocat/received_events", + "type": "User", + "site_admin": False, + "name": "monalisa octocat", + "company": "GitHub", + "blog": "https://github.com/blog", + "location": "San Francisco", + "email": "octocat@github.com", + "hireable": False, + "bio": "There once was...", + "twitter_username": "monatheoctocat", + "public_repos": 2, + "public_gists": 1, + "followers": 20, + "following": 0, + "created_at": "2008-01-14T04:33:35Z", + "updated_at": "2008-01-14T04:33:35Z", + } + self.client.get.return_value = response + + user = await self.api.current_user() + + self.client.get.assert_awaited_once_with("/user") + + self.assertEqual(user.id, 1) + + async def test_user_keys(self): + response = create_response() + response.json.return_value = [{"id": 1, "key": "ssh-rsa AAA..."}] + + self.client.get_all.return_value = AsyncIteratorMock([response]) + + async_it = aiter(self.api.user_keys("foo")) + key = await anext(async_it) + self.assertEqual(key.id, 1) + + with self.assertRaises(StopAsyncIteration): + await anext(async_it) + + self.client.get_all.assert_called_once_with( + "/users/foo/keys", + params={"per_page": "100"}, + ) + + async def test_keys(self): + response = create_response() + response.json.return_value = [{"id": 1, "key": "ssh-rsa AAA..."}] + + self.client.get_all.return_value = AsyncIteratorMock([response]) + + async_it = aiter(self.api.keys()) + key = await anext(async_it) + self.assertEqual(key.id, 1) + + with self.assertRaises(StopAsyncIteration): + await anext(async_it) + + self.client.get_all.assert_called_once_with( + "/user/keys", + params={"per_page": "100"}, + ) + + async def test_emails(self): + response = create_response() + response.json.return_value = [ + { + "email": "octocat@github.com", + "verified": True, + "primary": True, + "visibility": "public", + } + ] + + self.client.get_all.return_value = AsyncIteratorMock([response]) + + async_it = aiter(self.api.emails()) + email = await anext(async_it) + self.assertEqual(email.email, "octocat@github.com") + + with self.assertRaises(StopAsyncIteration): + await anext(async_it) + + self.client.get_all.assert_called_once_with( + "/user/emails", + params={"per_page": "100"}, + ) + + async def test_key(self): + response = create_response() + response.json.return_value = { + "key": "2Sg8iYjAxxmI2LvUXpJjkYrMxURPc8r+dB7TJyvv1234", + "id": 2, + "url": "https://api.github.com/user/keys/2", + "title": "ssh-rsa AAAAB3NzaC1yc2EAAA", + "created_at": "2020-06-11T21:31:57Z", + "verified": False, + "read_only": False, + } + self.client.get.return_value = response + + key = await self.api.key(2) + + self.client.get.assert_awaited_once_with( + "/user/keys/2", + ) + + self.assertEqual(key.id, 2) + + async def test_delete_key(self): + response = create_response() + self.client.get.return_value = response + + await self.api.delete_key(2) + + self.client.delete.assert_awaited_once_with( + "/user/keys/2", + ) + + async def test_create_key(self): + response = create_response() + response.json.return_value = { + "key": "2Sg8iYjAxxmI2LvUXpJjkYrMxURPc8r+dB7TJyvv1234", + "id": 2, + "url": "https://api.github.com/user/keys/2", + "title": "ssh-rsa AAAAB3NzaC1yc2EAAA", + "created_at": "2020-06-11T21:31:57Z", + "verified": False, + "read_only": False, + } + self.client.post.return_value = response + + key = await self.api.create_key( + "ssh-rsa AAAAB3NzaC1yc2EAAA", + "2Sg8iYjAxxmI2LvUXpJjkYrMxURPc8r+dB7TJyvv1234", + ) + + self.client.post.assert_awaited_once_with( + "/user/keys", + data={ + "title": "ssh-rsa AAAAB3NzaC1yc2EAAA", + "key": "2Sg8iYjAxxmI2LvUXpJjkYrMxURPc8r+dB7TJyvv1234", + }, + ) + + self.assertEqual(key.id, 2) diff --git a/tests/github/models/test_user.py b/tests/github/models/test_user.py new file mode 100644 index 00000000..53bb0ecc --- /dev/null +++ b/tests/github/models/test_user.py @@ -0,0 +1,67 @@ +# SPDX-FileCopyrightText: 2023 Greenbone AG +# +# SPDX-License-Identifier: GPL-3.0-or-later + +# ruff: noqa:E501 + +import unittest +from datetime import datetime, timezone + +from pontos.github.models.user import ( + EmailInformation, + SSHPublicKey, + SSHPublicKeyExtended, +) + + +class SSHPublicKeyTestCase(unittest.TestCase): + def test_from_dict(self): + key = SSHPublicKey.from_dict({"id": 1, "key": "ssh-rsa AAA..."}) + + self.assertEqual(key.id, 1) + self.assertEqual(key.key, "ssh-rsa AAA...") + + +class SSHPublicKeyExtendedTestCase(unittest.TestCase): + def test_from_dict(self): + key = SSHPublicKeyExtended.from_dict( + { + "key": "2Sg8iYjAxxmI2LvUXpJjkYrMxURPc8r+dB7TJyvv1234", + "id": 2, + "url": "https://api.github.com/user/keys/2", + "title": "ssh-rsa AAAAB3NzaC1yc2EAAA", + "created_at": "2020-06-11T21:31:57Z", + "verified": False, + "read_only": False, + } + ) + + self.assertEqual(key.id, 2) + self.assertEqual( + key.key, "2Sg8iYjAxxmI2LvUXpJjkYrMxURPc8r+dB7TJyvv1234" + ) + self.assertEqual(key.url, "https://api.github.com/user/keys/2") + self.assertEqual(key.title, "ssh-rsa AAAAB3NzaC1yc2EAAA") + self.assertFalse(key.verified) + self.assertFalse(key.read_only) + self.assertEqual( + key.created_at, + datetime(2020, 6, 11, 21, 31, 57, tzinfo=timezone.utc), + ) + + +class EmailInformationTestCase(unittest.TestCase): + def test_from_dict(self): + email = EmailInformation.from_dict( + { + "email": "octocat@github.com", + "verified": True, + "primary": True, + "visibility": "public", + } + ) + + self.assertEqual(email.email, "octocat@github.com") + self.assertEqual(email.visibility, "public") + self.assertTrue(email.primary) + self.assertTrue(email.verified)