[kimi] Add /api/matrix/health endpoint (#685) (#745)

tests/dashboard/test_world_api.py

@@ -1674,3 +1674,197 @@ class TestMatrixMemorySearchEndpoint:
 
         assert resp.status_code == 200
         mock_search.assert_called_once_with("bitcoin sovereignty", limit=5)
+
+
+# ---------------------------------------------------------------------------
+# Matrix Health Endpoint (/api/matrix/health)
+# ---------------------------------------------------------------------------
+
+
+class TestMatrixHealthEndpoint:
+    """Tests for the Matrix health endpoint."""
+
+    def test_health_endpoint_returns_json(self, matrix_client):
+        """GET /api/matrix/health returns JSON response."""
+        resp = matrix_client.get("/api/matrix/health")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert isinstance(data, dict)
+        assert "status" in data
+        assert "version" in data
+        assert "capabilities" in data
+        assert resp.headers["cache-control"] == "no-cache, no-store"
+
+    def test_health_endpoint_status_field(self, matrix_client):
+        """Response contains status field with valid values."""
+        resp = matrix_client.get("/api/matrix/health")
+        data = resp.json()
+
+        assert data["status"] in ["ok", "degraded"]
+
+    def test_health_endpoint_version_field(self, matrix_client):
+        """Response contains version string."""
+        resp = matrix_client.get("/api/matrix/health")
+        data = resp.json()
+
+        assert isinstance(data["version"], str)
+        assert len(data["version"]) > 0
+
+    def test_health_endpoint_capabilities_structure(self, matrix_client):
+        """Capabilities dict has all required features."""
+        resp = matrix_client.get("/api/matrix/health")
+        data = resp.json()
+
+        caps = data["capabilities"]
+        assert isinstance(caps, dict)
+
+        # All required capabilities
+        required_caps = ["thinking", "memory", "bark", "familiar", "lightning"]
+        for cap in required_caps:
+            assert cap in caps, f"Missing capability: {cap}"
+            assert isinstance(caps[cap], bool), f"Capability {cap} should be bool"
+
+    def test_health_endpoint_returns_200_when_degraded(self, matrix_client):
+        """Returns 200 even when some capabilities are unavailable."""
+        # Mock all capability checks to return False
+        with patch.multiple(
+            "dashboard.routes.world",
+            _check_capability_thinking=lambda: False,
+            _check_capability_memory=lambda: False,
+            _check_capability_bark=lambda: False,
+            _check_capability_familiar=lambda: False,
+            _check_capability_lightning=lambda: False,
+        ):
+            resp = matrix_client.get("/api/matrix/health")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["status"] == "degraded"
+
+    def test_health_endpoint_ok_when_core_caps_available(self, matrix_client):
+        """Status is 'ok' when core capabilities are available."""
+        with patch.multiple(
+            "dashboard.routes.world",
+            _check_capability_thinking=lambda: True,
+            _check_capability_memory=lambda: True,
+            _check_capability_bark=lambda: True,
+            _check_capability_familiar=lambda: False,
+            _check_capability_lightning=lambda: False,
+        ):
+            resp = matrix_client.get("/api/matrix/health")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["status"] == "ok"
+        assert data["capabilities"]["thinking"] is True
+        assert data["capabilities"]["memory"] is True
+        assert data["capabilities"]["bark"] is True
+        assert data["capabilities"]["familiar"] is False
+        assert data["capabilities"]["lightning"] is False
+
+
+class TestMatrixHealthCapabilityChecks:
+    """Tests for individual capability check functions."""
+
+    def test_check_thinking_returns_true_when_available(self):
+        """_check_capability_thinking returns True when thinking engine is available."""
+        from dashboard.routes.world import _check_capability_thinking
+
+        # Mock thinking engine with _db attribute
+        mock_engine = MagicMock()
+        mock_engine._db = "/path/to/thoughts.db"
+
+        with patch("timmy.thinking.thinking_engine", mock_engine):
+            result = _check_capability_thinking()
+
+        assert result is True
+
+    def test_check_thinking_returns_false_when_no_db(self):
+        """_check_capability_thinking returns False when _db is None."""
+        from dashboard.routes.world import _check_capability_thinking
+
+        mock_engine = MagicMock()
+        mock_engine._db = None
+
+        with patch("timmy.thinking.thinking_engine", mock_engine):
+            result = _check_capability_thinking()
+
+        assert result is False
+
+    def test_check_memory_returns_true_when_hot_memory_exists(self, tmp_path):
+        """_check_capability_memory returns True when HOT_MEMORY_PATH exists."""
+        from dashboard.routes.world import _check_capability_memory
+
+        # Create a temporary memory path
+        mock_path = MagicMock()
+        mock_path.exists.return_value = True
+
+        with patch("timmy.memory_system.HOT_MEMORY_PATH", mock_path):
+            result = _check_capability_memory()
+
+        assert result is True
+
+    def test_check_memory_returns_false_when_not_exists(self):
+        """_check_capability_memory returns False when HOT_MEMORY_PATH doesn't exist."""
+        from dashboard.routes.world import _check_capability_memory
+
+        mock_path = MagicMock()
+        mock_path.exists.return_value = False
+
+        with patch("timmy.memory_system.HOT_MEMORY_PATH", mock_path):
+            result = _check_capability_memory()
+
+        assert result is False
+
+    def test_check_bark_returns_true_when_produce_bark_available(self):
+        """_check_capability_bark returns True when produce_bark is callable."""
+        from dashboard.routes.world import _check_capability_bark
+
+        with patch(
+            "infrastructure.presence.produce_bark",
+            return_value={"type": "bark"},
+        ):
+            result = _check_capability_bark()
+
+        assert result is True
+
+    def test_check_familiar_returns_true_when_pip_available(self):
+        """_check_capability_familiar returns True when pip_familiar is available."""
+        from dashboard.routes.world import _check_capability_familiar
+
+        mock_pip = MagicMock()
+
+        with patch("timmy.familiar.pip_familiar", mock_pip):
+            result = _check_capability_familiar()
+
+        assert result is True
+
+    def test_check_familiar_returns_false_when_pip_none(self):
+        """_check_capability_familiar returns False when pip_familiar is None."""
+        from dashboard.routes.world import _check_capability_familiar
+
+        with patch("timmy.familiar.pip_familiar", None):
+            result = _check_capability_familiar()
+
+        assert result is False
+
+    def test_check_lightning_returns_false(self):
+        """_check_capability_lightning returns False (disabled per health.py)."""
+        from dashboard.routes.world import _check_capability_lightning
+
+        result = _check_capability_lightning()
+
+        assert result is False
+
+    def test_health_response_time_is_fast(self, matrix_client):
+        """Health endpoint responds quickly (<100ms for lightweight checks)."""
+        import time
+
+        start = time.time()
+        resp = matrix_client.get("/api/matrix/health")
+        elapsed = time.time() - start
+
+        assert resp.status_code == 200
+        # Should be very fast since checks are lightweight
+        assert elapsed < 1.0  # Generous timeout for test environments