39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
|
|
import pytest
|
|
from fastapi import HTTPException, Request
|
|
|
|
from crypto_spot_bot.auth import ApiAuthorizer
|
|
|
|
|
|
def _request(**headers: str) -> Request:
|
|
encoded = [(key.lower().encode(), value.encode()) for key, value in headers.items()]
|
|
return Request({"type": "http", "method": "GET", "path": "/", "headers": encoded})
|
|
|
|
|
|
def test_api_authorizer_accepts_direct_basic_token(make_settings, tmp_path) -> None:
|
|
settings = make_settings(tmp_path, api_auth_token="user:secret")
|
|
auth = ApiAuthorizer(settings)
|
|
basic = base64.b64encode(b"user:secret").decode("ascii")
|
|
|
|
asyncio.run(auth.require(_request(Authorization=f"Basic {basic}")))
|
|
|
|
|
|
def test_api_authorizer_accepts_trusted_proxy_header(make_settings, tmp_path) -> None:
|
|
settings = make_settings(tmp_path, trusted_proxy_user_header="X-TradeBot-Proxy-User")
|
|
auth = ApiAuthorizer(settings)
|
|
|
|
asyncio.run(auth.require(_request(**{"X-TradeBot-Proxy-User": "sevenhill"})))
|
|
|
|
|
|
def test_api_authorizer_rejects_missing_credentials(make_settings, tmp_path) -> None:
|
|
auth = ApiAuthorizer(make_settings(tmp_path))
|
|
|
|
with pytest.raises(HTTPException) as raised:
|
|
asyncio.run(auth.require(_request()))
|
|
|
|
assert raised.value.status_code == 401
|