diff --git a/src/lando/main/models/repo.py b/src/lando/main/models/repo.py index 824678951..a4e4dd2d9 100644 --- a/src/lando/main/models/repo.py +++ b/src/lando/main/models/repo.py @@ -275,6 +275,11 @@ def _github_repo_url(self) -> str | None: if self.is_github: return self.url.removesuffix(".git") + @property + def _github_repo_org(self) -> str | None: + if self.is_github: + return self._github_repo_url.split("/")[-2] + @property def git_repo_name(self) -> str: """Provide the bare name of the Git repo.""" diff --git a/src/lando/main/scm/git.py b/src/lando/main/scm/git.py index c35d1e7a4..5c99cef1a 100644 --- a/src/lando/main/scm/git.py +++ b/src/lando/main/scm/git.py @@ -1,4 +1,3 @@ -import asyncio import io import logging import os @@ -11,8 +10,6 @@ from pathlib import Path from typing import Any -from django.conf import settings -from simple_github import AppAuth, AppInstallationAuth from typing_extensions import override from lando.main.scm.commit import CommitData @@ -112,6 +109,8 @@ def push( tags: list[str] | None = None, ): """Push local code to the remote repository.""" + from lando.utils.github import GitHubAPI + push_command = ["push"] if force_push: @@ -134,7 +133,8 @@ def push( repo = match["repo"] repo_name = repo.removesuffix(".git") - token = self._get_github_token(owner, repo_name) + token = GitHubAPI._get_token(owner, repo_name) + if token: push_path = f"https://git:{token}@github.com/{owner}/{repo}" @@ -152,32 +152,6 @@ def push( self._git_run(*push_command, cwd=self.path) - @staticmethod - def _get_github_token(repo_owner: str, repo_name: str) -> str | None: - """Obtain a fresh GitHub token to push to the specified repo. - - This relies on GITHUB_APP_ID and GITHUB_APP_PRIVKEY to be set in the - settings. Returns None if those are missing. - - The app with ID GITHUB_APP_ID needs to be enabled for the target repo. - - """ - app_id = settings.GITHUB_APP_ID - private_key = settings.GITHUB_APP_PRIVKEY - - if not app_id or not private_key: - logger.warning( - f"Missing GITHUB_APP_ID or GITHUB_APP_PRIVKEY to authenticate against GitHub repo {repo_owner}/{repo_name}", - ) - return None - - app_auth = AppAuth( - app_id, - private_key, - ) - session = AppInstallationAuth(app_auth, repo_owner, repositories=[repo_name]) - return asyncio.run(session.get_token()) - def last_commit_for_path(self, path: str) -> str: """Find last commit to touch a path.""" command = ["log", "--max-count=1", "--format=%H", "--", path] diff --git a/src/lando/main/tests/test_git.py b/src/lando/main/tests/test_git.py index 16dbba200..9c413047b 100644 --- a/src/lando/main/tests/test_git.py +++ b/src/lando/main/tests/test_git.py @@ -7,6 +7,7 @@ from collections.abc import Callable from pathlib import Path from textwrap import dedent +from unittest import mock from unittest.mock import MagicMock import pytest @@ -668,17 +669,27 @@ def test_GitSCM_push( ) -def test_GitSCM_push_get_github_token(git_repo: Path): +@pytest.fixture +def mock_github_api_get_token(monkeypatch: pytest.MonkeyPatch): + mock_get_token = MagicMock() + mock_get_token.side_effect = ["ghs_yolo"] + + monkeypatch.setattr("lando.utils.github.GitHubAPI._get_token", mock_get_token) + + return mock_get_token + + +def test_GitSCM_push_get_github_token( + git_repo: Path, mock_github_api_get_token: mock.Mock +): scm = GitSCM(str(git_repo)) scm._git_run = MagicMock() - scm._get_github_token = MagicMock() - scm._get_github_token.side_effect = ["ghs_yolo"] scm.push("https://github.com/some/repo") assert scm._git_run.call_count == 1, "_git_run wasn't called when pushing" assert ( - scm._get_github_token.call_count == 1 + mock_github_api_get_token.call_count == 1 ), "_get_github_token wasn't called when pushing to a github-like URL" assert ( "git:ghs_yolo@github.com" in scm._git_run.call_args[0][1] diff --git a/src/lando/utils/github.py b/src/lando/utils/github.py new file mode 100644 index 000000000..277ff5217 --- /dev/null +++ b/src/lando/utils/github.py @@ -0,0 +1,91 @@ +import asyncio +import logging + +import requests +from django.conf import settings +from simple_github import AppAuth, AppInstallationAuth + +from lando.main.models.repo import Repo + +logger = logging.getLogger(__name__) + + +class GitHubAPI: + """A simple wrapper that authenticates with and communicates with the GitHub API.""" + + GITHUB_BASE_URL = "https://api.github.com" + + def __init__(self, repo: Repo): + repo_owner = repo._github_repo_org + repo_name = repo.git_repo_name + + self.session = requests.Session() + self.session.headers.update( + { + "Authorization": f"Bearer {self._get_token(repo_owner, repo_name)}", + "User-Agent": settings.HTTP_USER_AGENT, + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + ) + + @staticmethod + def _get_token(repo_owner: str, repo_name: str) -> str | None: + """Obtain a fresh GitHub token to push to the specified repo. + + This relies on GITHUB_APP_ID and GITHUB_APP_PRIVKEY to be set in the + environment. Returns None if those are missing. + + The app with ID GITHUB_APP_ID needs to be enabled for the target repo. + + """ + app_id = settings.GITHUB_APP_ID + private_key = settings.GITHUB_APP_PRIVKEY + + if not app_id or not private_key: + logger.warning( + f"Missing GITHUB_APP_ID or GITHUB_APP_PRIVKEY to authenticate against GitHub repo {repo_owner}/{repo_name}", + ) + return None + + app_auth = AppAuth( + app_id, + private_key, + ) + session = AppInstallationAuth(app_auth, repo_owner, repositories=[repo_name]) + return asyncio.run(session.get_token()) + + def get(self, path: str, *args, **kwargs) -> dict: + """Send a GET request to the GitHub API with given args and kwargs.""" + url = f"{self.GITHUB_BASE_URL}/{path}" + return self.session.get(url, *args, **kwargs) + + def post(self, path: str, *args, **kwargs) -> dict: + """Send a POST request to the GitHub API with given args and kwargs.""" + url = f"self.GITHUB_BASE_URL/{path}" + return self.session.post(url, *args, **kwargs) + + +class GitHubAPIClient: + """A convenience client that provides various methods to interact with the GitHub API.""" + + client = None + + def __init__(self, repo: Repo): + self.client = GitHubAPI(repo) + self.repo = repo + self.repo_base_url = ( + f"repos/{self.repo._github_repo_org}/{self.repo.git_repo_name}" + ) + + def _get(self, path: str, *args, **kwargs) -> dict: + result = self.client.get(path, *args, **kwargs) + return result.json() + + def list_pull_requests(self) -> list: + """List all pull requests in the repo.""" + return self._get(f"{self.repo_base_url}/pulls") + + def get_pull_request(self, pull_number: int) -> dict: + """Get a specific pull request from the repo.""" + return self._get(f"{self.repo_base_url}/pulls/{pull_number}")