rs_server_common/utils/pytest/pytest_authentication_utils.md

<< Back to index

Utility authentication functions used by the pytest unit tests.

init_app_cluster_mode()

Init the FastAPI application with all the cluster mode features (local mode=0)

Source code in docs/rs-server/services/common/rs_server_common/utils/pytest/pytest_authentication_utils.py

def init_app_cluster_mode():
    """Init the FastAPI application with all the cluster mode features (local mode=0)"""

    os.environ["RSPY_LOCAL_MODE"] = "0"
    os.environ["RSPY_LOCAL_CATALOG_MODE"] = "1"
    os.environ["RSPY_UAC_HOMEPAGE"] = RSPY_UAC_HOMEPAGE
    os.environ["RSPY_UAC_CHECK_URL"] = RSPY_UAC_CHECK_URL
    os.environ["OIDC_ENDPOINT"] = OIDC_ENDPOINT
    os.environ["OIDC_REALM"] = OIDC_REALM
    os.environ["OIDC_CLIENT_ID"] = "OIDC_CLIENT_ID"
    os.environ["OIDC_CLIENT_SECRET"] = "OIDC_CLIENT_SECRET"  # nosec
    os.environ["RSPY_COOKIE_SECRET"] = "RSPY_COOKIE_SECRET"  # nosec

init_authentication_test(mocker, httpx_mock, client, test_apikey, test_oauth2, iam_roles, user_attributes, mock_wrong_apikey=False, user_login='pyteam') async

Init mockers for tests in cluster mode with authentication.

Source code in docs/rs-server/services/common/rs_server_common/utils/pytest/pytest_authentication_utils.py

async def init_authentication_test(
    mocker,
    httpx_mock: HTTPXMock,
    client: TestClient,
    test_apikey: bool,
    test_oauth2: bool,
    iam_roles: list[str],
    user_attributes: dict[str, Any],
    mock_wrong_apikey: bool = False,
    user_login="pyteam",
):
    """Init mockers for tests in cluster mode with authentication."""

    # Needs init_app_cluster_mode()
    from rs_server_common.utils.pytest.pytest_utils import (  # pylint: disable=import-outside-toplevel
        mock_oauth2,
    )

    # Mock cluster mode to enable authentication. See: https://stackoverflow.com/a/69685866
    mocker.patch("rs_server_common.settings.LOCAL_MODE", new=False, autospec=False)
    mocker.patch("rs_server_common.settings.CLUSTER_MODE", new=True, autospec=False)

    # Clear oauth2 cookies
    client.cookies.clear()

    if test_apikey:
        # With a valid api key in headers, the uac manager will give access to the endpoint
        ttl_cache.clear()  # clear the cached response
        httpx_mock.add_response(
            url=RSPY_UAC_CHECK_URL,
            match_headers={APIKEY_HEADER: VALID_APIKEY},
            status_code=HTTP_200_OK,
            json={
                "name": "test_apikey",
                "user_login": user_login,
                "is_active": True,
                "never_expire": True,
                "expiration_date": "2024-04-10T13:57:28.475052",
                "total_queries": 0,
                "latest_sync_date": "2024-03-26T13:57:28.475058",
                "iam_roles": iam_roles,
                "config": user_attributes,
                "allowed_referers": ["toto"],
            },
        )

        # With a wrong api key, it returns 403
        if mock_wrong_apikey:
            httpx_mock.add_response(
                url=RSPY_UAC_CHECK_URL,
                match_headers={APIKEY_HEADER: WRONG_APIKEY},
                status_code=HTTP_403_FORBIDDEN,
            )

    # If we test the oauth2 authentication, we login the user.
    # His authentication information is saved in the client session cookies.
    # Note: we use the "login from console" because we need the client to follow redirections,
    # and they are disabled in these tests.
    if test_oauth2:
        await mock_oauth2(
            mocker,
            client,
            "/auth/login_from_console",
            "oauth2_user_id",
            user_login,
            iam_roles,
            user_attributes,
        )

    # Mock the OAuth2 server responses that are used for the STAC extensions (not for the authentication)
    mocker.patch.object(
        StarletteOAuth2App,
        "load_server_metadata",
        return_value={"authorization_endpoint": OAUTH2_AUTHORIZATION_ENDPOINT, "token_endpoint": OAUTH2_TOKEN_ENDPOINT},
    )