patrick
6d4b8a9f17
- Migration 0024: RLS + FORCE RLS auf 18 Tabellen
- Direkte company_id-Policies: users, departments, companies, absence_types,
audit_logs, kiosk_devices, ldap_configs, smtp_configs, caldav_company_configs,
work_schedules, overtime_balances
- JOIN-Policies (user_id → company_id): absences, sessions, password_resets,
time_entries, vacation_balances, caldav_user_configs
- public_holidays ausgenommen (globale Referenztabelle)
- database.py: get_db setzt bypass_rls='on' als Default (Auth-Endpoints unverändert)
- dependencies.py: get_current_user setzt app.company_id + bypass_rls='off'
für alle nicht-SUPER_ADMIN Rollen
- migrations/env.py: Alembic-Migrationen nutzen bypass_rls='on'
- tests/conftest.py: override_get_db setzt bypass_rls='on' für Test-Session
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
import pytest_asyncio
|
||
from httpx import AsyncClient, ASGITransport
|
||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.main import app
|
||
from app.core.database import Base, get_db
|
||
from app.core.limiter import limiter
|
||
|
||
# Echte PostgreSQL Test-Datenbank (kein SQLite – Models nutzen JSONB/UUID)
|
||
TEST_DATABASE_URL = "postgresql+asyncpg://timemaster:timemaster_secret_change_me@localhost:5432/timemaster_test"
|
||
|
||
test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
|
||
TestSessionLocal = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session", autouse=True)
|
||
async def setup_db():
|
||
async with test_engine.begin() as conn:
|
||
# Schema komplett neu anlegen – löst circular dependency departments↔users
|
||
await conn.execute(text("DROP SCHEMA public CASCADE"))
|
||
await conn.execute(text("CREATE SCHEMA public"))
|
||
await conn.execute(text("GRANT ALL ON SCHEMA public TO timemaster"))
|
||
await conn.execute(text("GRANT ALL ON SCHEMA public TO public"))
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
yield
|
||
async with test_engine.begin() as conn:
|
||
await conn.execute(text("DROP SCHEMA public CASCADE"))
|
||
await conn.execute(text("CREATE SCHEMA public"))
|
||
await conn.execute(text("GRANT ALL ON SCHEMA public TO timemaster"))
|
||
await conn.execute(text("GRANT ALL ON SCHEMA public TO public"))
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def db_session():
|
||
async with TestSessionLocal() as session:
|
||
yield session
|
||
await session.rollback()
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def client(db_session: AsyncSession):
|
||
async def override_get_db():
|
||
try:
|
||
# Tests use a shared session without a real transaction context per
|
||
# request. Set bypass_rls = 'on' so that all test queries succeed
|
||
# regardless of whether app.company_id is set.
|
||
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
||
yield db_session
|
||
await db_session.commit()
|
||
except Exception:
|
||
await db_session.rollback()
|
||
raise
|
||
|
||
app.dependency_overrides[get_db] = override_get_db
|
||
limiter.enabled = False # Rate-Limiter in Tests deaktivieren
|
||
|
||
async with AsyncClient(
|
||
transport=ASGITransport(app=app),
|
||
base_url="http://test",
|
||
) as ac:
|
||
yield ac
|
||
|
||
limiter.enabled = True
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="session", loop_scope="session")
|
||
async def registered_user(client: AsyncClient):
|
||
"""Register a company + admin user, return tokens + user data."""
|
||
resp = await client.post("/api/v1/auth/register", json={
|
||
"company_name": "Test GmbH",
|
||
"first_name": "Max",
|
||
"last_name": "Mustermann",
|
||
"email": "max@testgmbh.de",
|
||
"password": "Secret123",
|
||
})
|
||
assert resp.status_code == 201
|
||
tokens = resp.json()
|
||
|
||
me = await client.get(
|
||
"/api/v1/auth/me",
|
||
headers={"Authorization": f"Bearer {tokens['access_token']}"},
|
||
)
|
||
return {"tokens": tokens, "user": me.json()}
|