Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .env-example
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ CACHE__DB=0
CACHE__PASSWORD=
CACHE__ENCODING=utf-8

AUTH__JWT_SECRET=your-secret-key-here
AUTH__JWT_SECRET=your-super-super-long-secret-key-here
AUTH__JWT_ALGORITHM=HS256
AUTH__JWT_LIFETIME_SECONDS=1800
AUTH__REFRESH_TOKEN_LIFETIME_SECONDS=2592000
Expand Down
30 changes: 16 additions & 14 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -3,31 +3,31 @@ name = "fastapi-boilerplate"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"aiosqlite>=0.21.0",
"alembic>=1.17.2",
"aiosqlite>=0.22.1",
"alembic>=1.18.4",
"asyncpg>=0.31.0",
"fastapi>=0.115.0",
"fastapi-pagination[sqlalchemy]>=0.15.3",
"fastapi-users[sqlalchemy]>=15.0.1",
"fastapi>=0.129.0",
"fastapi-pagination[sqlalchemy]>=0.15.10",
"fastapi-users[sqlalchemy]>=15.0.4",
"httpx-oauth>=0.16.1",
"loguru>=0.7.3",
"pydantic-settings>=2.12.0",
"tenacity>=9.0.0",
"redis[hiredis]>=7.1.0",
"sqlmodel>=0.0.27",
"uvicorn[standard]>=0.32.0",
"pydantic-settings>=2.13.1",
"tenacity>=9.1.4",
"redis[hiredis]>=7.2.0",
"sqlmodel>=0.0.34",
"uvicorn[standard]>=0.41.0",
]

[dependency-groups]
dev = [
"fakeredis>=2.32.1",
"locust>=2.32.5",
"pytest>=8.3.3",
"fakeredis>=2.34.0",
"locust>=2.43.3",
"pytest>=9.0.2",
"pytest-asyncio>=1.2.0",
"pytest-cov>=6.1.1",
"pytest-timeout>=2.4.0",
"python-dotenv>=1.2.1",
"ruff>=0.12.8",
"ruff>=0.15.1",
]

[build-system]
Expand All @@ -42,6 +42,8 @@ packages = ["src"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
asyncio_default_test_loop_scope = "session"
# filterwarnings = ["error::DeprecationWarning"]
timeout = 5
pythonpath = ["."]
Expand Down
46 changes: 29 additions & 17 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,13 @@

import dotenv
import pytest
import pytest_asyncio
from fakeredis import FakeAsyncRedis
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from loguru import logger
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel

Expand Down Expand Up @@ -74,23 +75,36 @@ async def async_client(full_app: FastAPI) -> AsyncGenerator[AsyncClient, None]:
yield client


@pytest.fixture
async def test_db():
@pytest_asyncio.fixture(scope="session")
async def test_engine(tmp_path_factory) -> AsyncGenerator[AsyncEngine, None]:
db_file = tmp_path_factory.mktemp("db") / "test.db"
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
f"sqlite+aiosqlite:///{db_file}",
echo=False,
connect_args={"check_same_thread": False},
)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async with engine.connect() as conn:
await conn.execute(text("PRAGMA foreign_keys=ON"))
await conn.commit()
@event.listens_for(engine.sync_engine, "connect")
def _set_sqlite_pragma(dbapi_conn, _):
cursor = dbapi_conn.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()

yield engine
await engine.dispose()


@pytest.fixture(scope="function")
async def test_db(test_engine: AsyncEngine):
async_session = sessionmaker(
test_engine, class_=AsyncSession, expire_on_commit=False
)

async with engine.begin() as conn:
async with test_engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.drop_all)
await conn.run_sync(SQLModel.metadata.create_all)

async with engine.connect() as conn:
async with test_engine.connect() as conn:
await insert_rbac_seed_data_async(conn)

async def override_get_session():
Expand All @@ -99,12 +113,10 @@ async def override_get_session():

app.dependency_overrides[get_session] = override_get_session

yield async_session

async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.drop_all)

app.dependency_overrides.clear()
try:
yield async_session
finally:
app.dependency_overrides.clear()


@pytest.fixture
Expand Down
103 changes: 103 additions & 0 deletions tests/integration/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
import pytest
from fastapi_users.password import PasswordHelper

from src.auth.models import User
from src.auth.rbac.models import Permission, RolePermission, UserRole


@pytest.fixture
async def rbac_data(test_db):
async with test_db() as session:
perm_read = Permission(id=1, code="user:read", name="Read Users", module="user")
perm_write = Permission(
id=2, code="user:write", name="Write Users", module="user"
)
perm_delete = Permission(
id=3, code="user:delete", name="Delete Users", module="user"
)
perm_admin_all = Permission(
id=4, code="admin:*", name="Admin All", module="admin"
)
session.add_all([perm_read, perm_write, perm_delete, perm_admin_all])
await session.commit()

role_perms = [
RolePermission(role_id=1, permission_id=1),
RolePermission(role_id=1, permission_id=2),
RolePermission(role_id=1, permission_id=3),
RolePermission(role_id=1, permission_id=4),
RolePermission(role_id=2, permission_id=1),
]
session.add_all(role_perms)
await session.commit()


@pytest.fixture
async def admin_user(test_db, rbac_data):
password_helper = PasswordHelper()
hashed_password = password_helper.hash("admin123")

async with test_db() as session:
user = User(
username="adminuser",
email="admin@example.com",
hashed_password=hashed_password,
is_active=True,
is_verified=True,
is_superuser=False,
)
session.add(user)
await session.commit()
await session.refresh(user)

user_role = UserRole(user_id=user.id, role_id=1)
session.add(user_role)
await session.commit()

yield user


@pytest.fixture
async def regular_user(test_db, rbac_data):
password_helper = PasswordHelper()
hashed_password = password_helper.hash("user123")

async with test_db() as session:
user = User(
username="regularuser",
email="user@example.com",
hashed_password=hashed_password,
is_active=True,
is_verified=True,
is_superuser=False,
)
session.add(user)
await session.commit()
await session.refresh(user)

user_role = UserRole(user_id=user.id, role_id=2)
session.add(user_role)
await session.commit()

yield user


@pytest.fixture
async def superuser_user(test_db, rbac_data):
password_helper = PasswordHelper()
hashed_password = password_helper.hash("super123")

async with test_db() as session:
user = User(
username="superuser",
email="super@example.com",
hashed_password=hashed_password,
is_active=True,
is_verified=True,
is_superuser=True,
)
session.add(user)
await session.commit()
await session.refresh(user)

yield user
98 changes: 0 additions & 98 deletions tests/integration/test_audit_rbac.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,108 +4,10 @@
from sqlalchemy import select

from src.auth import require_permissions, require_roles
from src.auth.models import User
from src.auth.rbac.models import Permission, RolePermission, UserRole
from src.audit.schemas import AuditAction, AuditLog, AuditResult
from src.shared.errors import ErrorCode


@pytest.fixture
async def rbac_data(test_db):
async with test_db() as session:
perm_read = Permission(id=1, code="user:read", name="Read Users", module="user")
perm_write = Permission(
id=2, code="user:write", name="Write Users", module="user"
)
session.add_all([perm_read, perm_write])
await session.commit()

role_perms = [
RolePermission(role_id=1, permission_id=1),
RolePermission(role_id=1, permission_id=2),
RolePermission(role_id=2, permission_id=1),
]
session.add_all(role_perms)
await session.commit()


@pytest.fixture
async def admin_user(test_db, rbac_data):
from fastapi_users.password import PasswordHelper

password_helper = PasswordHelper()
hashed_password = password_helper.hash("admin123")

async with test_db() as session:
user = User(
username="adminuser",
email="admin@example.com",
hashed_password=hashed_password,
is_active=True,
is_verified=True,
is_superuser=False,
)
session.add(user)
await session.commit()
await session.refresh(user)

user_role = UserRole(user_id=user.id, role_id=1)
session.add(user_role)
await session.commit()

yield user


@pytest.fixture
async def regular_user(test_db, rbac_data):
from fastapi_users.password import PasswordHelper

password_helper = PasswordHelper()
hashed_password = password_helper.hash("user123")

async with test_db() as session:
user = User(
username="regularuser",
email="user@example.com",
hashed_password=hashed_password,
is_active=True,
is_verified=True,
is_superuser=False,
)
session.add(user)
await session.commit()
await session.refresh(user)

user_role = UserRole(user_id=user.id, role_id=2)
session.add(user_role)
await session.commit()

yield user


@pytest.fixture
async def superuser_user(test_db, rbac_data):
from fastapi_users.password import PasswordHelper

password_helper = PasswordHelper()
hashed_password = password_helper.hash("super123")

async with test_db() as session:
user = User(
username="superuser",
email="super@example.com",
hashed_password=hashed_password,
is_active=True,
is_verified=True,
is_superuser=True,
)
session.add(user)
await session.commit()
await session.refresh(user)

yield user


@pytest.fixture
async def rbac_audit_client(
test_db, local_cache, admin_user, regular_user, superuser_user
Expand Down
Loading