tests/test_app.py

import json
import unittest
from unittest.mock import patch

from pydantic import ValidationError

from app import DEPTHS, SEARCH_CACHE, ResearchReport, SectionPlan, align_sections_to_plan, cached_search, chat_json, deduplicate, infer_section_source_ids, normalize_comparison, normalize_process, parse_json_object, prepare_evidence_inputs, preview_frame, render_presentation, render_section_content, render_value, safe_slug, search_literature, serializable_validation_errors


def base_report(section):
    return {
        "title": "Test report", "subtitle": "Topic-adaptive structured output",
        "research_type": "engineering decision",
        "executive_summary": ["A substantive conclusion supported by the report."],
        "sections": [dict(section, id=f"section-{i}") for i in range(3)],
        "references": [], "limitations": ["Recent literature was not externally verified."],
        "english_understanding": {},
    }


class AppTests(unittest.TestCase):
    def test_slug(self):
        self.assertEqual(safe_slug("Diffusion Transformer 研究"), "diffusion-transformer")

    def test_deduplicate_versions(self):
        self.assertEqual(len(deduplicate([{"id": "1234.5678v1"}, {"id": "1234.5678v2"}])), 1)

    def test_evidence_inputs_are_bounded_and_prioritize_deep_reads(self):
        papers = [{"id": str(i), "title": f"Paper {i}", "summary": "Summary"} for i in range(10)]
        result = prepare_evidence_inputs(papers, {"8": "deep text", "9": "deep text"}, 3)
        self.assertEqual(len(result), 3)
        self.assertEqual([item["id"] for item in result[:2]], ["8", "9"])
        self.assertIn("selected_text_extract", result[0])

    @patch("app.openalex_search", return_value=[])
    @patch("app.arxiv_search")
    def test_search_balances_multiple_query_intents(self, mocked_arxiv, _mocked_openalex):
        mocked_arxiv.side_effect = [
            [{"id": f"classic-{i}", "title": f"Classic {i}"} for i in range(5)],
            [{"id": f"recent-{i}", "title": f"Recent {i}"} for i in range(5)],
        ]
        papers, warnings = search_literature(["classic", "recent"], 4)
        self.assertEqual([paper["id"] for paper in papers], ["classic-0", "recent-0", "classic-1", "recent-1"])
        self.assertEqual(warnings, [])

    @patch("app.openalex_search", return_value=[{"id": "oa-1", "title": "Fallback"}])
    @patch("app.arxiv_search", return_value=[])
    def test_search_uses_openalex_when_arxiv_returns_empty(self, _mocked_arxiv, mocked_openalex):
        papers, _ = search_literature(["topic"], 5)
        self.assertEqual(papers[0]["id"], "oa-1")
        mocked_openalex.assert_called_once()

    @patch("app.openalex_search", side_effect=TimeoutError("offline"))
    @patch("app.arxiv_search", side_effect=TimeoutError("offline"))
    def test_search_stops_repeating_failed_providers(self, mocked_arxiv, mocked_openalex):
        papers, warnings = search_literature([f"query-{i}" for i in range(5)], 20)
        self.assertEqual(papers, [])
        self.assertEqual(mocked_arxiv.call_count, 2)
        self.assertEqual(mocked_openalex.call_count, 2)
        self.assertTrue(any("已停止剩余重复请求" in warning for warning in warnings))

    def test_search_cache_avoids_duplicate_external_calls(self):
        SEARCH_CACHE.clear()
        search_fn = unittest.mock.Mock(return_value=[{"id": "cached", "title": "Cached paper"}])
        first = cached_search("test-provider", "same query", 5, search_fn)
        second = cached_search("test-provider", "same query", 5, search_fn)
        self.assertEqual(first, second)
        self.assertEqual(search_fn.call_count, 1)

    def test_depth_increases_analysis_not_unbounded_search_calls(self):
        self.assertEqual([DEPTHS[name]["queries"] for name in ("quick", "standard", "deep")], [2, 3, 4])
        self.assertEqual([DEPTHS[name]["deep"] for name in ("quick", "standard", "deep")], [2, 4, 8])

    def test_missing_standard_and_deep_sections_are_exposed_for_repair(self):
        plans = [
            SectionPlan(id=f"section-{i}", title=f"Section {i}", purpose="Explain one distinct technical decision.", layout="text")
            for i in range(4)
        ]
        partial = {"sections": [{"id": "section-0", "takeaway": "Only the first section was returned.", "content": {"points": ["A substantive point"]}}]}
        aligned = align_sections_to_plan(partial, plans)
        self.assertEqual(len(aligned["sections"]), 4)
        self.assertEqual([item["id"] for item in aligned["sections"]], [plan.id for plan in plans])
        self.assertEqual(aligned["sections"][1]["content"], {})

    def test_parser_removes_invalid_string_line_continuation(self):
        self.assertEqual(parse_json_object('{"title":"A\\\nB","value":1}'), {"title": "AB", "value": 1})

    @patch("app.generate_text", side_effect=['{"title":"broken\\\n', '{"title":"valid"}'])
    def test_json_generation_retries_malformed_response(self, mocked_generate):
        self.assertEqual(chat_json("system", "prompt", 2048), {"title": "valid"})
        self.assertEqual(mocked_generate.call_count, 2)
        self.assertIn("Regenerate from scratch", mocked_generate.call_args.args[0])

    def test_render_is_downloadable_document(self):
        section = {"title": "Scope", "purpose": "Define the technical decision boundary.", "takeaway": "The boundary determines the useful evidence.", "layout": "text", "content": {"points": ["A meaningful point", "Another meaningful point"]}, "source_ids": []}
        output = render_presentation(base_report(section), [], "2026-01-01")
        self.assertIn("<!doctype html>", output)
        self.assertIn('<button id="print" title="打印或保存为 PDF">保存 PDF</button>', output)
        self.assertIn('id="overview"', output)
        self.assertIn('id="agenda"', output)
        self.assertIn('data-slide-target="section-1"', output)

    def test_preview_allows_browser_print_dialog(self):
        frame = preview_frame("<!doctype html><title>Deck</title>")
        self.assertIn('sandbox="allow-scripts allow-popups allow-modals"', frame)

    def test_source_chip_links_to_rendered_reference(self):
        section = {"title": "Evidence", "purpose": "Connect a claim to retrieved evidence.", "takeaway": "The cited result supports this bounded conclusion.", "layout": "evidence", "content": {"claim": "A supported claim", "conditions": ["A stated condition"]}, "source_ids": ["paper-1"]}
        report = base_report(section)
        report["references"] = ["paper-1"]
        paper = {"id": "paper-1", "title": "A Retrieved Paper", "authors": ["A. Author"], "published": "2025-01-01", "url": "https://example.com/paper"}
        output = render_presentation(report, [paper], "2026-01-01")
        self.assertIn('href="#ref-paper-1"', output)
        self.assertIn('id="ref-paper-1"', output)

    def test_empty_report_is_rejected(self):
        with self.assertRaises(ValidationError):
            ResearchReport.model_validate({})

    def test_technical_evolution_report(self):
        section = {"title": "Evolution", "purpose": "Explain changes across technical generations.", "takeaway": "Each generation resolves a prior bottleneck.", "layout": "timeline", "content": {"nodes": [{"year": "2020", "title": "A", "innovation": "First mechanism"}, {"year": "2024", "title": "B", "innovation": "Second mechanism"}]}, "source_ids": []}
        self.assertEqual(ResearchReport.model_validate(base_report(section)).sections[0].layout, "timeline")

    def test_paper_analysis_report(self):
        section = {"title": "Paper", "purpose": "Assess the paper contribution and evidence.", "takeaway": "The method trades accuracy for efficiency.", "layout": "paper-analysis", "content": {"method": "A detailed method", "evidence": ["A reported experiment"], "limitations": ["A narrow benchmark"]}, "source_ids": []}
        self.assertEqual(ResearchReport.model_validate(base_report(section)).sections[0].layout, "paper-analysis")

    def test_evidence_gap_report(self):
        section = {"title": "Evidence gap", "purpose": "Separate known facts from hypotheses.", "takeaway": "More evidence is required before choosing a method.", "layout": "evidence-gap", "content": {"interpretation": "The term may be ambiguous.", "related_topics": ["Adjacent field"], "hypotheses": ["Possible interpretation"], "required_inputs": ["Definition or internal data"]}, "source_ids": []}
        self.assertEqual(ResearchReport.model_validate(base_report(section)).sections[0].layout, "evidence-gap")

    def test_evidence_gap_accepts_equivalent_semantic_fields(self):
        section = {"title": "Evidence gap", "purpose": "Separate available evidence from unresolved questions.", "takeaway": "Deployment requires evidence that is not yet public.", "layout": "evidence-gap", "content": {"known_evidence": ["A related result exists"], "unknowns": ["No direct benchmark"], "next_steps": ["Run an internal evaluation"]}, "source_ids": []}
        validated = ResearchReport.model_validate(base_report(section))
        self.assertEqual(validated.sections[0].content["unknowns"], ["No direct benchmark"])

    def test_paper_analysis_accepts_equivalent_semantic_fields(self):
        section = {"title": "Paper analysis", "purpose": "Analyze a paper without enforcing field spelling.", "takeaway": "The result is useful under a bounded condition.", "layout": "paper-analysis", "content": {"method_summary": "A compact architecture", "experimental_results": ["A controlled comparison"], "caveats": ["A narrow dataset"]}, "source_ids": []}
        self.assertEqual(ResearchReport.model_validate(base_report(section)).sections[0].layout, "paper-analysis")

    def test_invalid_empty_section(self):
        section = {"title": "Empty", "purpose": "This section should fail validation.", "takeaway": "There is no substantive supporting content.", "layout": "text", "content": {}, "source_ids": []}
        with self.assertRaises(ValidationError):
            ResearchReport.model_validate(base_report(section))

    def test_validation_feedback_is_json_serializable(self):
        section = {"title": "Invalid", "purpose": "Trigger a custom validator error.", "takeaway": "The timeline is intentionally incomplete.", "layout": "timeline", "content": {"nodes": []}, "source_ids": []}
        try:
            ResearchReport.model_validate(base_report(section))
        except ValidationError as exc:
            errors = serializable_validation_errors(exc)
        else:
            self.fail("Expected report validation to fail")
        json.dumps(errors)
        self.assertNotIn("ctx", errors[0])

    def test_comparison_requires_rows(self):
        section = {"title": "Compare", "purpose": "Compare engineering alternatives fairly.", "takeaway": "No choice is universally optimal here.", "layout": "comparison", "content": {"columns": ["speed", "quality"], "rows": []}, "source_ids": []}
        with self.assertRaises(ValidationError):
            ResearchReport.model_validate(base_report(section))

    def test_nested_json_is_rendered_as_semantic_html(self):
        output = render_value({"findings": [{"claim": "Works well", "conditions": ["GPU", "FP16"]}]})
        self.assertIn("<h4>", output)
        self.assertIn("Works well", output)
        self.assertIn("GPU", output)
        self.assertNotIn("{&quot;", output)
        self.assertNotIn("['GPU'", output)

    def test_detail_and_subpoints_stay_in_one_point_card(self):
        output = render_value({
            "detail": "生成模型学习数据分布。",
            "subpoints": ["显式密度模型", "隐式生成模型", "扩散模型"],
        })
        self.assertIn('class="point-detail"', output)
        self.assertIn('class="point-subpoints"', output)
        self.assertNotIn("<h4>Detail</h4>", output)
        self.assertNotIn("<h4>Subpoints</h4>", output)
        self.assertLess(output.index("生成模型学习数据分布"), output.index("显式密度模型"))

    def test_timeline_keeps_nonstandard_fields(self):
        output = render_section_content({"layout": "timeline", "content": {"nodes": [
            {"stage": "Early", "method": "Baseline", "problem": "High cost", "tradeoffs": ["Memory", "Latency"]},
            {"stage": "Later", "method": "Improved", "problem": "Low recall", "tradeoffs": ["Complexity"]},
        ]}})
        self.assertIn("High cost", output)
        self.assertIn("Memory", output)
        self.assertNotIn("{'", output)

    def test_comparison_does_not_duplicate_dimension_column(self):
        content = {
            "columns": ["方法维度", "Debevec & Malik (1997) - HDR重建", "Mertens et al. (2007) - 多曝光融合"],
            "rows": [
                {"name": "核心目标", "values": ["恢复场景物理辐射度图", "直接生成视觉质量高的显示图像"]},
                {"dimension": "是否需要CRF标定", "values": ["是", "否"]},
            ],
        }
        row_label, columns, rows = normalize_comparison(content)
        self.assertEqual(row_label, "方法维度")
        self.assertEqual(len(columns), 2)
        self.assertTrue(all(len(values) == 2 for _, values in rows))
        output = render_section_content({"layout": "comparison", "content": content})
        self.assertEqual(output.count("<th>方法维度</th>"), 1)
        self.assertIn("恢复场景物理辐射度图", output)

    def test_process_is_split_into_interactive_branches(self):
        content = {"steps": [
            {"name": "输入多曝光LDR图像序列", "description": "公共输入"},
            {"name": "HDR重建：辐射度标定", "description": "恢复CRF"},
            {"name": "HDR重建：权重融合", "description": "辐射域融合"},
            {"name": "多曝光融合：多尺度分解", "description": "拉普拉斯金字塔"},
            {"name": "多曝光融合：权重图生成与融合", "description": "质量权重"},
        ]}
        common, branches = normalize_process(content)
        self.assertEqual(len(common), 1)
        self.assertEqual([branch["name"] for branch in branches], ["HDR重建", "多曝光融合"])
        output = render_section_content({"layout": "process", "content": content})
        self.assertIn("flow-branches", output)
        self.assertIn("<details", output)
        self.assertIn("恢复CRF", output)

    def test_named_retrieved_paper_recovers_missing_source_id(self):
        section = {"title": "经典路线", "takeaway": "Debevec and Malik established the calibrated HDR pipeline.", "content": {"details": "The 1997 method estimates the camera response function."}, "source_ids": []}
        papers = [{"id": "paper-1", "title": "Recovering High Dynamic Range Radiance Maps from Photographs", "authors": ["Paul Debevec", "Jitendra Malik"], "published": "1997-01-01"}]
        self.assertEqual(infer_section_source_ids(section, papers), ["paper-1"])


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