Bruno Charest
ecefbc8611
Some checks failed
Tests / Backend Tests (Python) (3.10) (push) Has been cancelled
Tests / Backend Tests (Python) (3.11) (push) Has been cancelled
Tests / Backend Tests (Python) (3.12) (push) Has been cancelled
Tests / Frontend Tests (JS) (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / All Tests Passed (push) Has been cancelled
412 lines
13 KiB
Python
412 lines
13 KiB
Python
"""
|
|
Fixtures partagées pour les tests backend.
|
|
|
|
Ce fichier configure:
|
|
- Base de données SQLite in-memory pour isolation totale
|
|
- Client HTTP async avec override des dépendances
|
|
- Mocks pour services externes (Ansible, ntfy, WebSocket)
|
|
- Factories pour créer des données de test
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from typing import AsyncGenerator, Generator
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import event
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
# Set test environment before imports
|
|
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
|
|
os.environ["API_KEY"] = "test-api-key-12345"
|
|
os.environ["JWT_SECRET_KEY"] = "test-jwt-secret-key-for-testing-only"
|
|
os.environ["NTFY_ENABLED"] = "false"
|
|
os.environ["ANSIBLE_DIR"] = "."
|
|
|
|
from app.models.database import Base
|
|
from app.core.dependencies import get_db, verify_api_key, get_current_user
|
|
from app import create_app
|
|
|
|
|
|
# ============================================================================
|
|
# DATABASE FIXTURES
|
|
# ============================================================================
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop() -> Generator:
|
|
"""Create event loop for session-scoped async fixtures."""
|
|
loop = asyncio.new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def async_engine():
|
|
"""Create async engine with in-memory SQLite."""
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
echo=False,
|
|
)
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
yield engine
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session(async_engine) -> AsyncGenerator[AsyncSession, None]:
|
|
"""Provide a transactional database session for each test."""
|
|
async_session_factory = async_sessionmaker(
|
|
async_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
autoflush=False,
|
|
)
|
|
|
|
async with async_session_factory() as session:
|
|
yield session
|
|
await session.rollback()
|
|
|
|
|
|
# ============================================================================
|
|
# APP & CLIENT FIXTURES
|
|
# ============================================================================
|
|
|
|
@pytest_asyncio.fixture
|
|
async def app(db_session: AsyncSession):
|
|
"""Create FastAPI app with overridden dependencies."""
|
|
application = create_app()
|
|
|
|
# Override database dependency
|
|
async def override_get_db():
|
|
yield db_session
|
|
|
|
# Override auth to always pass
|
|
async def override_verify_api_key():
|
|
return True
|
|
|
|
async def override_get_current_user():
|
|
return {
|
|
"type": "test",
|
|
"authenticated": True,
|
|
"user_id": "test-user-id",
|
|
"username": "testuser",
|
|
"role": "admin",
|
|
}
|
|
|
|
application.dependency_overrides[get_db] = override_get_db
|
|
application.dependency_overrides[verify_api_key] = override_verify_api_key
|
|
application.dependency_overrides[get_current_user] = override_get_current_user
|
|
|
|
yield application
|
|
|
|
application.dependency_overrides.clear()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client(app) -> AsyncGenerator[AsyncClient, None]:
|
|
"""Provide async HTTP client for testing endpoints."""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def unauthenticated_client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
|
|
"""Provide async HTTP client without auth overrides."""
|
|
application = create_app()
|
|
|
|
async def override_get_db():
|
|
yield db_session
|
|
|
|
application.dependency_overrides[get_db] = override_get_db
|
|
|
|
transport = ASGITransport(app=application)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
application.dependency_overrides.clear()
|
|
|
|
|
|
# ============================================================================
|
|
# SERVICE MOCKS
|
|
# ============================================================================
|
|
|
|
@pytest.fixture
|
|
def mock_ansible_service():
|
|
"""Mock AnsibleService to avoid real subprocess calls."""
|
|
with patch("app.services.ansible_service.ansible_service") as mock:
|
|
mock.get_playbooks.return_value = [
|
|
{
|
|
"name": "health-check",
|
|
"filename": "health-check.yml",
|
|
"path": "/playbooks/health-check.yml",
|
|
"category": "monitoring",
|
|
"subcategory": "other",
|
|
"hosts": "all",
|
|
"size": 1024,
|
|
"modified": datetime.now(timezone.utc).isoformat(),
|
|
"description": "Check host health",
|
|
}
|
|
]
|
|
mock.get_hosts_from_inventory.return_value = []
|
|
mock.get_groups.return_value = ["env_prod", "env_dev", "role_web"]
|
|
mock.get_env_groups.return_value = ["env_prod", "env_dev"]
|
|
mock.get_role_groups.return_value = ["role_web"]
|
|
mock.execute_playbook = AsyncMock(return_value={
|
|
"success": True,
|
|
"return_code": 0,
|
|
"stdout": "PLAY RECAP\nok=1 changed=0 failed=0",
|
|
"stderr": "",
|
|
"execution_time": 1.5,
|
|
"playbook": "health-check.yml",
|
|
"target": "all",
|
|
"check_mode": False,
|
|
})
|
|
mock.invalidate_cache = MagicMock()
|
|
mock.host_exists.return_value = False
|
|
mock.add_host_to_inventory = MagicMock()
|
|
mock.remove_host_from_inventory = MagicMock()
|
|
mock.update_host_groups = MagicMock()
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_ws_manager():
|
|
"""Mock WebSocket manager."""
|
|
with patch("app.services.websocket_service.ws_manager") as mock:
|
|
mock.broadcast = AsyncMock()
|
|
mock.connect = AsyncMock()
|
|
mock.disconnect = MagicMock()
|
|
mock.connection_count = 0
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_notification_service():
|
|
"""Mock notification service to avoid real HTTP calls."""
|
|
with patch("app.services.notification_service.notification_service") as mock:
|
|
mock.enabled = False
|
|
mock.send = AsyncMock(return_value=True)
|
|
mock.send_request = AsyncMock()
|
|
mock.notify_task_completed = AsyncMock(return_value=True)
|
|
mock.notify_task_failed = AsyncMock(return_value=True)
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_scheduler_service():
|
|
"""Mock scheduler service."""
|
|
with patch("app.services.scheduler_service.scheduler_service") as mock:
|
|
mock._started = False
|
|
mock.start_async = AsyncMock()
|
|
mock.shutdown = MagicMock()
|
|
mock.get_all_schedules.return_value = []
|
|
mock.get_schedule.return_value = None
|
|
mock.add_schedule_to_cache = MagicMock()
|
|
mock.remove_schedule_from_cache = MagicMock()
|
|
mock.validate_cron_expression.return_value = {
|
|
"valid": True,
|
|
"expression": "0 2 * * *",
|
|
"next_runs": [],
|
|
"error": None,
|
|
}
|
|
yield mock
|
|
|
|
|
|
# ============================================================================
|
|
# DATA FACTORIES
|
|
# ============================================================================
|
|
|
|
class HostFactory:
|
|
"""Factory for creating test hosts."""
|
|
_counter = 0
|
|
|
|
@classmethod
|
|
def build(cls, **kwargs) -> dict:
|
|
cls._counter += 1
|
|
return {
|
|
"id": kwargs.get("id", f"host-{cls._counter:04d}"),
|
|
"name": kwargs.get("name", f"test-host-{cls._counter}.local"),
|
|
"ip_address": kwargs.get("ip_address", f"192.168.1.{cls._counter}"),
|
|
"ansible_group": kwargs.get("ansible_group", "env_test"),
|
|
"status": kwargs.get("status", "unknown"),
|
|
"reachable": kwargs.get("reachable", False),
|
|
"last_seen": kwargs.get("last_seen"),
|
|
}
|
|
|
|
@classmethod
|
|
async def create(cls, session: AsyncSession, **kwargs) -> "Host":
|
|
from app.crud.host import HostRepository
|
|
data = cls.build(**kwargs)
|
|
repo = HostRepository(session)
|
|
host = await repo.create(**data)
|
|
await session.commit()
|
|
return host
|
|
|
|
|
|
class TaskFactory:
|
|
"""Factory for creating test tasks."""
|
|
_counter = 0
|
|
|
|
@classmethod
|
|
def build(cls, **kwargs) -> dict:
|
|
cls._counter += 1
|
|
return {
|
|
"id": kwargs.get("id", f"task-{cls._counter:04d}"),
|
|
"action": kwargs.get("action", "health-check"),
|
|
"target": kwargs.get("target", "all"),
|
|
"playbook": kwargs.get("playbook", "health-check.yml"),
|
|
"status": kwargs.get("status", "pending"),
|
|
}
|
|
|
|
@classmethod
|
|
async def create(cls, session: AsyncSession, **kwargs) -> "Task":
|
|
from app.crud.task import TaskRepository
|
|
data = cls.build(**kwargs)
|
|
repo = TaskRepository(session)
|
|
task = await repo.create(**data)
|
|
await session.commit()
|
|
return task
|
|
|
|
|
|
class UserFactory:
|
|
"""Factory for creating test users."""
|
|
_counter = 0
|
|
|
|
@classmethod
|
|
def build(cls, **kwargs) -> dict:
|
|
cls._counter += 1
|
|
from app.services.auth_service import hash_password
|
|
return {
|
|
"username": kwargs.get("username", f"testuser{cls._counter}"),
|
|
"hashed_password": kwargs.get("hashed_password", hash_password("testpassword123")),
|
|
"email": kwargs.get("email", f"test{cls._counter}@example.com"),
|
|
"display_name": kwargs.get("display_name", f"Test User {cls._counter}"),
|
|
"role": kwargs.get("role", "admin"),
|
|
"is_active": kwargs.get("is_active", True),
|
|
}
|
|
|
|
@classmethod
|
|
async def create(cls, session: AsyncSession, **kwargs) -> "User":
|
|
from app.crud.user import UserRepository
|
|
data = cls.build(**kwargs)
|
|
repo = UserRepository(session)
|
|
user = await repo.create(**data)
|
|
await session.commit()
|
|
return user
|
|
|
|
|
|
class ScheduleFactory:
|
|
"""Factory for creating test schedules."""
|
|
_counter = 0
|
|
|
|
@classmethod
|
|
def build(cls, **kwargs) -> dict:
|
|
cls._counter += 1
|
|
return {
|
|
"id": kwargs.get("id", f"schedule-{cls._counter:04d}"),
|
|
"name": kwargs.get("name", f"Test Schedule {cls._counter}"),
|
|
"playbook": kwargs.get("playbook", "health-check.yml"),
|
|
"target": kwargs.get("target", "all"),
|
|
"schedule_type": kwargs.get("schedule_type", "recurring"),
|
|
"recurrence_type": kwargs.get("recurrence_type", "daily"),
|
|
"recurrence_time": kwargs.get("recurrence_time", "02:00"),
|
|
"enabled": kwargs.get("enabled", True),
|
|
}
|
|
|
|
@classmethod
|
|
async def create(cls, session: AsyncSession, **kwargs) -> "Schedule":
|
|
from app.crud.schedule import ScheduleRepository
|
|
data = cls.build(**kwargs)
|
|
repo = ScheduleRepository(session)
|
|
schedule = await repo.create(**data)
|
|
await session.commit()
|
|
return schedule
|
|
|
|
|
|
@pytest.fixture
|
|
def host_factory():
|
|
"""Provide HostFactory."""
|
|
HostFactory._counter = 0
|
|
return HostFactory
|
|
|
|
|
|
@pytest.fixture
|
|
def task_factory():
|
|
"""Provide TaskFactory."""
|
|
TaskFactory._counter = 0
|
|
return TaskFactory
|
|
|
|
|
|
@pytest.fixture
|
|
def user_factory():
|
|
"""Provide UserFactory."""
|
|
UserFactory._counter = 0
|
|
return UserFactory
|
|
|
|
|
|
@pytest.fixture
|
|
def schedule_factory():
|
|
"""Provide ScheduleFactory."""
|
|
ScheduleFactory._counter = 0
|
|
return ScheduleFactory
|
|
|
|
|
|
# ============================================================================
|
|
# UTILITY FIXTURES
|
|
# ============================================================================
|
|
|
|
@pytest.fixture
|
|
def api_headers():
|
|
"""Provide standard API headers with auth."""
|
|
return {
|
|
"X-API-Key": "test-api-key-12345",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_ansible_dir(tmp_path):
|
|
"""Create temporary Ansible directory structure."""
|
|
ansible_dir = tmp_path / "ansible"
|
|
playbooks_dir = ansible_dir / "playbooks"
|
|
inventory_dir = ansible_dir / "inventory"
|
|
|
|
playbooks_dir.mkdir(parents=True)
|
|
inventory_dir.mkdir(parents=True)
|
|
|
|
# Create sample playbook
|
|
(playbooks_dir / "health-check.yml").write_text("""
|
|
- name: Health Check
|
|
hosts: all
|
|
tasks:
|
|
- name: Ping
|
|
ping:
|
|
""")
|
|
|
|
# Create sample inventory
|
|
(inventory_dir / "hosts.yml").write_text("""
|
|
all:
|
|
children:
|
|
env_test:
|
|
hosts:
|
|
test-host-1:
|
|
ansible_host: 192.168.1.10
|
|
""")
|
|
|
|
return ansible_dir
|