|
1 | 1 | import uuid |
2 | 2 | from collections.abc import Iterator |
3 | 3 | from dataclasses import dataclass |
| 4 | +from typing import Any |
4 | 5 | from unittest.mock import MagicMock, Mock, patch |
5 | 6 |
|
6 | 7 | import jwt |
7 | 8 | import pytest |
8 | 9 | from bluesky.protocols import Stoppable |
9 | 10 | from fastapi import status |
10 | 11 | from fastapi.testclient import TestClient |
| 12 | +from httpx import Headers |
11 | 13 | from pydantic import BaseModel, ValidationError |
12 | 14 | from pydantic_core import InitErrorDetails |
13 | 15 | from super_state_machine.errors import TransitionError |
@@ -62,11 +64,15 @@ def client(mock_runner: Mock) -> Iterator[TestClient]: |
62 | 64 |
|
63 | 65 | @pytest.fixture |
64 | 66 | def client_with_auth( |
65 | | - mock_runner: Mock, oidc_config: OIDCConfig |
| 67 | + mock_runner: Mock, oidc_config: OIDCConfig, valid_token_with_jwt: dict[str, Any] |
66 | 68 | ) -> Iterator[TestClient]: |
67 | 69 | with patch("blueapi.service.interface.worker"): |
68 | 70 | main.setup_runner(runner=mock_runner) |
69 | | - yield TestClient(main.get_app(ApplicationConfig(oidc=oidc_config))) |
| 71 | + access_token = valid_token_with_jwt.get("access_token") |
| 72 | + assert access_token is not None |
| 73 | + client = TestClient(main.get_app(ApplicationConfig(oidc=oidc_config))) |
| 74 | + client.headers = Headers(headers={"Authorization": f"Bearer {access_token}"}) |
| 75 | + yield client |
70 | 76 | main.teardown_runner() |
71 | 77 |
|
72 | 78 |
|
@@ -720,3 +726,39 @@ def test_health_probe(client: TestClient): |
720 | 726 |
|
721 | 727 | assert response.status_code == status.HTTP_200_OK |
722 | 728 | assert response.json() == {"status": "ok"} |
| 729 | + |
| 730 | + |
| 731 | +def test_logout( |
| 732 | + mock_runner: Mock, |
| 733 | + mock_authn_server, |
| 734 | + oidc_config: OIDCConfig, |
| 735 | + client_with_auth: TestClient, |
| 736 | +): |
| 737 | + oidc_config.logout_redirect_endpoint = "/oauth2/logout" |
| 738 | + mock_runner.run.return_value = oidc_config |
| 739 | + client_with_auth.follow_redirects = False |
| 740 | + response = client_with_auth.get("/logout") |
| 741 | + assert response.status_code == status.HTTP_308_PERMANENT_REDIRECT |
| 742 | + assert ( |
| 743 | + response.headers.get("X-Auth-Request-Redirect") |
| 744 | + == oidc_config.end_session_endpoint |
| 745 | + ) |
| 746 | + assert response.headers.get("location") == oidc_config.logout_redirect_endpoint |
| 747 | + |
| 748 | + |
| 749 | +@pytest.mark.parametrize("has_oidc_config", [True, False]) |
| 750 | +def test_logout_when_oidc_config_invalid( |
| 751 | + has_oidc_config: bool, |
| 752 | + mock_runner: Mock, |
| 753 | + oidc_config: OIDCConfig, |
| 754 | + mock_authn_server, |
| 755 | + client_with_auth: TestClient, |
| 756 | +): |
| 757 | + if has_oidc_config: |
| 758 | + oidc_config.logout_redirect_endpoint = "" |
| 759 | + mock_runner.run.return_value = oidc_config |
| 760 | + else: |
| 761 | + mock_runner.run.return_value = None |
| 762 | + |
| 763 | + response = client_with_auth.get("/logout") |
| 764 | + assert response.status_code == status.HTTP_205_RESET_CONTENT |
0 commit comments