sentinel-home-ai/fam-edge/tests/test_qa.py

from fam_edge.qa import QAOrchestrator


class _FakeRoleAdapter:
    """用于 __init__ 过滤逻辑测试，只需要 get_role()，不需要真的能 chat。"""
    def __init__(self, provider_name, role):
        self.provider_name = provider_name
        self.role = role

    def get_role(self):
        return self.role


def test_init_only_keeps_text_role_adapters_in_config_order(monkeypatch):
    """核心诉求: 问答链路只用 role='text' 的适配器（跟视频分析用的
    role='vision' 完全隔离），且顺序沿用 config.yaml 里 models 数组的出现
    顺序，不需要额外的 qa_order 配置。"""
    fake_adapters = [
        _FakeRoleAdapter("gemini", "vision"),   # 视频分析用的 gemini-flash，不该出现
        _FakeRoleAdapter("nvidia", "vision"),    # 视频分析用的 nvidia omni，不该出现
        _FakeRoleAdapter("nvidia", "text"),      # 新的问答专用 nvidia 文字模型链
        _FakeRoleAdapter("gemini", "text"),      # 新的问答专用 gemini 非 flash 文字模型
        _FakeRoleAdapter("ollama", "text"),      # 本地兜底
    ]
    monkeypatch.setattr("fam_edge.qa.load_config", lambda: {"models": []})
    monkeypatch.setattr("fam_edge.qa.build_adapters", lambda models: fake_adapters)
    qa = QAOrchestrator()
    assert [a.role for a in qa.adapters] == ["text", "text", "text"]
    assert len(qa.adapters) == 3


class _FakeAdapter:
    def __init__(self, provider_name, chunks=None, raises=False):
        self.provider_name = provider_name
        self._chunks = chunks or []
        self._raises = raises

    def chat_stream(self, prompt, max_tokens=512):
        if self._raises:
            raise RuntimeError("boom")
        for c in self._chunks:
            yield c

    def chat(self, prompt, max_tokens=512):
        return ''.join(self._chunks) or None


def _orchestrator(adapters):
    qa = QAOrchestrator.__new__(QAOrchestrator)  # 跳过 __init__（不需要真实 config/adapters）
    qa.adapters = adapters
    return qa


def test_run_qa_stream_first_provider_success():
    qa = _orchestrator([_FakeAdapter("gemini", chunks=["你", "好"])])
    events = list(qa.run_qa_stream("hi"))
    types = [e["type"] for e in events]
    assert types == ["provider_trying", "chunk", "chunk", "done"]
    assert events[1]["text"] == "你"
    assert events[2]["text"] == "好"
    assert events[-1]["provider"] == "gemini"


def test_run_qa_stream_falls_back_when_first_yields_nothing():
    """核心诉求: 第一个 provider 一个字都没吐出来才允许换下一个——不是失败就切，
    是"完全没有产出"才切。"""
    qa = _orchestrator([
        _FakeAdapter("gemini", chunks=[]),
        _FakeAdapter("nvidia", chunks=["答案"]),
    ])
    events = list(qa.run_qa_stream("hi"))
    types = [e["type"] for e in events]
    assert types == ["provider_trying", "provider_failed", "provider_trying", "chunk", "done"]
    assert events[-1]["provider"] == "nvidia"


def test_run_qa_stream_does_not_switch_after_partial_output():
    """核心诉求: 已经开始吐字之后中途失败，不能悄悄换下一个 provider 接着写
    （会出现两段风格/内容不连贯的回答拼在一起）——直接结束这次生成。"""
    class _PartialThenRaise:
        provider_name = "gemini"
        def chat_stream(self, prompt, max_tokens=512):
            yield "先吐"
            raise RuntimeError("connection reset")

    qa = _orchestrator([_PartialThenRaise(), _FakeAdapter("nvidia", chunks=["不该被用到"])])
    events = list(qa.run_qa_stream("hi"))
    types = [e["type"] for e in events]
    assert types == ["provider_trying", "chunk", "done"]
    assert events[1]["text"] == "先吐"
    assert events[-1]["provider"] == "gemini"


def test_run_qa_stream_all_providers_fail():
    qa = _orchestrator([
        _FakeAdapter("gemini", chunks=[]),
        _FakeAdapter("nvidia", chunks=[], raises=True),
    ])
    events = list(qa.run_qa_stream("hi"))
    assert events[-1]["type"] == "all_failed"
    assert "provider_failed" in [e["type"] for e in events]


def test_run_qa_stream_exception_treated_as_no_output():
    qa = _orchestrator([_FakeAdapter("gemini", raises=True), _FakeAdapter("nvidia", chunks=["ok"])])
    events = list(qa.run_qa_stream("hi"))
    assert events[0] == {"type": "provider_trying", "provider": "gemini"}
    assert events[1] == {"type": "provider_failed", "provider": "gemini"}
    assert events[-1]["provider"] == "nvidia"