feat(L+M): configurable dashboard widget system + test framework
Phase L: Dashboard widget system - Migration 046: dashboard_configs table (user/tenant/role fallback cascade) - DashboardConfig model + dashboard_service with get/upsert per-user and tenant-default - API router: GET/PUT /api/dashboard/config, GET/PUT /api/dashboard/tenant-default - Frontend: 5 widget components (ProductionStats, QueueStatus, RecentRenders, CostOverview, WorkerStatus) - DashboardGrid with API-backed config, DashboardCustomizeModal (checkbox selection, save/cancel) - Dashboard.tsx: widget grid + analytics section (privileged users) - Admin.tsx: restructured with new section order and Maintenance 2-col grid Phase M: Test framework - Backend: pytest-asyncio + pytest-cov + factory-boy in pyproject.toml - conftest.py: excel_dir fixtures + async DB fixtures + mock storage/celery stubs - Domain tests: billing_service, cache_service, notifications_service, imports_sanity - Integration: test_api_health.py smoke test (requires running backend) - Frontend: vitest + @testing-library/react + msw added to package.json - vite.config.ts: test block (jsdom + globals + setupFiles) - tsconfig.json: exclude src/__tests__ from main tsc (test runner handles its own types) - MSW handlers for /api/auth/me, Billing.test.tsx, formatters.test.ts Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -109,3 +109,140 @@ def parsed_linear_schiene(parsed_excel_all):
|
||||
@pytest.fixture(scope="session")
|
||||
def parsed_anschlagplatten(parsed_excel_all):
|
||||
return parsed_excel_all["Anschlagplatten"]
|
||||
|
||||
|
||||
# ── Test-DB (nutzt separate Test-Datenbank) ──────────────────────────────────
|
||||
|
||||
import os
|
||||
import uuid
|
||||
import pytest_asyncio
|
||||
from typing import AsyncGenerator
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||||
|
||||
TEST_DB_URL = os.environ.get(
|
||||
"TEST_DATABASE_URL",
|
||||
"postgresql+asyncpg://schaeffler:schaeffler@localhost:5432/schaeffler_test"
|
||||
)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def test_engine():
|
||||
from app.database import Base
|
||||
from sqlalchemy import text
|
||||
import app.models # noqa - register all models
|
||||
engine = create_async_engine(TEST_DB_URL, echo=False)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield engine
|
||||
# Use CASCADE to handle circular FK dependencies in drop
|
||||
async with engine.begin() as conn:
|
||||
await conn.execute(text("DROP SCHEMA public CASCADE"))
|
||||
await conn.execute(text("CREATE SCHEMA public"))
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db(test_engine) -> AsyncGenerator[AsyncSession, None]:
|
||||
session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
await session.rollback()
|
||||
|
||||
|
||||
# ── Mock-Storage (LocalStorage in tmpdir) ───────────────────────────────────
|
||||
|
||||
@pytest.fixture
|
||||
def mock_storage(tmp_path, monkeypatch):
|
||||
"""Ersetzt MinIO-Storage durch lokalen tmpdir."""
|
||||
import app.core.storage as storage_module
|
||||
from unittest.mock import MagicMock
|
||||
mock = MagicMock()
|
||||
mock.exists.return_value = False
|
||||
mock.upload.return_value = "test/key"
|
||||
mock.download_bytes.return_value = b"fake-stl-data"
|
||||
mock.get_presigned_url.return_value = "http://localhost/presigned"
|
||||
monkeypatch.setattr(storage_module, "_storage_instance", mock)
|
||||
return mock
|
||||
|
||||
|
||||
# ── FastAPI Test-Client ──────────────────────────────────────────────────────
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(db) -> AsyncGenerator[AsyncClient, None]:
|
||||
"""HTTP-Client mit überschriebener DB-Dependency."""
|
||||
from app.main import app
|
||||
from app.database import get_db
|
||||
|
||||
async def override_get_db():
|
||||
yield db
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
|
||||
yield ac
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
# ── Test-User + Auth-Token ───────────────────────────────────────────────────
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def admin_user(db):
|
||||
from app.domains.auth.models import User, UserRole
|
||||
from app.utils.auth import hash_password
|
||||
user = User(
|
||||
id=uuid.uuid4(),
|
||||
email=f"test-admin-{uuid.uuid4().hex[:8]}@test.com",
|
||||
password_hash=hash_password("TestPass123!"),
|
||||
full_name="Test Admin",
|
||||
role=UserRole.admin,
|
||||
is_active=True,
|
||||
)
|
||||
db.add(user)
|
||||
await db.commit()
|
||||
await db.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_token(admin_user):
|
||||
from app.utils.auth import create_access_token
|
||||
return create_access_token(str(admin_user.id), admin_user.role.value)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_headers(admin_token):
|
||||
return {"Authorization": f"Bearer {admin_token}"}
|
||||
|
||||
|
||||
# ── Celery-Mock ──────────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_celery_tasks(monkeypatch):
|
||||
"""Verhindert echte Celery-Task-Dispatching in Tests."""
|
||||
from unittest.mock import MagicMock
|
||||
mock_result = MagicMock()
|
||||
mock_result.id = "test-task-" + str(uuid.uuid4())[:8]
|
||||
|
||||
def mock_delay(*args, **kwargs):
|
||||
return mock_result
|
||||
|
||||
task_paths = [
|
||||
"app.domains.materials.tasks.refresh_asset_library_catalog",
|
||||
"app.tasks.step_tasks.process_step_file",
|
||||
"app.tasks.step_tasks.render_step_thumbnail",
|
||||
"app.tasks.step_tasks.generate_stl_cache",
|
||||
"app.domains.imports.tasks.validate_excel_import",
|
||||
"app.domains.rendering.tasks.render_still_task",
|
||||
"app.domains.rendering.tasks.render_turntable_task",
|
||||
]
|
||||
for path in task_paths:
|
||||
try:
|
||||
parts = path.rsplit(".", 1)
|
||||
module_path, attr = parts
|
||||
import importlib
|
||||
module = importlib.import_module(module_path)
|
||||
task = getattr(module, attr, None)
|
||||
if task and hasattr(task, "delay"):
|
||||
monkeypatch.setattr(task, "delay", mock_delay)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
Reference in New Issue
Block a user