Fusion
67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
from typing import Any
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from app.config import settings
|
|
from app.models.user import User
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dev_bypass_creates_user(auth_client: AsyncClient) -> None:
|
|
response = await auth_client.get("/api/v1/users/me")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["authentik_sub"] == "dev-user"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_token_raises_401_when_bypass_disabled(client: AsyncClient) -> None:
|
|
original_debug = settings.debug
|
|
original_bypass = settings.auth_dev_bypass
|
|
settings.debug = False
|
|
settings.auth_dev_bypass = False
|
|
|
|
response = await client.get("/api/v1/users/me")
|
|
|
|
settings.debug = original_debug
|
|
settings.auth_dev_bypass = original_bypass
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inactive_user_raises_403(
|
|
auth_client: AsyncClient, db_session: async_sessionmaker[Any]
|
|
) -> None:
|
|
async with db_session() as session:
|
|
result = await session.execute(select(User).where(User.authentik_sub == "dev-user"))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
user = User(
|
|
authentik_sub="dev-user",
|
|
email="dev@localhost",
|
|
display_name="Dev User",
|
|
is_active=True,
|
|
)
|
|
session.add(user)
|
|
await session.commit()
|
|
|
|
async with db_session() as session:
|
|
result = await session.execute(select(User).where(User.authentik_sub == "dev-user"))
|
|
user = result.scalar_one()
|
|
user.is_active = False
|
|
await session.commit()
|
|
|
|
response = await auth_client.get("/api/v1/users/me")
|
|
|
|
async with db_session() as session:
|
|
result = await session.execute(select(User).where(User.authentik_sub == "dev-user"))
|
|
user = result.scalar_one()
|
|
user.is_active = True
|
|
await session.commit()
|
|
|
|
assert response.status_code == 403
|