Update test_pipeline.py

Author: BryceWDesign

synapdrive_ai/tests/test_pipeline.py

@@ -1,109 +1,106 @@
 from __future__ import annotations
 
-from typing import Any, Dict, Optional
-import random
-
-from synapdrive_ai.bci.signal_simulator import BrainSignalSimulator
-from synapdrive_ai.agi.core_reasoning import AGICoreReasoner
-from synapdrive_ai.agi.cognitive_optimizer import CognitiveOptimizer
-from synapdrive_ai.agi.meta_evaluator import MetaEvaluator
-from synapdrive_ai.action.decision_router import DecisionRouter
-from synapdrive_ai.memory.episodic_memory import EpisodicMemory
-from synapdrive_ai.safety.safety_guard import SafetyGuard
-from synapdrive_ai.vision.visual_inference import VisualInferenceEngine
+import pytest
+
 from synapdrive_ai.bci.intent_generator import generate_intent
+from synapdrive_ai.pipeline import SynapDrivePipeline
+
+
+# ---------------------------------------------------------------------------
+# Text command path
+# ---------------------------------------------------------------------------
+
+def test_text_command_known_intent_succeeds(pipeline: SynapDrivePipeline) -> None:
+    out = pipeline.run_text_command("move left", image_label="road")
+    assert out["status"] in {"success", "blocked"}
+    assert "intent" in out
+    assert "evaluation" in out
+
+
+def test_text_command_stop_succeeds(pipeline: SynapDrivePipeline) -> None:
+    out = pipeline.run_text_command("stop")
+    assert out["status"] in {"success", "blocked"}
+
+
+def test_text_command_unknown_input_is_blocked(pipeline: SynapDrivePipeline) -> None:
+    # Unknown inputs resolve to low confidence and should be blocked by safety.
+    out = pipeline.run_text_command("xyzzy nonsense")
+    assert out["status"] == "blocked"
+
+
+def test_text_command_returns_required_keys(pipeline: SynapDrivePipeline) -> None:
+    out = pipeline.run_text_command("move right")
+    for key in ("status", "intent", "result", "evaluation"):
+        assert key in out, f"Missing key: {key}"
+
+
+# ---------------------------------------------------------------------------
+# Signal path
+# ---------------------------------------------------------------------------
+
+def test_signal_event_known_labels(pipeline: SynapDrivePipeline) -> None:
+    for label in ("walk", "stop", "left_arm", "right_arm", "calculate", "recall", "explore"):
+        out = pipeline.run_signal_event(label=label)
+        assert out["status"] in {"success", "blocked"}
+
+
+def test_signal_event_random_label(pipeline: SynapDrivePipeline) -> None:
+    out = pipeline.run_signal_event()
+    assert out["status"] in {"success", "blocked"}
+
+
+def test_signal_event_unknown_label_raises(pipeline: SynapDrivePipeline) -> None:
+    with pytest.raises(ValueError, match="Unknown signal label"):
+        pipeline.run_signal_event(label="brain_blast")
+
+
+# ---------------------------------------------------------------------------
+# Safety gate
+# ---------------------------------------------------------------------------
+
+def test_blocked_result_has_reason(pipeline: SynapDrivePipeline) -> None:
+    out = pipeline.run_text_command("xyzzy")
+    assert out["status"] == "blocked"
+    assert "reason" in out
+
+
+def test_blocked_result_has_zero_evaluation_score(pipeline: SynapDrivePipeline) -> None:
+    out = pipeline.run_text_command("xyzzy")
+    assert out["status"] == "blocked"
+    assert out["evaluation"]["score"] == 0.0
+
+
+# ---------------------------------------------------------------------------
+# Action log
+# ---------------------------------------------------------------------------
+
+def test_action_log_grows(pipeline: SynapDrivePipeline) -> None:
+    pipeline.run_text_command("move left", image_label="road")
+    pipeline.run_signal_event(label="walk")
+    log = pipeline.get_action_log()
+    assert len(log) >= 1  # blocked intents do not reach actuation
+
+
+def test_action_log_entries_have_schema(pipeline: SynapDrivePipeline) -> None:
+    pipeline.run_text_command("move right")
+    for entry in pipeline.get_action_log():
+        for key in ("intent", "confidence", "status", "duration"):
+            assert key in entry
+
+
+# ---------------------------------------------------------------------------
+# run_intent_packet (public entrypoint used by replay + integrations)
+# ---------------------------------------------------------------------------
+
+def test_run_intent_packet_roundtrip(pipeline: SynapDrivePipeline) -> None:
+    packet = generate_intent("move left")
+    out = pipeline.run_intent_packet(packet, image_label="road")
+    assert out["status"] in {"success", "blocked"}
+    assert out["result"]["intent"] is not None
 
 
-class SynapDrivePipeline:
-    """
-    Canonical end-to-end simulation pipeline.
-
-    Determinism support:
-      simulate_delay=False disables actuation sleep so replay/tests are fast and repeatable.
-    """
-
-    def __init__(self, simulate_delay: bool = True) -> None:
-        self.simulator = BrainSignalSimulator()
-        self.reasoner = AGICoreReasoner()
-
-        self.memory = EpisodicMemory()
-        self.visual = VisualInferenceEngine()
-
-        self.optimizer = CognitiveOptimizer()
-        self.optimizer.memory = self.memory
-        self.optimizer.visual = self.visual
-
-        self.guard = SafetyGuard()
-        self.router = DecisionRouter(simulate_delay=simulate_delay)
-        self.evaluator = MetaEvaluator()
-
-    def run_text_command(self, command_text: str, image_label: Optional[str] = None) -> Dict[str, Any]:
-        intent_packet = generate_intent(command_text)
-        return self.run_intent_packet(intent_packet, image_label=image_label)
-
-    def run_signal_event(self, label: Optional[str] = None, image_label: Optional[str] = None) -> Dict[str, Any]:
-        label = label or random.choice(["left_arm", "right_arm", "walk", "stop", "calculate", "recall", "explore"])
-        signal = self._generate_signal_for_label(label)
-        intent_packet = self.reasoner.receive_signal(label, signal)
-        return self.run_intent_packet(intent_packet, image_label=image_label)
-
-    def run_intent_packet(self, intent_packet: Dict[str, Any], image_label: Optional[str] = None) -> Dict[str, Any]:
-        """
-        Public entrypoint for integrations + replay.
-        """
-        return self._run_common(intent_packet, image_label=image_label)
-
-    def _generate_signal_for_label(self, label: str):
-        patterns = {
-            "left_arm": 10,
-            "right_arm": 12,
-            "walk": 8,
-            "stop": 3,
-            "calculate": 25,
-            "recall": 18,
-            "explore": 30,
-        }
-        if label not in patterns:
-            raise ValueError(f"Unknown signal label: {label}")
-        return self.simulator.generate_waveform(patterns[label])
-
-    def _run_common(self, intent_packet: Dict[str, Any], image_label: Optional[str]) -> Dict[str, Any]:
-        optimized = self.optimizer.optimize(intent_packet, image_label=image_label)
-
-        is_safe, reason = self.guard.evaluate_safety(optimized)
-        if not is_safe:
-            return {
-                "status": "blocked",
-                "reason": reason,
-                "intent": optimized,
-                "result": {
-                    "status": "blocked",
-                    "intent": optimized.get("intent", "unknown"),
-                    "confidence": optimized.get("confidence", 0.0),
-                    "duration": 0.0,
-                },
-                "evaluation": {
-                    "score": 0.0,
-                    "total_actions": 0,
-                    "avg_score": 0.0,
-                },
-            }
-
-        result = self.router.route(optimized)
-
-        try:
-            self.memory.record_episode(optimized, result)
-        except Exception:
-            pass
-
-        evaluation = self.evaluator.evaluate(optimized, result)
-
-        return {
-            "status": result["status"],
-            "intent": optimized,
-            "result": result,
-            "evaluation": evaluation,
-        }
-
-    def get_action_log(self):
-        return self.router.get_action_log()
+def test_run_intent_packet_with_visual_context(pipeline: SynapDrivePipeline) -> None:
+    for image_label in ("road", "hazard", "person", "vehicle"):
+        packet = generate_intent("move forward")
+        out = pipeline.run_intent_packet(packet, image_label=image_label)
+        assert out["status"] in {"success", "blocked"}