54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
"""Authentication tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from tests.conftest import AUTH_HEADERS
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_valid_token_passes(async_client: AsyncClient):
|
|
"""Requests with correct Bearer token are not rejected at the auth layer.
|
|
|
|
We don't need a real DB for this — a 422 from missing body means auth passed.
|
|
"""
|
|
# POST /logs without a body — if auth passes we get 422 (validation), not 401
|
|
response = await async_client.post("/logs", json={}, headers=AUTH_HEADERS)
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_missing_header(async_client: AsyncClient):
|
|
"""Missing Authorization header returns 403 (no credentials) or 401."""
|
|
response = await async_client.post("/logs", json={})
|
|
assert response.status_code in {401, 403}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_wrong_token(async_client: AsyncClient):
|
|
"""Wrong Bearer token returns 401."""
|
|
response = await async_client.post(
|
|
"/logs",
|
|
json={"agent_id": "x", "message": "x"},
|
|
headers={"Authorization": "Bearer totally-wrong"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_not_required_for_health(async_client: AsyncClient):
|
|
"""GET /health does not require authentication."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
mock_conn = AsyncMock()
|
|
mock_conn.fetchval.return_value = 1
|
|
mock_pool = MagicMock()
|
|
mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with patch("mals.main.get_pool", AsyncMock(return_value=mock_pool)):
|
|
response = await async_client.get("/health")
|
|
assert response.status_code == 200
|