ewoooc/tests/test_hermes_ollama_cascade.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Hermes 分析師必須透過 OllamaService 三主機級聯。"""

import time
import inspect
from types import SimpleNamespace

import pytest

import services.ai_call_logger as logger_mod
import services.hermes_analyst_service as hermes_mod
from services.ai_call_logger import _reset_kill_switch


@pytest.fixture(autouse=True)
def reset_ai_logger(monkeypatch):
    _reset_kill_switch()
    captured = []

    def fake_write(state):
        captured.append({
            'caller': state.caller,
            'provider': state.provider,
            'model': state.model,
            'status': state.status,
            'fallback_to': state.fallback_to,
            'error': state.error,
            'meta': dict(state.meta),
        })

    monkeypatch.setattr(logger_mod, '_write_to_db', fake_write)
    monkeypatch.setenv('AI_CALL_LOGGING_ENABLED', 'true')
    yield captured


def _wait_for(captured, n=1, timeout=2.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if len(captured) >= n:
            return True
        time.sleep(0.01)
    return False


def _stub_ollama(monkeypatch, *, content: str, host: str):
    fake_resp = SimpleNamespace(
        success=True,
        content=content,
        model=hermes_mod.HERMES_MODEL,
        error=None,
        total_duration=1.2,
        host=host,
        input_tokens=33,
        output_tokens=22,
    )

    class FakeOllamaService:
        instances = []

        def __init__(self, *args, **kwargs):
            self.init_args = args
            self.init_kwargs = kwargs
            self.generate_calls = []
            FakeOllamaService.instances.append(self)

        def generate(self, **kwargs):
            self.generate_calls.append(kwargs)
            return fake_resp

    monkeypatch.setattr(hermes_mod, 'OllamaService', FakeOllamaService)
    return FakeOllamaService


def test_hermes_intent_uses_ollama_service_and_logs_actual_host(monkeypatch, reset_ai_logger):
    fake_service = _stub_ollama(
        monkeypatch,
        content='{"intent":"query_sales","confidence":0.9,"complexity_score":0.8,'
                '"requires_data_fetch":true,"preliminary_answer":""}',
        host='http://34.21.145.224:11434',
    )

    svc = hermes_mod.HermesAnalystService()
    result = svc._call_hermes_intent("本週業績如何？")

    assert result['intent'] == 'query_sales'
    assert result['metadata']['source'] == 'hermes_llm'
    call_kwargs = fake_service.instances[0].generate_calls[0]
    assert call_kwargs['model'] == hermes_mod.HERMES_MODEL
    assert call_kwargs['keep_alive'] == hermes_mod.HERMES_KEEP_ALIVE
    assert call_kwargs['allow_111_fallback'] is False

    assert _wait_for(reset_ai_logger, 1)
    rec = reset_ai_logger[0]
    assert rec['caller'] == 'hermes_intent'
    assert rec['provider'] == 'ollama_secondary'
    assert rec['meta']['host_label'] == 'GCP-SSD-2'


def test_hermes_batch_analyze_uses_ollama_service_and_logs_secondary(monkeypatch, reset_ai_logger):
    fake_service = _stub_ollama(
        monkeypatch,
        content='[{"sku":"A1","name":"測試商品","category":"家電","momo_price":120,'
                '"pchome_price":100,"gap_pct":20,"sales_7d_delta_pct":-30,'
                '"risk":"HIGH","recommended_action":"建議人工評估","confidence":0.8}]',
        host='http://34.21.145.224:11434',
    )
    monkeypatch.setattr(hermes_mod, 'build_mcp_context', lambda *args, **kwargs: 'MCP context')

    candidates = [{
        'sku': 'A1',
        'name': '測試商品',
        'category': '家電',
        'momo_price': 120,
        'pchome_price': 100,
        'sales_7d_prev': 1000,
        'sales_7d_curr': 700,
        'competitor_tags': [
            'identity_v2',
            'match_type_same_product_different_pack',
            'price_basis_unit_price',
            'alert_tier_unit_price_review',
        ],
        'competitor_match_score': 0.74,
        'competitor_product_id': 'PCH-UNIT',
        'competitor_product_name': '測試商品 2入組',
    }]

    svc = hermes_mod.HermesAnalystService()
    raw_threats, items = svc._batch_analyze(candidates)

    assert raw_threats[0]['sku'] == 'A1'
    assert items[0]['gap_pct'] == 20.0
    assert items[0]['match_type'] == 'same_product_different_pack'
    assert items[0]['price_basis'] == 'unit_price'
    assert items[0]['alert_tier'] == 'unit_price_review'
    call_kwargs = fake_service.instances[0].generate_calls[0]
    assert call_kwargs['system_prompt'] == svc.SYSTEM_PROMPT
    assert call_kwargs['keep_alive'] == hermes_mod.HERMES_KEEP_ALIVE
    assert call_kwargs['allow_111_fallback'] is False

    assert _wait_for(reset_ai_logger, 1)
    rec = reset_ai_logger[0]
    assert rec['caller'] == 'hermes_analyst'
    assert rec['provider'] == 'ollama_secondary'
    assert rec['meta']['host_label'] == 'GCP-SSD-2'


def test_hermes_candidate_sql_only_joins_direct_price_alert_matches():
    sql_text = inspect.getsource(hermes_mod.HermesAnalystService.fetch_candidates)
    compact_sql = "".join(sql_text.split())

    assert '"match_type":"exact"' in compact_sql
    assert '"price_basis":"total_price"' in compact_sql
    assert '"alert_tier":"price_alert_exact"' in compact_sql
    assert "match_type_exact" in sql_text
    assert "price_basis_total_price" in sql_text
    assert "alert_tier_price_alert_exact" in sql_text


def test_hermes_keep_alive_defaults_to_short_runner_residency():
    assert hermes_mod.HERMES_KEEP_ALIVE == "5m"


def test_hermes_disables_111_fallback_by_default():
    assert hermes_mod.HERMES_ALLOW_111_FALLBACK is False