Spaces:
Running
Running
| """Tests for plugins API routes.""" | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| class TestPluginsAPI: | |
| """Test plugins API endpoints.""" | |
| def test_list_all_plugins(self, client: TestClient) -> None: | |
| """Test GET /api/plugins returns all plugins.""" | |
| response = client.get("/api/plugins") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| # Check response structure | |
| assert "plugins" in data | |
| assert "categories" in data | |
| assert "stats" in data | |
| # Check stats structure | |
| stats = data["stats"] | |
| assert "total" in stats | |
| assert "installed" in stats | |
| assert "available" in stats | |
| assert stats["total"] >= 0 | |
| assert stats["installed"] >= 0 | |
| assert stats["available"] >= 0 | |
| def test_list_plugins_by_category(self, client: TestClient) -> None: | |
| """Test GET /api/plugins?category=apis filters by category.""" | |
| response = client.get("/api/plugins?category=apis") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| # Should only contain the filtered category | |
| plugins = data["plugins"] | |
| if "apis" in plugins: | |
| # All plugins in the response should be from apis category | |
| for plugin in plugins["apis"]: | |
| assert plugin["category"] == "apis" | |
| def test_list_installed_plugins(self, client: TestClient) -> None: | |
| """Test GET /api/plugins/installed returns only installed plugins.""" | |
| response = client.get("/api/plugins/installed") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "plugins" in data | |
| assert "count" in data | |
| # All returned plugins should be installed | |
| for plugin in data["plugins"]: | |
| assert plugin["installed"] is True | |
| # Count should match number of plugins | |
| assert data["count"] == len(data["plugins"]) | |
| def test_get_specific_plugin_exists(self, client: TestClient) -> None: | |
| """Test GET /api/plugins/{plugin_id} for existing plugin.""" | |
| # First get list of plugins to find a valid ID | |
| list_response = client.get("/api/plugins") | |
| assert list_response.status_code == 200 | |
| plugins_data = list_response.json() | |
| # Find first plugin from any category | |
| plugin_id = None | |
| for category, plugins in plugins_data["plugins"].items(): | |
| if plugins: | |
| plugin_id = plugins[0]["id"] | |
| break | |
| if plugin_id: | |
| response = client.get(f"/api/plugins/{plugin_id}") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["id"] == plugin_id | |
| assert "name" in data | |
| assert "category" in data | |
| assert "installed" in data | |
| def test_get_nonexistent_plugin(self, client: TestClient) -> None: | |
| """Test GET /api/plugins/{plugin_id} for non-existent plugin.""" | |
| response = client.get("/api/plugins/nonexistent-plugin") | |
| assert response.status_code == 404 | |
| data = response.json() | |
| assert "not found" in data["detail"].lower() | |
| def test_install_plugin_success(self, client: TestClient) -> None: | |
| """Test POST /api/plugins/install for valid plugin.""" | |
| # First get a plugin that's not installed | |
| list_response = client.get("/api/plugins") | |
| assert list_response.status_code == 200 | |
| plugins_data = list_response.json() | |
| # Find an uninstalled plugin | |
| plugin_id = None | |
| for category, plugins in plugins_data["plugins"].items(): | |
| for plugin in plugins: | |
| if not plugin["installed"]: | |
| plugin_id = plugin["id"] | |
| break | |
| if plugin_id: | |
| break | |
| if plugin_id: | |
| payload = {"plugin_id": plugin_id} | |
| response = client.post("/api/plugins/install", json=payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] == "success" | |
| assert data["plugin"]["id"] == plugin_id | |
| assert data["plugin"]["installed"] is True | |
| assert "installed successfully" in data["message"] | |
| def test_install_already_installed_plugin(self, client: TestClient) -> None: | |
| """Test installing a plugin that's already installed.""" | |
| # First install a plugin | |
| list_response = client.get("/api/plugins") | |
| assert list_response.status_code == 200 | |
| plugins_data = list_response.json() | |
| # Find an uninstalled plugin to install first | |
| plugin_id = None | |
| for category, plugins in plugins_data["plugins"].items(): | |
| for plugin in plugins: | |
| if not plugin["installed"]: | |
| plugin_id = plugin["id"] | |
| break | |
| if plugin_id: | |
| break | |
| if plugin_id: | |
| # Install it | |
| payload = {"plugin_id": plugin_id} | |
| response = client.post("/api/plugins/install", json=payload) | |
| assert response.status_code == 200 | |
| # Try to install again | |
| response = client.post("/api/plugins/install", json=payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] == "already_installed" | |
| assert "already installed" in data["message"] | |
| def test_install_nonexistent_plugin(self, client: TestClient) -> None: | |
| """Test installing a non-existent plugin.""" | |
| payload = {"plugin_id": "nonexistent-plugin"} | |
| response = client.post("/api/plugins/install", json=payload) | |
| assert response.status_code == 404 | |
| data = response.json() | |
| assert "not found" in data["detail"].lower() | |
| def test_uninstall_plugin_success(self, client: TestClient) -> None: | |
| """Test POST /api/plugins/uninstall for installed non-core plugin.""" | |
| # First install a non-core plugin | |
| list_response = client.get("/api/plugins") | |
| assert list_response.status_code == 200 | |
| plugins_data = list_response.json() | |
| # Find a non-core plugin to install and then uninstall | |
| core_plugins = { | |
| "mcp-browser", | |
| "mcp-search", | |
| "mcp-html", | |
| "mcp-python-sandbox", | |
| "proc-json", | |
| "proc-python", | |
| "proc-pandas", | |
| "proc-numpy", | |
| "proc-bs4", | |
| } | |
| plugin_id = None | |
| for category, plugins in plugins_data["plugins"].items(): | |
| for plugin in plugins: | |
| if plugin["id"] not in core_plugins and not plugin["installed"]: | |
| plugin_id = plugin["id"] | |
| break | |
| if plugin_id: | |
| break | |
| if plugin_id: | |
| # Install it first | |
| install_payload = {"plugin_id": plugin_id} | |
| install_response = client.post("/api/plugins/install", json=install_payload) | |
| assert install_response.status_code == 200 | |
| # Now uninstall it | |
| uninstall_payload = {"plugin_id": plugin_id} | |
| response = client.post("/api/plugins/uninstall", json=uninstall_payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] == "success" | |
| assert data["plugin"]["id"] == plugin_id | |
| assert data["plugin"]["installed"] is False | |
| assert "uninstalled successfully" in data["message"] | |
| def test_uninstall_core_plugin_forbidden(self, client: TestClient) -> None: | |
| """Test uninstalling core plugin is forbidden.""" | |
| # Try to uninstall a core plugin | |
| core_plugin_id = "mcp-browser" # This should be a core plugin | |
| payload = {"plugin_id": core_plugin_id} | |
| response = client.post("/api/plugins/uninstall", json=payload) | |
| assert response.status_code == 400 | |
| data = response.json() | |
| assert "Cannot uninstall core plugin" in data["detail"] | |
| def test_uninstall_not_installed_plugin(self, client: TestClient) -> None: | |
| """Test uninstalling a plugin that's not installed.""" | |
| # Find an uninstalled non-core plugin | |
| list_response = client.get("/api/plugins") | |
| assert list_response.status_code == 200 | |
| plugins_data = list_response.json() | |
| core_plugins = { | |
| "mcp-browser", | |
| "mcp-search", | |
| "mcp-html", | |
| "mcp-python-sandbox", | |
| "proc-json", | |
| "proc-python", | |
| "proc-pandas", | |
| "proc-numpy", | |
| "proc-bs4", | |
| } | |
| plugin_id = None | |
| for category, plugins in plugins_data["plugins"].items(): | |
| for plugin in plugins: | |
| if plugin["id"] not in core_plugins and not plugin["installed"]: | |
| plugin_id = plugin["id"] | |
| break | |
| if plugin_id: | |
| break | |
| if plugin_id: | |
| payload = {"plugin_id": plugin_id} | |
| response = client.post("/api/plugins/uninstall", json=payload) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["status"] == "not_installed" | |
| assert "not installed" in data["message"] | |
| def test_uninstall_nonexistent_plugin(self, client: TestClient) -> None: | |
| """Test uninstalling a non-existent plugin.""" | |
| payload = {"plugin_id": "nonexistent-plugin"} | |
| response = client.post("/api/plugins/uninstall", json=payload) | |
| assert response.status_code == 404 | |
| data = response.json() | |
| assert "not found" in data["detail"].lower() | |
| def test_get_categories(self, client: TestClient) -> None: | |
| """Test that plugins list includes categories.""" | |
| response = client.get("/api/plugins") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert "categories" in data | |
| categories = data["categories"] | |
| assert isinstance(categories, list) | |
| assert len(categories) > 0 | |
| # Categories are returned as strings (category IDs) | |
| expected_categories = ["apis", "mcps", "processors"] | |
| for expected in expected_categories: | |
| assert expected in categories | |
| # Agents/skills are intentionally managed via /api/agents, not /api/plugins | |
| assert "skills" not in categories | |
| def test_plugin_structure_validation(self, client: TestClient) -> None: | |
| """Test that all plugins have required fields.""" | |
| response = client.get("/api/plugins") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| required_fields = ["id", "name", "category", "description", "version", "installed"] | |
| for category, plugins in data["plugins"].items(): | |
| for plugin in plugins: | |
| for field in required_fields: | |
| assert field in plugin, ( | |
| f"Plugin {plugin.get('id', 'unknown')} missing field {field}" | |
| ) | |
| def test_install_uninstall_payload_validation(self, client: TestClient) -> None: | |
| """Test payload validation for install/uninstall endpoints.""" | |
| # Missing plugin_id for install | |
| response = client.post("/api/plugins/install", json={}) | |
| assert response.status_code == 422 | |
| # Missing plugin_id for uninstall | |
| response = client.post("/api/plugins/uninstall", json={}) | |
| assert response.status_code == 422 | |
| # Invalid payload type | |
| response = client.post("/api/plugins/install", json={"plugin_id": 123}) | |
| assert response.status_code == 422 | |
| def test_plugins_state_persistence(self, client: TestClient) -> None: | |
| """Test that plugin installation state persists across calls.""" | |
| # Find a non-core plugin | |
| list_response = client.get("/api/plugins") | |
| assert list_response.status_code == 200 | |
| plugins_data = list_response.json() | |
| core_plugins = { | |
| "mcp-browser", | |
| "mcp-search", | |
| "mcp-html", | |
| "mcp-python-sandbox", | |
| "proc-json", | |
| "proc-python", | |
| "proc-pandas", | |
| "proc-numpy", | |
| "proc-bs4", | |
| } | |
| plugin_id = None | |
| for category, plugins in plugins_data["plugins"].items(): | |
| for plugin in plugins: | |
| if plugin["id"] not in core_plugins: | |
| plugin_id = plugin["id"] | |
| break | |
| if plugin_id: | |
| break | |
| if plugin_id: | |
| # Check initial state | |
| response = client.get(f"/api/plugins/{plugin_id}") | |
| initial_state = response.json()["installed"] | |
| # Toggle state by installing if not installed, or uninstalling if installed and not core | |
| if not initial_state: | |
| payload = {"plugin_id": plugin_id} | |
| response = client.post("/api/plugins/install", json=payload) | |
| assert response.status_code == 200 | |
| # Verify state changed | |
| response = client.get(f"/api/plugins/{plugin_id}") | |
| assert response.json()["installed"] is True | |
| else: | |
| # If already installed and not core, uninstall | |
| if plugin_id not in core_plugins: | |
| payload = {"plugin_id": plugin_id} | |
| response = client.post("/api/plugins/uninstall", json=payload) | |
| assert response.status_code == 200 | |
| # Verify state changed | |
| response = client.get(f"/api/plugins/{plugin_id}") | |
| assert response.json()["installed"] is False | |