Spaces:
Running on Zero
Running on Zero
| 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]) | |
| 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, []) | |
| 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() | |
| 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}) | |
| 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("{"", 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() | |