Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions src/github_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,15 @@
from __future__ import annotations

import contextlib
import logging
import time
from typing import Generator

import jwt
import requests

logger = logging.getLogger(__name__)


class GithubAppToken:
def __init__(self, private_key, app_id) -> None:
Expand All @@ -29,10 +32,18 @@ def get_token(self, installation_id: int) -> Generator[str, None, None]:
# This token expires in an hour
yield resp["token"]
finally:
requests.delete(
"https://api.github.com/installation/token",
headers={"Authorization": f"token {resp['token']}"},
)
try:
requests.delete(
"https://api.github.com/installation/token",
headers={"Authorization": f"token {resp['token']}"},
)
except requests.RequestException:
# Revocation is best-effort; don't fail event processing on cleanup noise.
logger.warning(
"Failed to revoke GitHub App installation token for installation %s",
installation_id,
exc_info=True,
)

def get_jwt_token(self, private_key, app_id):
payload = {
Expand Down
54 changes: 54 additions & 0 deletions tests/test_github_app.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
from __future__ import annotations

from unittest import mock

import pytest
import requests

from src.github_app import GithubAppToken


def _token_manager() -> GithubAppToken:
manager = GithubAppToken.__new__(GithubAppToken)
manager.headers = {"Authorization": "Bearer jwt"}
return manager


def test_get_token_ignores_revoke_errors():
token_manager = _token_manager()
token_response = mock.Mock()
token_response.raise_for_status.return_value = None
token_response.json.return_value = {"token": "test-token"}

with (
mock.patch("src.github_app.requests.post", return_value=token_response),
mock.patch(
"src.github_app.requests.delete",
side_effect=requests.ConnectionError("remote closed"),
) as delete_mock,
mock.patch("src.github_app.logger.warning") as warning_mock,
):
with token_manager.get_token(installation_id=1234) as token:
assert token == "test-token"

delete_mock.assert_called_once_with(
"https://api.github.com/installation/token",
headers={"Authorization": "token test-token"},
)
warning_mock.assert_called_once()


def test_get_token_raises_when_access_token_request_fails():
token_manager = _token_manager()
token_response = mock.Mock()
token_response.raise_for_status.side_effect = requests.HTTPError("bad gateway")

with (
mock.patch("src.github_app.requests.post", return_value=token_response),
mock.patch("src.github_app.requests.delete") as delete_mock,
pytest.raises(requests.HTTPError),
):
with token_manager.get_token(installation_id=1234):
pass

delete_mock.assert_not_called()
Loading