feat(FN-004): merge fusion/fn-004
This commit is contained in:
@@ -0,0 +1,66 @@
|
||||
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
|
||||
Reference in New Issue
Block a user