| from datetime import datetime, timedelta |
| |
| import jwt |
| import pytest |
| from starlette.testclient import TestClient |
| |
| |
| @pytest.mark.parametrize( |
| "test_server_path, test_client_path, method, status_code, user, response_body", [ |
| ('/dataset1/resource2', '/dataset1/resource2', 'GET', 200, 'alice', 'ok'), |
| ('/dataset1/resource2', '/dataset1/resource2', 'GET', 403, 'notalice', 'Forbidden'), |
| ('/dataset1/resource2', '/dataset1/resource2', 'OPTIONS', 200, 'notalice', 'ok'), |
| ('/dataset1/resource1', '/dataset1/resource1', 'POST', 200, 'alice', 'ok'), |
| ] |
| ) |
| def test_jwt_middleware_authed(jwt_app_fixture, test_server_path, test_client_path, method, status_code, user, |
| response_body): |
| @getattr(jwt_app_fixture, method.lower())(test_server_path) |
| async def index(): |
| return 'ok' |
| |
| JWT_SECRET_KEY = "secret" |
| test_client = TestClient(jwt_app_fixture) |
| expire = datetime.utcnow() + timedelta( |
| minutes=60 |
| ) |
| token = jwt.encode({"exp": expire, "username": user}, JWT_SECRET_KEY, algorithm="HS256") |
| |
| test_response = getattr(test_client, method.lower())(test_client_path, headers={'Authorization': 'Bearer ' + token}) |
| |
| assert test_response.status_code == status_code |
| assert test_response.json() == response_body |
| |
| |
| @pytest.mark.parametrize( |
| "test_server_path, test_client_path, method, status_code, response_body", [ |
| ('/login', '/login', 'GET', 200, 'ok'), |
| ('/', '/', 'GET', 200, 'ok') |
| ] |
| ) |
| def test_jwt_middleware_not_authed(jwt_app_fixture, test_server_path, test_client_path, method, status_code, |
| response_body): |
| @getattr(jwt_app_fixture, method.lower())(test_server_path) |
| async def index(): |
| return 'ok' |
| |
| test_client = TestClient(jwt_app_fixture) |
| |
| test_response = getattr(test_client, method.lower())(test_client_path) |
| |
| assert test_response.status_code == status_code |
| assert test_response.json() == response_body |
| |
| |
| if __name__ == '__main__': |
| pytest.main() |