blob: 63e311700ad4d548b9ab2c5724faba239c2c8a11 [file] [log] [blame]
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()