test_basic.py - master - AxeOn/ProjectAi

test_basic.py
284 строки10 KB
Дмитрий
feat(architecture): introduce core dataclasses in models.py and type DB returns
25 апр 2026, 16:23
25 апр 2026, 16:23cdbece0
# test_basic.py
"""
Базовый тест работоспособности системы

Запуск:
    python test_basic.py

Проверяет:
- Создание БД
- Добавление/чтение/удаление записей
- Валидацию дат
- Thread safety
"""

import sys
import os

# Добавляем текущую папку в путь
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from database import DBManager, ValidationError
import threading
import time


def test_database_basic():
    """Базовые операции с БД"""
    print("=" * 60)
    print("ТЕСТ 1: Базовые операции с БД")
    print("=" * 60)
    
    # Создаем тестовую БД
    db = DBManager("test_basic.db")
    
    try:
        # 1. Добавление сотрудника
        print("\n1. Добавление сотрудника...")
        emp_id = db.upsert_employee({
            'name': 'Тестовый Сотрудник',
            'department': 'Контроль',
            'email': 'test@example.com'
        })
        print(f"   ✅ Сотрудник создан с ID: {emp_id}")
        
        # 2. Чтение сотрудника
        print("\n2. Чтение сотрудника...")
        emp = db.get_employee_by_id(emp_id)
        assert emp.name == 'Тестовый Сотрудник'
        print(f"   ✅ Сотрудник прочитан: {emp.name}")

        # 3. Обновление сотрудника
        print("\n3. Обновление сотрудника...")
        db.upsert_employee({
            'id': emp_id,
            'name': 'Тестовый Сотрудник',
            'department': 'Расчет',
            'email': 'test@example.com'
        })
        emp = db.get_employee_by_id(emp_id)
        assert emp.department == 'Расчет'
        print(f"   ✅ Сотрудник обновлен: отдел = {emp.department}")
        
        # 4. Удаление сотрудника
        print("\n4. Удаление сотрудника...")
        db.delete_employee(emp_id)
        emp = db.get_employee_by_id(emp_id)
        assert emp is None
        print("   ✅ Сотрудник удален")
        
        print("\n✅ ТЕСТ 1 ПРОЙДЕН!")
        return True
        
    finally:
        db.close()
        # Удаляем тестовую БД
        if os.path.exists("test_basic.db"):
            os.remove("test_basic.db")


def test_date_validation():
    """Валидация дат"""
    print("\n" + "=" * 60)
    print("ТЕСТ 2: Валидация дат")
    print("=" * 60)
    
    db = DBManager("test_validation.db")
    
    try:
        # Создаем тестового сотрудника
        emp_id = db.upsert_employee({
            'name': 'Test User',
            'department': 'Test'
        })
        
        # 1. Корректная дата - должна пройти
        print("\n1. Корректная дата (2026-02-15)...")
        try:
            db.set_employee_day_override(emp_id, "2026-02-15", True)
            print("   ✅ Корректная дата принята")
        except ValidationError as e:
            print(f"   ❌ ОШИБКА: Корректная дата отклонена! {e}")
            return False
        
        # 2. Некорректная дата (короткая) - должна упасть
        print("\n2. Некорректная дата (2026-02)...")
        try:
            db.set_employee_day_override(emp_id, "2026-02", True)
            print("   ❌ ОШИБКА: Некорректная дата принята!")
            return False
        except ValidationError:
            print("   ✅ Некорректная дата отклонена")
        
        # 3. Некорректная дата (несуществующая) - должна упасть
        print("\n3. Несуществующая дата (2026-13-01)...")
        try:
            db.set_employee_day_override(emp_id, "2026-13-01", True)
            print("   ❌ ОШИБКА: Несуществующая дата принята!")
            return False
        except ValidationError:
            print("   ✅ Несуществующая дата отклонена")
        
        # 4. Задача с корректной датой
        print("\n4. Создание задачи с корректной датой...")
        try:
            task_id = db.add_task(
                "Тестовая задача",
                "Средний",
                "Test User",  # Сотрудник существует
                "2026-03-15"
            )
            print(f"   ✅ Задача создана с ID: {task_id}")
        except ValidationError as e:
            print(f"   ❌ ОШИБКА: Корректная дата отклонена! {e}")
            return False
        except Exception as e:
            print(f"   ❌ ОШИБКА при создании задачи: {e}")
            return False
        
        print("\n✅ ТЕСТ 2 ПРОЙДЕН!")
        return True
        
    finally:
        db.close()
        if os.path.exists("test_validation.db"):
            os.remove("test_validation.db")


def test_thread_safety():
    """Thread safety (многопоточность)"""
    print("\n" + "=" * 60)
    print("ТЕСТ 3: Thread Safety (10 потоков одновременно)")
    print("=" * 60)
    
    db = DBManager("test_threads.db")
    errors = []
    
    def worker(worker_id):
        """Рабочий поток"""
        try:
            for i in range(10):
                # Создаем сотрудника
                emp_id = db.upsert_employee({
                    'name': f'Worker_{worker_id}_Item_{i}',
                    'department': 'Test'
                })
                
                # Читаем его обратно
                emp = db.get_employee_by_id(emp_id)
                
                if emp is None:
                    errors.append(f"Worker {worker_id}: Employee {emp_id} not found!")
                
                # Небольшая задержка
                time.sleep(0.001)
            
            print(f"   ✅ Worker {worker_id} завершил работу")
            
        except Exception as e:
            errors.append(f"Worker {worker_id} error: {e}")
    
    try:
        # Запускаем 10 потоков
        print("\nЗапуск 10 потоков...")
        threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
        
        for t in threads:
            t.start()
        
        for t in threads:
            t.join()
        
        # Проверяем результаты
        if errors:
            print("\n❌ ОШИБКИ:")
            for error in errors:
                print(f"   {error}")
            return False
        
        # Проверяем количество записей
        employees = db.get_employees()
        expected_count = 10 * 10  # 10 потоков * 10 записей
        
        if len(employees) != expected_count:
            print(f"\n❌ ОШИБКА: Ожидалось {expected_count} записей, получено {len(employees)}")
            return False
        
        print(f"\n✅ Создано {len(employees)} записей без ошибок")
        print("\n✅ ТЕСТ 3 ПРОЙДЕН!")
        return True
        
    finally:
        # Закрываем все соединения
        try:
            db.close()
        except Exception as e:
            logger.debug(f"Error closing DB: {e}")
        
        # Даем время на закрытие всех соединений (особенно в Windows)
        time.sleep(1.0)
        
        # Удаляем тестовую БД (некритично если не получится)
        for filename in ["test_threads.db", "test_threads.db-wal", "test_threads.db-shm"]:
            try:
                if os.path.exists(filename):
                    os.remove(filename)
            except Exception as e:
                # Не критично - просто игнорируем
                pass


def main():
    """Запуск всех тестов"""
    print("\n")
    print("🧪 ТЕСТИРОВАНИЕ ОФИС-ПЛАНИРОВЩИКА v5.0")
    print("=" * 60)
    
    results = []
    
    # Тест 1: Базовые операции
    try:
        results.append(("Базовые операции с БД", test_database_basic()))
    except Exception as e:
        print(f"\n❌ ТЕСТ 1 ПРОВАЛЕН: {e}")
        results.append(("Базовые операции с БД", False))
    
    # Тест 2: Валидация дат
    try:
        results.append(("Валидация дат", test_date_validation()))
    except Exception as e:
        print(f"\n❌ ТЕСТ 2 ПРОВАЛЕН: {e}")
        results.append(("Валидация дат", False))
    
    # Тест 3: Thread safety
    try:
        results.append(("Thread Safety", test_thread_safety()))
    except Exception as e:
        print(f"\n❌ ТЕСТ 3 ПРОВАЛЕН: {e}")
        results.append(("Thread Safety", False))
    
    # Итоги
    print("\n" + "=" * 60)
    print("ИТОГИ ТЕСТИРОВАНИЯ")
    print("=" * 60)
    
    for name, passed in results:
        status = "✅ ПРОЙДЕН" if passed else "❌ ПРОВАЛЕН"
        print(f"{name:.<40} {status}")
    
    all_passed = all(passed for _, passed in results)
    
    print("=" * 60)
    if all_passed:
        print("🎉 ВСЕ ТЕСТЫ ПРОЙДЕНЫ!")
        print("Система работает корректно.")
    else:
        print("⚠️ НЕКОТОРЫЕ ТЕСТЫ НЕ ПРОЙДЕНЫ")
        print("Проверьте логи и исправьте ошибки.")
    print("=" * 60)
    
    return 0 if all_passed else 1


if __name__ == "__main__":
    sys.exit(main())