from datetime import date

import pytest
from app.services.rag import AnomalyCaseRAG


def _row(**values):
    return {
        "id": 1,
        "date": date(2026, 1, 10),
        "product_name": "洗发水 500ml",
        "product_category": "洗护类",
        "material_type": "软管",
        "supplier": "祥印",
        "factory": "晶品",
        "stage_new": "产品开发",
        "anomaly_type": "密封性异常",
        "description": "软管漏液",
        "solution": "增加密封性测试",
        "root_cause": "封口参数不稳定",
        "defect_level": "B",
        **values,
    }


def test_rag_terms_and_structured_text_score() -> None:
    terms = AnomalyCaseRAG._terms("软管密封性异常", "洗发水", "软管")
    score = AnomalyCaseRAG._score(
        _row(),
        query="洗发水软管密封性异常",
        category="洗发水",
        material="软管",
        supplier="祥印",
        factory=None,
        stage="产品开发",
        anomaly_type="密封性异常",
        terms=terms,
    )

    assert "密封性" in terms
    assert score > 20
    case = AnomalyCaseRAG._to_case(_row())
    assert case.product == "洗发水 500ml"
    assert case.problem == "密封性异常"
    assert case.solution == "增加密封性测试"


@pytest.mark.asyncio
async def test_rag_does_not_call_embedding_until_vectors_are_stored() -> None:
    class FakeMappings:
        def all(self):
            return [_row()]

    class FakeResult:
        def mappings(self):
            return FakeMappings()

    class FakeSession:
        async def execute(self, _statement):
            return FakeResult()

    rag = AnomalyCaseRAG(FakeSession(), top_k=1)

    async def vector_column_ready():
        return True

    async def has_stored_vectors():
        return False

    async def embedding_must_not_run(_query):
        raise AssertionError("没有已回填向量时不应调用 Embedding")

    rag._vector_column_ready = vector_column_ready
    rag._has_stored_vectors = has_stored_vectors
    rag._query_embedding = embedding_must_not_run

    rows = await rag._load_rows(
        "洗发水软管密封性异常",
        category="洗发水",
        material="软管",
        supplier=None,
        factory=None,
        stage=None,
        anomaly_type=None,
    )

    assert len(rows) == 1