66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from src.application.auth.errors import InvalidApiKeyError, TenantInactiveError
|
|
from src.application.auth.service import resolve_auth_context
|
|
from tests.support.factories import create_api_key, create_tenant
|
|
|
|
pytestmark = [
|
|
pytest.mark.integration,
|
|
pytest.mark.postgres,
|
|
pytest.mark.asyncio(loop_scope="session"),
|
|
]
|
|
|
|
|
|
async def test_resolve_auth_context_accepts_valid_key(
|
|
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
|
) -> None:
|
|
tenant = await create_tenant(db_session)
|
|
_, full_key = await create_api_key(db_session, tenant=tenant, scopes=["files:write"])
|
|
await db_session.commit()
|
|
|
|
auth = await resolve_auth_context(db_sessionmaker, full_key)
|
|
|
|
assert auth.tenant_id == tenant.id
|
|
assert auth.has_scope("files:write")
|
|
|
|
|
|
async def test_resolve_auth_context_rejects_wrong_secret(
|
|
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
|
) -> None:
|
|
tenant = await create_tenant(db_session)
|
|
api_key, _ = await create_api_key(db_session, tenant=tenant)
|
|
await db_session.commit()
|
|
|
|
with pytest.raises(InvalidApiKeyError):
|
|
await resolve_auth_context(db_sessionmaker, f"sk_{api_key.key_prefix}_wrong-secret")
|
|
|
|
|
|
async def test_resolve_auth_context_rejects_unknown_prefix(
|
|
db_sessionmaker: async_sessionmaker[AsyncSession],
|
|
) -> None:
|
|
with pytest.raises(InvalidApiKeyError):
|
|
await resolve_auth_context(db_sessionmaker, "sk_doesnotexist_secret")
|
|
|
|
|
|
async def test_resolve_auth_context_rejects_revoked_key(
|
|
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
|
) -> None:
|
|
tenant = await create_tenant(db_session)
|
|
_, full_key = await create_api_key(db_session, tenant=tenant, status="revoked")
|
|
await db_session.commit()
|
|
|
|
with pytest.raises(InvalidApiKeyError):
|
|
await resolve_auth_context(db_sessionmaker, full_key)
|
|
|
|
|
|
async def test_resolve_auth_context_rejects_suspended_tenant(
|
|
db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
|
|
) -> None:
|
|
tenant = await create_tenant(db_session, status="suspended")
|
|
_, full_key = await create_api_key(db_session, tenant=tenant)
|
|
await db_session.commit()
|
|
|
|
with pytest.raises(TenantInactiveError):
|
|
await resolve_auth_context(db_sessionmaker, full_key)
|