tests/test_ui_helpers.py
Rama main · réplica de lectura del repositorio original.
"""
Tests for the UI helper modules.
Acceptance:
- format_analysis_event handles all known event types
- validate_explainer_inputs checks the four mandatory fields
- build_explainer_preflight_data builds a consistent dict
- format_task_plan_markdown returns a readable preview for a TaskPlan
- the should_show_preview heuristics work correctly
- extract_plan_metadata_info returns complete metadata
- PreflightChecker validates context-sensitively
- USE_CASE_REGISTRY has all analysis use cases with mandatory fields
"""
import unittest
from src.pipeline.analysis_pipeline import (
PreflightChecker,
Requirement,
USE_CASE_REGISTRY,
)
from src.pipeline.models import (
ResearchPlan,
ResearchQuestion,
SubTask,
TaskPlan,
)
from src.ui.analysis_runner import (
build_explainer_preflight_data,
format_analysis_event,
validate_explainer_inputs,
)
from src.ui.components.plan_preview import (
extract_plan_metadata_info,
format_task_plan_markdown,
format_research_plan_markdown,
should_show_preview,
should_show_research_preview,
)
# ───────────────────────────────────────────────────────────────────
# format_analysis_event
# ───────────────────────────────────────────────────────────────────
class TestFormatAnalysisEvent(unittest.TestCase):
def test_empty_event_returns_empty(self):
self.assertEqual(format_analysis_event("", {}), "")
def test_node_start(self):
out = format_analysis_event("node_start", {"node": "analysis_layer_1"})
self.assertIn("Starting", out)
self.assertIn("analysis_layer_1", out)
def test_node_done_with_metadata(self):
out = format_analysis_event("node_done", {
"node": "layer_1",
"metadata": {"executed": 3, "done": 3, "failed": 0},
})
self.assertIn("✅", out)
self.assertIn("3/3 OK", out)
def test_node_done_with_failures(self):
out = format_analysis_event("node_done", {
"node": "layer_2",
"metadata": {"executed": 5, "done": 4, "failed": 1},
})
self.assertIn("4/5 OK", out)
self.assertIn("1 failed", out)
def test_node_skipped(self):
out = format_analysis_event("node_skipped", {
"node": "format_agent", "reason": "Schema schon da",
})
self.assertIn("⊘", out)
self.assertIn("format_agent", out)
self.assertIn("Schema schon da", out)
def test_node_failed_with_long_error_truncated(self):
long_error = "x" * 500
out = format_analysis_event("node_failed", {
"node": "harvest", "error": long_error,
})
self.assertIn("❌", out)
self.assertIn("…", out, "Lange Fehler werden mit Ellipse abgekürzt")
self.assertLess(len(out), 300)
def test_pipeline_done(self):
out = format_analysis_event("pipeline_done", {"nodes": 15})
self.assertIn("🏁", out)
self.assertIn("15", out)
def test_pipeline_stopped(self):
out = format_analysis_event("pipeline_stopped", {})
self.assertIn("⏹", out)
def test_plan_ready(self):
out = format_analysis_event("plan_ready", {"n_tasks": 8})
self.assertIn("📋", out)
self.assertIn("8 tasks", out)
def test_unknown_event_silent(self):
out = format_analysis_event("never_heard_of", {"x": 1})
self.assertEqual(out, "")
# ───────────────────────────────────────────────────────────────────
# validate_explainer_inputs / build_explainer_preflight_data
# ───────────────────────────────────────────────────────────────────
class TestExplainerInputs(unittest.TestCase):
"""The registry check must accept exactly what the explainer panel offers."""
def test_valid_inputs(self):
ok, msg = validate_explainer_inputs(
topic="How does quantum entanglement work?",
audience="Physics undergraduates without quantum mechanics",
length="short",
purpose="teaching",
)
self.assertTrue(ok, msg)
self.assertEqual(msg, "")
def test_missing_required_topic(self):
ok, msg = validate_explainer_inputs(
topic="", audience="Physics undergraduates, first year",
length="short", purpose="",
)
self.assertFalse(ok)
self.assertIn("Topic", msg)
def test_audience_too_short(self):
ok, msg = validate_explainer_inputs(
topic="How does quantum entanglement work?", audience="Aliens",
length="short", purpose="",
)
self.assertFalse(ok)
self.assertIn("Audience", msg)
def test_purpose_optional(self):
"""purpose is optional, an empty value is fine."""
ok, msg = validate_explainer_inputs(
topic="How does quantum entanglement work?",
audience="Physics undergraduates, first year",
length="short", purpose="",
)
self.assertTrue(ok, msg)
def test_every_offered_choice_validates(self):
"""Every length/purpose the UI dropdown offers passes the check."""
from src.pipeline.analysis_pipeline import (
EXPLAINER_LENGTHS, EXPLAINER_PURPOSES,
)
for _, length in EXPLAINER_LENGTHS:
for _, purpose in EXPLAINER_PURPOSES:
ok, msg = validate_explainer_inputs(
"How does quantum entanglement work?",
"Physics undergraduates, first year", length, purpose,
)
self.assertTrue(ok, msg)
def test_build_preflight_data_strips_values(self):
d = build_explainer_preflight_data(
topic=" Quantum mechanics ",
audience="Students",
length="short",
purpose=" teaching ",
)
self.assertEqual(d["topic"], "Quantum mechanics")
self.assertEqual(d["purpose"], "teaching")
def test_build_preflight_data_drops_empty_optional(self):
d = build_explainer_preflight_data(
topic="x", audience="Students",
length="short", purpose="",
)
self.assertNotIn("purpose", d,
"an empty optional field should not be in the dict")
self.assertIn("topic", d)
# ───────────────────────────────────────────────────────────────────
# format_task_plan_markdown / format_research_plan_markdown
# ───────────────────────────────────────────────────────────────────
class TestFormatPlanMarkdown(unittest.TestCase):
def test_none_plan(self):
out = format_task_plan_markdown(None)
self.assertIn("no ", out.lower())
def test_empty_tasks(self):
plan = TaskPlan(use_case="explainer", tasks=[])
out = format_task_plan_markdown(plan)
self.assertIn("explainer", out)
def test_full_plan_lists_tasks_grouped_by_phase(self):
plan = TaskPlan(use_case="explainer", tasks=[
SubTask(id="K1", phase="concept",
description="Konzept extrahieren",
prompt_template="x"),
SubTask(id="E1", phase="explanation",
description="Erklärung 1",
prompt_template="x", depends_on=["K1"]),
SubTask(id="E2", phase="explanation",
description="Erklärung 2",
prompt_template="x", depends_on=["K1"]),
])
out = format_task_plan_markdown(plan)
self.assertIn("explainer", out)
self.assertIn("3 tasks", out)
self.assertIn("2 phases", out)
# The task IDs are in it
self.assertIn("K1", out)
self.assertIn("E1", out)
# Dependency display
self.assertIn("← K1", out)
class TestFormatResearchPlanMarkdown(unittest.TestCase):
def test_delegates_to_plan_editor(self):
"""format_research_plan_markdown delegates to plan_editor."""
plan = ResearchPlan(
summary="Test-Plan-Zusammenfassung",
questions=[ResearchQuestion(
id="F1", question="Test?",
search_terms=["test"], priority="hoch",
)],
)
out = format_research_plan_markdown(plan)
self.assertIn("Research plan", out)
self.assertIn("F1", out)
self.assertIn("Test?", out)
# ───────────────────────────────────────────────────────────────────
# should_show_preview / should_show_research_preview
# ───────────────────────────────────────────────────────────────────
class TestShouldShowPreview(unittest.TestCase):
def test_none_plan_never_shown(self):
self.assertFalse(should_show_preview(None))
def test_force_overrides(self):
self.assertTrue(should_show_preview(None, force=True))
def test_empty_plan_not_shown(self):
plan = TaskPlan(use_case="x", tasks=[])
self.assertFalse(should_show_preview(plan))
def test_one_task_one_phase_not_shown(self):
"""Trivial plan — only 1 task, 1 phase: do not show."""
plan = TaskPlan(use_case="x", tasks=[
SubTask(id="T1", phase="p1", description="x",
prompt_template="x"),
])
self.assertFalse(should_show_preview(plan))
def test_three_or_more_tasks_shown(self):
plan = TaskPlan(use_case="x", tasks=[
SubTask(id=f"T{i}", phase="p1",
description="x", prompt_template="x")
for i in range(3)
])
self.assertTrue(should_show_preview(plan))
def test_two_phases_shown(self):
plan = TaskPlan(use_case="x", tasks=[
SubTask(id="T1", phase="p1", description="x",
prompt_template="x"),
SubTask(id="T2", phase="p2", description="x",
prompt_template="x"),
])
self.assertTrue(should_show_preview(plan))
class TestShouldShowResearchPreview(unittest.TestCase):
def test_none_not_shown(self):
self.assertFalse(should_show_research_preview(None))
def test_empty_plan_not_shown(self):
self.assertFalse(should_show_research_preview(ResearchPlan()))
def test_two_questions_shown(self):
plan = ResearchPlan(questions=[
ResearchQuestion(id="F1", question="?", search_terms=["x"]),
ResearchQuestion(id="F2", question="?", search_terms=["y"]),
])
self.assertTrue(should_show_research_preview(plan))
def test_one_question_alone_not_shown(self):
"""Only 1 question and nothing else: trivial, do not show."""
plan = ResearchPlan(questions=[
ResearchQuestion(id="F1", question="?", search_terms=["x"]),
])
self.assertFalse(should_show_research_preview(plan))
def test_one_question_plus_url_shown(self):
from src.pipeline.models import DirectURL
plan = ResearchPlan(
questions=[ResearchQuestion(
id="F1", question="?", search_terms=["x"],
)],
direct_urls=[DirectURL(url="https://x.com", reason="test")],
)
self.assertTrue(should_show_research_preview(plan))
# ───────────────────────────────────────────────────────────────────
# extract_plan_metadata_info
# ───────────────────────────────────────────────────────────────────
class TestExtractPlanMetadataInfo(unittest.TestCase):
def test_none_plan_empty_dict(self):
self.assertEqual(extract_plan_metadata_info(None), {})
def test_full_metadata(self):
plan = TaskPlan(use_case="explainer", tasks=[
SubTask(id="T1", phase="concept", description="x",
prompt_template="x"),
SubTask(id="T2", phase="explanation", description="x",
prompt_template="x", depends_on=["T1"]),
SubTask(id="T3", phase="explanation", description="x",
prompt_template="x", depends_on=["T1"]),
], estimated_calls=10, estimated_duration_seconds=120)
info = extract_plan_metadata_info(plan)
self.assertEqual(info["use_case"], "explainer")
self.assertEqual(info["n_tasks"], 3)
self.assertEqual(info["n_phases"], 2)
self.assertEqual(info["phases"], ["concept", "explanation"])
self.assertEqual(info["estimated_calls"], 10)
self.assertEqual(info["estimated_duration_seconds"], 120)
# ───────────────────────────────────────────────────────────────────
# USE_CASE_REGISTRY and PreflightChecker
# ───────────────────────────────────────────────────────────────────
class TestUseCaseRegistry(unittest.TestCase):
def test_all_seven_use_cases_registered(self):
expected = {
"explainer", "peer_review", "decision_analysis",
"grant_proposal", "literature_review", "research_design",
"literature_finder",
}
self.assertEqual(set(USE_CASE_REGISTRY.keys()), expected)
def test_each_uc_has_preflight_with_requirements(self):
for uc, cfg in USE_CASE_REGISTRY.items():
self.assertIn("preflight", cfg, f"use case {uc} without preflight")
checker = cfg["preflight"]
reqs = checker.get_requirements()
self.assertGreater(
len(reqs), 0,
f"use case {uc} has no mandatory fields",
)
def test_requirement_has_field_attribute(self):
"""gradio_app.py expects r.field for every requirement."""
for cfg in USE_CASE_REGISTRY.values():
for req in cfg["preflight"].get_requirements():
self.assertTrue(hasattr(req, "field"))
self.assertIsInstance(req.field, str)
self.assertTrue(req.field, "field must not be empty")
class TestPreflightChecker(unittest.TestCase):
def test_required_field_missing(self):
checker = PreflightChecker([
Requirement(field="x", label="X", required=True),
])
ok, errors = checker.validate({})
self.assertFalse(ok)
self.assertEqual(len(errors), 1)
def test_optional_field_missing_ok(self):
checker = PreflightChecker([
Requirement(field="x", label="X", required=False),
])
ok, errors = checker.validate({})
self.assertTrue(ok)
def test_choice_invalid_value(self):
checker = PreflightChecker([
Requirement(field="x", label="X", kind="choice",
choices=["a", "b"]),
])
ok, errors = checker.validate({"x": "c"})
self.assertFalse(ok)
self.assertTrue(any("allowed options" in e for e in errors))
def test_max_length_exceeded(self):
checker = PreflightChecker([
Requirement(field="x", label="X", max_length=5),
])
ok, errors = checker.validate({"x": "viel zu lang"})
self.assertFalse(ok)
self.assertTrue(any("characters" in e for e in errors))
def test_whitespace_only_treated_as_empty(self):
checker = PreflightChecker([
Requirement(field="x", label="X", required=True),
])
ok, errors = checker.validate({"x": " "})
self.assertFalse(ok)
if __name__ == "__main__":
unittest.main()