siemens_ragas/tests/test_advisor_writer.py

import shutil
import unittest
from pathlib import Path

from rag_eval.advisor.rules import Diagnosis
from rag_eval.advisor.writer import write_advice, _format_log_summary


class TestWriteAdvice(unittest.TestCase):
    def setUp(self):
        self.tmp = Path("tests/.tmp/test_advisor_writer")
        shutil.rmtree(self.tmp, ignore_errors=True)
        self.tmp.mkdir(parents=True, exist_ok=True)
        self.advice_path = self.tmp / "optimization_advice.md"

    def tearDown(self):
        shutil.rmtree(self.tmp, ignore_errors=True)

    def _make_diagnosis(self, metric="faithfulness", severity="warning"):
        return Diagnosis(
            metric=metric,
            mean_score=0.55,
            threshold=0.7,
            severity=severity,
            root_causes=["原因1", "原因2"],
            suggested_actions=["建议1", "建议2"],
            low_samples=[
                {"sample_id": "s1", "question": "问题1", "answer": "答案1",
                 "ground_truth": "标准1", metric: 0.4},
            ],
        )

    def test_write_creates_file(self):
        diag = self._make_diagnosis()
        write_advice(
            diagnoses=[diag],
            llm_markdown="## faithfulness\n\nLLM 建议内容",
            advice_path=self.advice_path,
            scenario_name="test-scenario",
            run_id="2026-01-01T00-00-00",
            judge_model="deepseek-v4-flash",
        )
        self.assertTrue(self.advice_path.exists())

    def test_write_contains_scenario_name_and_run_id(self):
        diag = self._make_diagnosis()
        write_advice(
            diagnoses=[diag],
            llm_markdown="## faithfulness\n\nLLM 建议",
            advice_path=self.advice_path,
            scenario_name="siemens-test",
            run_id="2026-01-01T00-00-00",
            judge_model="deepseek-v4-flash",
        )
        content = self.advice_path.read_text(encoding="utf-8")
        self.assertIn("siemens-test", content)
        self.assertIn("2026-01-01T00-00-00", content)

    def test_write_contains_llm_markdown(self):
        diag = self._make_diagnosis()
        write_advice(
            diagnoses=[diag],
            llm_markdown="## faithfulness\n\n具体建议文本",
            advice_path=self.advice_path,
            scenario_name="test",
            run_id="rid",
            judge_model="model",
        )
        content = self.advice_path.read_text(encoding="utf-8")
        self.assertIn("具体建议文本", content)

    def test_write_fallback_when_no_llm_markdown(self):
        """When llm_markdown is empty, writer emits rule-only report."""
        diag = self._make_diagnosis()
        write_advice(
            diagnoses=[diag],
            llm_markdown="",
            advice_path=self.advice_path,
            scenario_name="test",
            run_id="rid",
            judge_model="model",
        )
        content = self.advice_path.read_text(encoding="utf-8")
        self.assertIn("faithfulness", content)
        self.assertIn("原因1", content)

    def test_log_summary_format(self):
        diags = [
            self._make_diagnosis("faithfulness", "critical"),
            self._make_diagnosis("context_recall", "warning"),
        ]
        summary = _format_log_summary(diags, self.advice_path)
        self.assertIn("faithfulness", summary)
        self.assertIn("严重", summary)   # "critical" maps to Chinese label
        self.assertIn("context_recall", summary)
        self.assertIn("警告", summary)   # "warning" maps to Chinese label

    def test_write_empty_diagnoses_still_creates_file(self):
        write_advice(
            diagnoses=[],
            llm_markdown="",
            advice_path=self.advice_path,
            scenario_name="test",
            run_id="rid",
            judge_model="model",
        )
        self.assertTrue(self.advice_path.exists())
        content = self.advice_path.read_text(encoding="utf-8")
        self.assertIn("未发现明显指标异常", content)


if __name__ == "__main__":
    unittest.main()