path: root/pydis_site/apps/api/tests/test_github_utils.py

import dataclasses
import datetime
import typing
import unittest
from unittest import mock

import django.test
import httpx
import jwt
import rest_framework.response
import rest_framework.test
from django.urls import reverse

from pydis_site import settings
from pydis_site.apps.api import github_utils


class GeneralUtilityTests(unittest.TestCase):
    """Test the utility methods which do not fit in another class."""

    def test_token_generation(self):
        """Test that the a valid JWT token is generated."""
        def encode(payload: dict, _: str, algorithm: str, *args, **kwargs) -> str:
            """
            Intercept the encode method.

            The result is encoded with an algorithm which does not require a PEM key, as it may
            not be available in testing environments.
            """
            self.assertEqual("RS256", algorithm, "The GitHub App JWT must be signed using RS256.")
            return original_encode(
                payload, "secret-encoding-key", *args, algorithm="HS256", **kwargs
            )

        original_encode = jwt.encode
        with mock.patch("jwt.encode", new=encode):
            token = github_utils.generate_token()
        decoded = jwt.decode(token, "secret-encoding-key", algorithms=["HS256"])

        delta = datetime.timedelta(minutes=10)
        self.assertAlmostEqual(decoded["exp"] - decoded["iat"], delta.total_seconds())
        then = datetime.datetime.now(tz=datetime.UTC) + delta
        self.assertLess(decoded["exp"], then.timestamp())


class CheckRunTests(unittest.TestCase):
    """Tests the check_run_status utility."""

    run_kwargs: typing.Mapping = {
        "name": "run_name",
        "head_sha": "sha",
        "status": "completed",
        "conclusion": "success",
        "created_at": datetime.datetime.now(tz=datetime.UTC).strftime(settings.GITHUB_TIMESTAMP_FORMAT),
        "artifacts_url": "url",
    }

    def test_completed_run(self):
        """Test that an already completed run returns the correct URL."""
        final_url = "some_url_string_1234"

        kwargs = dict(self.run_kwargs, artifacts_url=final_url)
        result = github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))
        self.assertEqual(final_url, result)

    def test_pending_run(self):
        """Test that a pending run raises the proper exception."""
        kwargs = dict(self.run_kwargs, status="pending")
        with self.assertRaises(github_utils.RunPendingError):
            github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))

    def test_timeout_error(self):
        """Test that a timeout is declared after a certain duration."""
        kwargs = dict(self.run_kwargs, status="pending")
        # Set the creation time to well before the MAX_RUN_TIME
        # to guarantee the right conclusion
        kwargs["created_at"] = (
            datetime.datetime.now(tz=datetime.UTC)
            - github_utils.MAX_RUN_TIME - datetime.timedelta(minutes=10)
        ).strftime(settings.GITHUB_TIMESTAMP_FORMAT)

        with self.assertRaises(github_utils.RunTimeoutError):
            github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))

    def test_failed_run(self):
        """Test that a failed run raises the proper exception."""
        kwargs = dict(self.run_kwargs, conclusion="failed")
        with self.assertRaises(github_utils.ActionFailedError):
            github_utils.check_run_status(github_utils.WorkflowRun(**kwargs))


def get_response_authorize(_: httpx.Client, request: httpx.Request, **__) -> httpx.Response:
    """
    Helper method for the authorize tests.

    Requests are intercepted before being sent out, and the appropriate responses are returned.
    """
    path = request.url.path
    auth = request.headers.get("Authorization")

    if request.method == "GET":
        if path == "/app/installations":
            if auth == "bearer JWT initial token":
                return httpx.Response(200, request=request, json=[{
                    "account": {"login": "VALID_OWNER"},
                    "access_tokens_url": "https://example.com/ACCESS_TOKEN_URL"
                }])
            return httpx.Response(
                401, json={"error": "auth app/installations"}, request=request
            )

        elif path == "/installation/repositories":  # noqa: RET505
            if auth == "bearer app access token":
                return httpx.Response(200, request=request, json={
                    "repositories": [{
                        "name": "VALID_REPO"
                    }]
                })
        return httpx.Response(  # pragma: no cover
            401, json={"error": "auth installation/repositories"}, request=request
        )

    elif request.method == "POST":  # noqa: RET505
        if path == "/ACCESS_TOKEN_URL":
            if auth == "bearer JWT initial token":
                return httpx.Response(200, request=request, json={"token": "app access token"})
            return httpx.Response(401, json={"error": "auth access_token"}, request=request)  # pragma: no cover

    # Reaching this point means something has gone wrong
    return httpx.Response(500, request=request)  # pragma: no cover


@mock.patch("httpx.Client.send", new=get_response_authorize)
@mock.patch.object(github_utils, "generate_token", new=mock.Mock(return_value="JWT initial token"))
class AuthorizeTests(unittest.TestCase):
    """Test the authorize utility."""

    def test_invalid_apps_auth(self):
        """Test that an exception is raised if authorization was attempted with an invalid token."""
        with mock.patch.object(github_utils, "generate_token", return_value="Invalid token"):  # noqa: SIM117
            with self.assertRaises(httpx.HTTPStatusError) as error:
                github_utils.authorize("VALID_OWNER", "VALID_REPO")

        exception: httpx.HTTPStatusError = error.exception
        self.assertEqual(401, exception.response.status_code)
        self.assertEqual("auth app/installations", exception.response.json()["error"])

    def test_missing_repo(self):
        """Test that an exception is raised when the selected owner or repo are not available."""
        with self.assertRaises(github_utils.NotFoundError):
            github_utils.authorize("INVALID_OWNER", "VALID_REPO")
        with self.assertRaises(github_utils.NotFoundError):
            github_utils.authorize("VALID_OWNER", "INVALID_REPO")

    def test_valid_authorization(self):
        """Test that an accessible repository can be accessed."""
        client = github_utils.authorize("VALID_OWNER", "VALID_REPO")
        self.assertEqual("bearer app access token", client.headers.get("Authorization"))


class ArtifactFetcherTests(unittest.TestCase):
    """Test the get_artifact utility."""

    @staticmethod
    def get_response_get_artifact(request: httpx.Request, **_) -> httpx.Response:
        """
        Helper method for the get_artifact tests.

        Requests are intercepted before being sent out, and the appropriate responses are returned.
        """
        path = request.url.path

        if "force_error" in path:
            return httpx.Response(404, request=request)

        if request.method == "GET":
            if path == "/repos/owner/repo/actions/runs":
                run = github_utils.WorkflowRun(
                    name="action_name",
                    head_sha="action_sha",
                    created_at=(
                        datetime.datetime
                        .now(tz=datetime.UTC)
                        .strftime(settings.GITHUB_TIMESTAMP_FORMAT)
                    ),
                    status="completed",
                    conclusion="success",
                    artifacts_url="artifacts_url"
                )
                return httpx.Response(
                    200, request=request, json={"workflow_runs": [dataclasses.asdict(run)]}
                )
            elif path == "/artifact_url":  # noqa: RET505
                return httpx.Response(
                    200, request=request, json={"artifacts": [{
                        "name": "artifact_name",
                        "archive_download_url": "artifact_download_url"
                    }]}
                )
            elif path == "/artifact_download_url":
                response = httpx.Response(302, request=request)
                response.next_request = httpx.Request(
                    "GET",
                    httpx.URL("https://final_download.url")
                )
                return response

        # Reaching this point means something has gone wrong
        return httpx.Response(500, request=request)  # pragma: no cover

    def setUp(self) -> None:
        self.call_args = ["owner", "repo", "action_sha", "action_name", "artifact_name"]
        self.client = httpx.Client(base_url="https://example.com", timeout=5)

        self.patchers = [
            mock.patch.object(self.client, "send", new=self.get_response_get_artifact),
            mock.patch.object(github_utils, "authorize", return_value=self.client),
            mock.patch.object(github_utils, "check_run_status", return_value="artifact_url"),
        ]

        for patcher in self.patchers:
            patcher.start()

    def tearDown(self) -> None:
        for patcher in self.patchers:
            patcher.stop()

    def test_client_closed_on_errors(self):
        """Test that the client is terminated even if an error occurs at some point."""
        self.call_args[0] = "force_error"
        with self.assertRaises(httpx.HTTPStatusError):
            github_utils.get_artifact(*self.call_args)
        self.assertTrue(self.client.is_closed)

    def test_missing(self):
        """Test that an exception is raised if the requested artifact was not found."""
        cases = (
            "invalid sha",
            "invalid action name",
            "invalid artifact name",
        )
        for i, name in enumerate(cases, 2):
            with self.subTest(f"Test {name} raises an error"):
                new_args = self.call_args.copy()
                new_args[i] = name

                with self.assertRaises(github_utils.NotFoundError):
                    github_utils.get_artifact(*new_args)

    def test_valid(self):
        """Test that the correct download URL is returned for valid requests."""
        url = github_utils.get_artifact(*self.call_args)
        self.assertEqual("https://final_download.url", url)
        self.assertTrue(self.client.is_closed)


@mock.patch.object(github_utils, "get_artifact")
class GitHubArtifactViewTests(django.test.TestCase):
    """Test the GitHub artifact fetch API view."""

    def setUp(self):
        self.kwargs = {
            "owner": "test_owner",
            "repo": "test_repo",
            "sha": "test_sha",
            "action_name": "test_action",
            "artifact_name": "test_artifact",
        }
        self.url = reverse("api:github-artifacts", kwargs=self.kwargs)

    def test_correct_artifact(self, artifact_mock: mock.Mock):
        """Test a proper response is returned with proper input."""
        artifact_mock.return_value = "final download url"
        result = self.client.get(self.url)

        self.assertIsInstance(result, rest_framework.response.Response)
        self.assertEqual({"url": artifact_mock.return_value}, result.data)

    def test_failed_fetch(self, artifact_mock: mock.Mock):
        """Test that a proper error is returned when the request fails."""
        artifact_mock.side_effect = github_utils.NotFoundError("Test error message")
        result = self.client.get(self.url)

        self.assertIsInstance(result, rest_framework.response.Response)
        self.assertEqual({
            "error_type": github_utils.NotFoundError.__name__,
            "error": "Test error message",
            "requested_resource": "/".join(self.kwargs.values())
        }, result.data)