tests/test_ensemble_integration.py - main - Paff/declarant

tests/test_ensemble_integration.py
297 строк10 KB
1pash1985-gif
feat: AI Ensemble - DeepSeek + Kimi parallel parsing with debates
27 июл 2026, 11:14
27 июл 2026, 11:145ea7945
"""
Интеграционные тесты для AI-ансамбля.

Требуют переменные окружения:
- DEEPSEEK_API_KEY
- KIMI_API_KEY

Запуск:
    # Только быстрые тесты (без API):
    pytest -m "not integration and not slow"

    # Все тесты (требует API-ключи):
    pytest -m "integration"

    # Все тесты:
    pytest
"""

import json
import os
import sys
from pathlib import Path

import pytest

# Добавляем web/ в путь для импорта модулей
sys.path.insert(0, str(Path(__file__).parent.parent / 'web'))


# ============================================================================
# Fixtures
# ============================================================================

@pytest.fixture
def sample_invoice_text():
    """Пример текста инвойса для тестирования."""
    return """
INVOICE #INV-2025-001
Date: 2025-07-15

Supplier: FLORES AURORA S.A.
Buyer: CACTUS LLC

Items:
1. ROSE RED PINK - 100 stems - $0.50/stem
2. CARNATION WHITE - 200 stems - $0.30/stem
3. GYPSOPHILIA - 50 stems - $0.80/stem

Total: $230.00
"""


@pytest.fixture
def sample_packing_text():
    """Пример текста упаковочного листа."""
    return """
PACKING LIST
Date: 2025-07-15

Supplier: KENYA FLOWERS LTD
Buyer: EURO FLORA

Shipment: KE-2025-0789
Boxes: 25
Weight: 150 kg

Items:
1. ROSE AVALANCHE - 500 stems - 10 boxes
2. CARNATION DYNAMIC YELLOW - 300 stems - 5 boxes
3. STATICE - 200 stems - 3 boxes
"""


@pytest.fixture
def mock_ensemble_meta():
    """Пример метаданных ансамбля."""
    return {
        'models': ['deepseek-chat', 'kimi-k3'],
        'rounds': 2,
        'disputes': [
            {
                'path': 'items.0.stems',
                'values': {'deepseek-chat': 100, 'kimi-k3': 110},
                'resolution': 'operator',
            }
        ],
        'confidence': 0.85,
    }


# ============================================================================
# Unit Tests (без API)
# ============================================================================

class TestEnsembleHelpers:
    """Тесты вспомогательных функций ансамбля."""

    def test_load_ensemble_config(self):
        """Проверка загрузки конфигурации ансамбля."""
        from config import (
            ENSEMBLE_ENABLED,
            ENSEMBLE_ENABLE_PARSING,
            ENSEMBLE_DEBATE_MAX_ROUNDS,
            ENSEMBLE_TIMEOUT_SECONDS,
        )
        # Конфигурация должна загружаться без ошибок
        assert isinstance(ENSEMBLE_ENABLED, bool)
        assert isinstance(ENSEMBLE_ENABLE_PARSING, bool)
        assert isinstance(ENSEMBLE_DEBATE_MAX_ROUNDS, int)
        assert ENSEMBLE_DEBATE_MAX_ROUNDS >= 0
        assert isinstance(ENSEMBLE_TIMEOUT_SECONDS, int)
        assert ENSEMBLE_TIMEOUT_SECONDS > 0

    def test_json_diff_identical(self):
        """_json_diff должен возвращать пустой список для идентичных JSON."""
        from ai_ensemble import _json_diff

        a = {'supplier': 'ROSE CO', 'items': [{'marking': 'A1', 'stems': 100}]}
        b = {'supplier': 'ROSE CO', 'items': [{'marking': 'A1', 'stems': 100}]}

        diffs = _json_diff(a, b)

        assert len(diffs) == 0

    def test_json_diff_with_dispute(self):
        """_json_diff должен обнаруживать расхождения."""
        from ai_ensemble import _json_diff, _get_at_path

        a = {'supplier': 'ROSE CO', 'total': 100}
        b = {'supplier': 'ROSES INC', 'total': 100}

        diffs = _json_diff(a, b)

        assert len(diffs) == 1
        assert diffs[0] == ['supplier']  # path is a list of keys
        assert _get_at_path(a, diffs[0]) == 'ROSE CO'
        assert _get_at_path(b, diffs[0]) == 'ROSES INC'

    def test_apply_field_rules(self):
        """_apply_field_rules_to_result должен применять правила к результату."""
        from ai_ensemble import _apply_field_rules_to_result

        # Без правил должно вернуть оригинал
        result = {'supplier': 'ROSE CO', 'items': [{'stems': 100}]}
        original = _apply_field_rules_to_result(result, None)
        assert original == result

    def test_load_rules(self):
        """_load_rules должен загружать правила из corrections.json."""
        from ai_ensemble import _load_rules

        # Должно работать без ошибок (может вернуть None если corrections.json пустой)
        rules = _load_rules()
        assert rules is None or isinstance(rules, dict)


class TestFallback:
    """Тесты fallback-поведения."""

    def test_parse_without_api_keys(self, sample_invoice_text):
        """Парсинг без API-ключей должен вернуть ошибку."""
        # Временно очищаем API-ключи
        original_deepseek = os.environ.get('DEEPSEEK_API_KEY')
        original_kimi = os.environ.get('KIMI_API_KEY')

        try:
            os.environ['DEEPSEEK_API_KEY'] = ''
            os.environ['KIMI_API_KEY'] = ''

            # Перезагружаем config
            import importlib
            import config
            importlib.reload(config)

            from config import DEEPSEEK_API_KEY, KIMI_API_KEY
            assert not DEEPSEEK_API_KEY
            assert not KIMI_API_KEY

        finally:
            # Восстанавливаем оригинальные значения
            if original_deepseek:
                os.environ['DEEPSEEK_API_KEY'] = original_deepseek
            else:
                os.environ.pop('DEEPSEEK_API_KEY', None)

            if original_kimi:
                os.environ['KIMI_API_KEY'] = original_kimi
            else:
                os.environ.pop('KIMI_API_KEY', None)

            # Перезагружаем config
            import importlib
            import config
            importlib.reload(config)


# ============================================================================
# Integration Tests (требуют API-ключи)
# ============================================================================

@pytest.mark.integration
class TestEnsembleIntegration:
    """Интеграционные тесты с реальными API-вызовами."""

    @pytest.fixture(autouse=True)
    def check_api_keys(self):
        """Пропускаем тесты, если нет API-ключей."""
        from config import DEEPSEEK_API_KEY, KIMI_API_KEY
        if not DEEPSEEK_API_KEY or not KIMI_API_KEY:
            pytest.skip("Требуется DEEPSEEK_API_KEY и KIMI_API_KEY")

    def test_parse_invoice_with_ensemble(self, sample_invoice_text):
        """Парсинг инвойса ансамблем должен возвращать валидный JSON."""
        from ai_ensemble import parse_invoice_with_ensemble

        result = parse_invoice_with_ensemble(sample_invoice_text)

        assert 'error' not in result or result.get('error') is None
        assert 'supplier' in result
        assert 'items' in result
        assert isinstance(result['items'], list)

    def test_parse_packing_with_ensemble(self, sample_packing_text):
        """Парсинг упаковочного листа ансамблем."""
        from ai_ensemble import parse_packing_with_ensemble

        result = parse_packing_with_ensemble(sample_packing_text)

        assert 'error' not in result or result.get('error') is None
        assert 'supplier' in result

    def test_concurrent_model_calls(self, sample_invoice_text):
        """Параллельные вызовы моделей должны работать."""
        from ai_ensemble import _call_model
        from concurrent.futures import ThreadPoolExecutor

        def call_deepseek():
            return _call_model(
                'deepseek-chat',
                'You are a parser.',
                f'Parse this invoice: {sample_invoice_text}',
                temperature=0.1,
                max_tokens=1000,
            )

        def call_kimi():
            return _call_model(
                'kimi-k3',
                'You are a parser.',
                f'Parse this invoice: {sample_invoice_text}',
                temperature=1.0,  # Kimi требует temperature=1.0
                max_tokens=1000,
            )

        with ThreadPoolExecutor(max_workers=2) as executor:
            future_deepseek = executor.submit(call_deepseek)
            future_kimi = executor.submit(call_kimi)

            result_deepseek = future_deepseek.result(timeout=300)
            result_kimi = future_kimi.result(timeout=300)

        assert 'error' not in result_deepseek
        assert 'error' not in result_kimi


@pytest.mark.slow
@pytest.mark.integration
class TestEnsembleDebate:
    """Тесты дебатов (медленные, требуют API)."""

    @pytest.fixture(autouse=True)
    def check_api_keys(self):
        """Пропускаем тесты, если нет API-ключей."""
        from config import DEEPSEEK_API_KEY, KIMI_API_KEY
        if not DEEPSEEK_API_KEY or not KIMI_API_KEY:
            pytest.skip("Требуется DEEPSEEK_API_KEY и KIMI_API_KEY")

    def test_debate_resolves_disputes(self, sample_invoice_text):
        """Дебаты должны разрешать расхождения."""
        from ai_ensemble import run_debate

        # Создаём искусственные расхождения
        result_a = {'supplier': 'ROSE CO', 'items': [{'stems': 100}]}
        result_b = {'supplier': 'ROSES INC', 'items': [{'stems': 100}]}

        disputes = [
            {
                'path': 'supplier',
                'values': {'deepseek': 'ROSE CO', 'kimi': 'ROSES INC'},
                'resolution': 'operator',
            }
        ]

        votes = run_debate(sample_invoice_text, result_a, result_b, disputes)

        # Дебаты должны вернуть голоса (может быть пустым, если модели не согласились)
        assert isinstance(votes, dict)