Fix database

backend/tests/conftest.py

@@ -2,7 +2,7 @@
 import os
 import pytest
 from sqlalchemy import create_engine, text
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import sessionmaker, scoped_session
 from fastapi.testclient import TestClient
 from typing import Generator, Any
 
@@ -17,40 +17,43 @@ from .test_fixtures import *  # импортируем все фикстуры
 
 # Создаем тестовый движок базы данных
 DATABASE_URL = os.getenv("DATABASE_URL", test_settings.get_database_url())
-engine = create_engine(DATABASE_URL)
+engine = create_engine(
+    DATABASE_URL,
+    pool_size=5,
+    max_overflow=10,
+    pool_timeout=30,
+    pool_pre_ping=True
+)
 
 # Создаем тестовую фабрику сессий
-TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+TestingSessionLocal = scoped_session(
+    sessionmaker(autocommit=False, autoflush=False, bind=engine)
+)
 
 @pytest.fixture(scope="session", autouse=True)
 def setup_test_db() -> Generator[None, Any, None]:
     """Setup test database"""
-    try:
+    # Создаем все таблицы
-        # Создаем все таблицы
+    Base.metadata.drop_all(bind=engine)
-        Base.metadata.drop_all(bind=engine)  # Сначала удаляем все таблицы
+    Base.metadata.create_all(bind=engine)
-        Base.metadata.create_all(bind=engine)  # Затем создаем заново
+    yield
-        yield
+    # Удаляем все таблицы и закрываем соединения
-    finally:
+    TestingSessionLocal.remove()
-        # Удаляем все таблицы
+    Base.metadata.drop_all(bind=engine)
-        Base.metadata.drop_all(bind=engine)
+    engine.dispose()
-        # Закрываем соединение с тестовой базой
-        engine.dispose()
 
-@pytest.fixture
+@pytest.fixture(scope="function")
 def db_session() -> Generator[Any, Any, None]:
     """Get database session"""
-    connection = engine.connect()
+    session = TestingSessionLocal()
-    transaction = connection.begin()
-    session = TestingSessionLocal(bind=connection)
-    
     try:
         yield session
     finally:
+        session.rollback()
         session.close()
-        transaction.rollback()
+        TestingSessionLocal.remove()
-        connection.close()
 
-@pytest.fixture
+@pytest.fixture(scope="function")
 def client(db_session: Any) -> Generator[TestClient, Any, None]:
     """Get test client"""
     def override_get_db() -> Generator[Any, Any, None]:

backend/tests/test_fixtures.py

@@ -12,53 +12,73 @@ from app.models.employee import Employee
 @pytest.fixture(scope="function")
 def test_employee(db_session: Session) -> Employee:
     """Create test employee"""
-    # Удаляем существующего сотрудника, если есть
+    try:
-    db_session.query(Employee).filter(
+        # Удаляем существующего сотрудника, если есть
-        Employee.first_name == "Test",
+        db_session.query(Employee).filter(
-        Employee.last_name == "User"
+            Employee.first_name == "Test",
-    ).delete()
+            Employee.last_name == "User"
-    db_session.commit()
+        ).delete(synchronize_session=False)
-
+        
-    employee = EmployeeCreate(
+        employee = EmployeeCreate(
-        first_name="Test",
+            first_name="Test",
-        last_name="User",
+            last_name="User",
-        department="IT",
+            department="IT",
-        office="101",
+            office="101",
-        password="testpass123",
+            password="testpass123",
-        is_admin=False
+            is_admin=False
-    )
+        )
-    hashed_password = get_password_hash(employee.password)
+        hashed_password = get_password_hash(employee.password)
-    db_employee = employees.create_employee(db_session, employee, hashed_password)
+        db_employee = employees.create_employee(db_session, employee, hashed_password)
-    return db_employee
+        db_session.commit()
+        return db_employee
+    except Exception:
+        db_session.rollback()
+        raise
 
 @pytest.fixture(scope="function")
 def test_admin(db_session: Session) -> Employee:
     """Create test admin"""
-    # Удаляем существующего админа, если есть
+    try:
-    db_session.query(Employee).filter(
+        # Удаляем существующего админа, если есть
-        Employee.first_name == "Admin",
+        db_session.query(Employee).filter(
-        Employee.last_name == "User"
+            Employee.first_name == "Admin",
-    ).delete()
+            Employee.last_name == "User"
-    db_session.commit()
+        ).delete(synchronize_session=False)
-
+        
-    admin = EmployeeCreate(
+        admin = EmployeeCreate(
-        first_name="Admin",
+            first_name="Admin",
-        last_name="User",
+            last_name="User",
-        department="IT",
+            department="IT",
-        office="102",
+            office="102",
-        password="adminpass123",
+            password="adminpass123",
-        is_admin=True
+            is_admin=True
-    )
+        )
-    hashed_password = get_password_hash(admin.password)
+        hashed_password = get_password_hash(admin.password)
-    db_admin = employees.create_employee(db_session, admin, hashed_password)
+        db_admin = employees.create_employee(db_session, admin, hashed_password)
-    return db_admin
+        db_session.commit()
+        return db_admin
+    except Exception:
+        db_session.rollback()
+        raise
 
 @pytest.fixture(scope="function")
 def employee_token(db_session: Session, test_employee: Employee) -> str:
     """Get employee token"""
-    return create_and_save_token(test_employee.id, db_session)
+    try:
+        token = create_and_save_token(test_employee.id, db_session)
+        db_session.commit()
+        return token
+    except Exception:
+        db_session.rollback()
+        raise
 
 @pytest.fixture(scope="function")
 def admin_token(db_session: Session, test_admin: Employee) -> str:
     """Get admin token"""
-    return create_and_save_token(test_admin.id, db_session) 
+    try:
+        token = create_and_save_token(test_admin.id, db_session)
+        db_session.commit()
+        return token
+    except Exception:
+        db_session.rollback()
+        raise