aurweb/test/test_asgi.py

import http
import os
import re

from typing import Callable
from unittest import mock

import fastapi
import pytest

from fastapi import HTTPException
from fastapi.testclient import TestClient

import aurweb.asgi
import aurweb.config
import aurweb.redis

from aurweb.exceptions import handle_form_exceptions
from aurweb.testing.requests import Request


@pytest.fixture
def setup(db_test, email_test):
    aurweb.redis.redis_connection().flushall()
    yield
    aurweb.redis.redis_connection().flushall()


@pytest.fixture
def mock_glab_request(monkeypatch):
    def wrapped(return_value=None, side_effect=None):
        def what_to_return(*args, **kwargs):
            if side_effect:
                return side_effect  # pragma: no cover
            return return_value
        monkeypatch.setattr("requests.post", what_to_return)
    return wrapped


def mock_glab_config(project: str = "test/project", token: str = "test-token"):
    config_get = aurweb.config.get

    def wrapper(section: str, key: str) -> str:
        if section == "notifications":
            if key == "error-project":
                return project
            elif key == "error-token":
                return token
        return config_get(section, key)
    return wrapper


@pytest.mark.asyncio
async def test_asgi_startup_session_secret_exception(monkeypatch):
    """ Test that we get an IOError on app_startup when we cannot
    connect to options.redis_address. """

    redis_addr = aurweb.config.get("options", "redis_address")

    def mock_get(section: str, key: str):
        if section == "fastapi" and key == "session_secret":
            return None
        return redis_addr

    with mock.patch("aurweb.config.get", side_effect=mock_get):
        with pytest.raises(Exception):
            await aurweb.asgi.app_startup()


@pytest.mark.asyncio
async def test_asgi_startup_exception(monkeypatch):
    with mock.patch.dict(os.environ, {"AUR_CONFIG": "conf/config.defaults"}):
        aurweb.config.rehash()
        with pytest.raises(Exception):
            await aurweb.asgi.app_startup()
    aurweb.config.rehash()


@pytest.mark.asyncio
async def test_asgi_http_exception_handler():
    exc = HTTPException(status_code=422, detail="EXCEPTION!")
    phrase = http.HTTPStatus(exc.status_code).phrase
    response = await aurweb.asgi.http_exception_handler(Request(), exc)
    assert response.status_code == 422
    content = response.body.decode()
    assert f"{exc.status_code} - {phrase}" in content
    assert "EXCEPTION!" in content


@pytest.mark.asyncio
async def test_asgi_app_unsupported_backends():
    config_get = aurweb.config.get

    # Test that the previously supported "sqlite" backend is now
    # unsupported by FastAPI.
    def mock_sqlite_backend(section: str, key: str):
        if section == "database" and key == "backend":
            return "sqlite"
        return config_get(section, key)

    with mock.patch("aurweb.config.get", side_effect=mock_sqlite_backend):
        expr = r"^.*\(sqlite\) is unsupported.*$"
        with pytest.raises(ValueError, match=expr):
            await aurweb.asgi.app_startup()


@pytest.fixture
def use_traceback():
    config_getboolean = aurweb.config.getboolean

    def mock_getboolean(section: str, key: str) -> bool:
        if section == "options" and key == "traceback":
            return True
        return config_getboolean(section, key)

    with mock.patch("aurweb.config.getboolean", side_effect=mock_getboolean):
        yield


class FakeResponse:
    def __init__(self, status_code: int = 201, text: str = "{}"):
        self.status_code = status_code
        self.text = text


def test_internal_server_error_bad_glab(setup: None, use_traceback: None,
                                        mock_glab_request: Callable,
                                        caplog: pytest.LogCaptureFixture):
    @aurweb.asgi.app.get("/internal_server_error")
    async def internal_server_error(request: fastapi.Request):
        raise ValueError("test exception")

    with mock.patch("aurweb.config.get", side_effect=mock_glab_config()):
        with TestClient(app=aurweb.asgi.app) as request:
            mock_glab_request(FakeResponse(status_code=404))
            resp = request.get("/internal_server_error")
    assert resp.status_code == int(http.HTTPStatus.INTERNAL_SERVER_ERROR)

    expr = r"ERROR.*Unable to report exception to"
    assert re.search(expr, caplog.text)

    expr = r"FATAL\[.{7}\]"
    assert re.search(expr, caplog.text)


def test_internal_server_error_no_token(setup: None, use_traceback: None,
                                        mock_glab_request: Callable,
                                        caplog: pytest.LogCaptureFixture):
    @aurweb.asgi.app.get("/internal_server_error")
    async def internal_server_error(request: fastapi.Request):
        raise ValueError("test exception")

    mock_get = mock_glab_config(token="set-me")
    with mock.patch("aurweb.config.get", side_effect=mock_get):
        with TestClient(app=aurweb.asgi.app) as request:
            mock_glab_request(FakeResponse())
            resp = request.get("/internal_server_error")
    assert resp.status_code == int(http.HTTPStatus.INTERNAL_SERVER_ERROR)

    expr = r"WARNING.*Unable to report an exception found"
    assert re.search(expr, caplog.text)

    expr = r"FATAL\[.{7}\]"
    assert re.search(expr, caplog.text)


def test_internal_server_error(setup: None, use_traceback: None,
                               mock_glab_request: Callable,
                               caplog: pytest.LogCaptureFixture):
    @aurweb.asgi.app.get("/internal_server_error")
    async def internal_server_error(request: fastapi.Request):
        raise ValueError("test exception")

    with mock.patch("aurweb.config.get", side_effect=mock_glab_config()):
        with TestClient(app=aurweb.asgi.app) as request:
            mock_glab_request(FakeResponse())
            # Test with a ?query=string to cover the request.url.query path.
            resp = request.get("/internal_server_error?query=string")
    assert resp.status_code == int(http.HTTPStatus.INTERNAL_SERVER_ERROR)

    # Assert that the exception got logged with with its traceback id.
    expr = r"FATAL\[.{7}\]"
    assert re.search(expr, caplog.text)

    # Let's do it again to exercise the cached path.
    caplog.clear()
    with mock.patch("aurweb.config.get", side_effect=mock_glab_config()):
        with TestClient(app=aurweb.asgi.app) as request:
            mock_glab_request(FakeResponse())
            resp = request.get("/internal_server_error")
    assert resp.status_code == int(http.HTTPStatus.INTERNAL_SERVER_ERROR)
    assert "FATAL" not in caplog.text


def test_internal_server_error_post(setup: None, use_traceback: None,
                                    mock_glab_request: Callable,
                                    caplog: pytest.LogCaptureFixture):
    @aurweb.asgi.app.post("/internal_server_error")
    @handle_form_exceptions
    async def internal_server_error(request: fastapi.Request):
        raise ValueError("test exception")

    data = {"some": "data"}
    with mock.patch("aurweb.config.get", side_effect=mock_glab_config()):
        with TestClient(app=aurweb.asgi.app) as request:
            mock_glab_request(FakeResponse())
            # Test with a ?query=string to cover the request.url.query path.
            resp = request.post("/internal_server_error", data=data)
    assert resp.status_code == int(http.HTTPStatus.INTERNAL_SERVER_ERROR)

    expr = r"FATAL\[.{7}\]"
    assert re.search(expr, caplog.text)