261 lines
9.2 KiB
Python
261 lines
9.2 KiB
Python
"""
|
||
Pytest fixtures for the HartOMat backend test suite.
|
||
|
||
The tests in this suite are divided into:
|
||
- Unit tests (no DB / network required): excel_parser, models, schemas
|
||
- Integration tests (require running Postgres + Redis): API endpoints, tasks
|
||
|
||
Unit tests run offline; integration tests are gated by the 'integration'
|
||
pytest mark so they can be skipped in CI without infrastructure.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Make sure the backend package is importable when tests are run from the
|
||
# repo root or from the backend/ directory.
|
||
# ---------------------------------------------------------------------------
|
||
BACKEND_DIR = Path(__file__).resolve().parent.parent # …/backend
|
||
if str(BACKEND_DIR) not in sys.path:
|
||
sys.path.insert(0, str(BACKEND_DIR))
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Paths
|
||
# ---------------------------------------------------------------------------
|
||
EXCEL_DIR = Path(__file__).resolve().parent.parent.parent / "Excel-Order-Lists"
|
||
|
||
EXCEL_FILES: dict[str, Path] = {
|
||
"TRB": EXCEL_DIR / "TRB_Testscope_20260128.xlsx",
|
||
"Kugellager": EXCEL_DIR / "Kugellager_Testscope_20260128.xlsx",
|
||
"CRB": EXCEL_DIR / "CRB_Testscope_20260128.xlsx",
|
||
"Gleitlager": EXCEL_DIR / "Gleitlager_Testscope_20260128.xlsx",
|
||
"SRB_TORB": EXCEL_DIR / "SRB_TORB_Testscope_20260128.xlsx",
|
||
"Linear_schiene": EXCEL_DIR / "Linear_schiene_Testscope_20260128.xlsx",
|
||
"Anschlagplatten": EXCEL_DIR / "Anschlagplatten_Testscope_20260128.xlsx",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures – Excel file paths
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.fixture(scope="session")
|
||
def excel_dir() -> Path:
|
||
"""Return the directory that contains all sample Excel order lists."""
|
||
assert EXCEL_DIR.is_dir(), f"Excel sample directory not found: {EXCEL_DIR}"
|
||
return EXCEL_DIR
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def excel_paths() -> dict[str, Path]:
|
||
"""Return a mapping of category key → absolute path for each sample file."""
|
||
missing = [k for k, p in EXCEL_FILES.items() if not p.exists()]
|
||
if missing:
|
||
pytest.skip(f"Sample Excel files missing: {missing}")
|
||
return EXCEL_FILES
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures – parsed Excel results (cached per test session)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_excel_all(excel_paths: dict[str, Path]) -> dict:
|
||
"""Parse all 7 sample Excel files and return {category_key: ParsedExcel}."""
|
||
from app.services.excel_parser import parse_excel
|
||
|
||
return {cat: parse_excel(path) for cat, path in excel_paths.items()}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers exposed as fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_trb(parsed_excel_all):
|
||
return parsed_excel_all["TRB"]
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_kugellager(parsed_excel_all):
|
||
return parsed_excel_all["Kugellager"]
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_crb(parsed_excel_all):
|
||
return parsed_excel_all["CRB"]
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_gleitlager(parsed_excel_all):
|
||
return parsed_excel_all["Gleitlager"]
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_srb_torb(parsed_excel_all):
|
||
return parsed_excel_all["SRB_TORB"]
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_linear_schiene(parsed_excel_all):
|
||
return parsed_excel_all["Linear_schiene"]
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def parsed_anschlagplatten(parsed_excel_all):
|
||
return parsed_excel_all["Anschlagplatten"]
|
||
|
||
|
||
# ── Test-DB (nutzt separate Test-Datenbank) ──────────────────────────────────
|
||
|
||
import uuid
|
||
import pytest_asyncio
|
||
from typing import AsyncGenerator
|
||
from httpx import AsyncClient, ASGITransport
|
||
from sqlalchemy.engine import make_url
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||
|
||
from app.config import settings
|
||
from tests.db_test_utils import reset_public_schema_async, resolve_test_db_url
|
||
|
||
|
||
def _resolve_test_db_url() -> str:
|
||
return resolve_test_db_url(async_driver=True)
|
||
|
||
|
||
def _sync_settings_to_test_database() -> None:
|
||
resolved = make_url(resolve_test_db_url(async_driver=False))
|
||
settings.postgres_host = resolved.host or settings.postgres_host
|
||
settings.postgres_port = int(resolved.port or settings.postgres_port)
|
||
settings.postgres_user = resolved.username or settings.postgres_user
|
||
settings.postgres_password = resolved.password or settings.postgres_password
|
||
settings.postgres_db = resolved.database or settings.postgres_db
|
||
|
||
|
||
_sync_settings_to_test_database()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def test_engine():
|
||
from app.database import Base
|
||
import app.models # noqa - register all models
|
||
engine = create_async_engine(_resolve_test_db_url(), echo=False)
|
||
async with engine.begin() as conn:
|
||
await reset_public_schema_async(conn)
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
yield engine
|
||
async with engine.begin() as conn:
|
||
await reset_public_schema_async(conn)
|
||
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_graph_thumbnail",
|
||
"app.tasks.step_tasks.render_step_thumbnail",
|
||
"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
|