from omniff.graph.planner import GraphPlanner def test_plan_text_simple(): planner = GraphPlanner() graph = planner.plan("TEXT_SIMPLE") assert len(graph.nodes) == 4 assert len(graph.edges) == 3 assert graph.nodes[0].node_type == "demuxer" assert graph.nodes[-1].node_type == "muxer" def test_plan_image_caption(): planner = GraphPlanner() graph = planner.plan("IMAGE_CAPTION") node_types = [n.node_type for n in graph.nodes] assert "vlm_infer" in node_types def test_plan_audio_transcribe(): planner = GraphPlanner() graph = planner.plan("AUDIO_TRANSCRIBE_ONLY") node_types = [n.node_type for n in graph.nodes] assert "asr_infer" in node_types def test_plan_with_controls(): planner = GraphPlanner() graph = planner.plan("TEXT_SIMPLE", controls={"model_id": "test/model"}) for node in graph.nodes: assert "model_id" in node.config def test_plan_fallback_unknown_route(): planner = GraphPlanner() graph = planner.plan("UNKNOWN_ROUTE") assert len(graph.nodes) == 4 def test_plan_text_normal_uses_text_simple_template(): planner = GraphPlanner() graph = planner.plan("TEXT_NORMAL") node_types = [n.node_type for n in graph.nodes] assert "llm_infer" in node_types def test_available_routes(): planner = GraphPlanner() routes = planner.available_routes() assert "TEXT_SIMPLE" in routes assert "IMAGE_CAPTION" in routes assert len(routes) >= 7 def test_plan_edges_are_sequential(): planner = GraphPlanner() graph = planner.plan("TEXT_SIMPLE") for i, edge in enumerate(graph.edges): assert edge.source == graph.nodes[i].id assert edge.target == graph.nodes[i + 1].id