c73e39f6e4
- Fix Python formatting (sync no longer strips blank lines from .py files) - Remove test_community_build.py (tests private sync module) - Fix ruff warnings in test files - Add CODE_OF_CONDUCT.md - Add .github/PULL_REQUEST_TEMPLATE.md Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
554 lines
21 KiB
Python
554 lines
21 KiB
Python
"""Tests for WordPress Plugin (plugins/wordpress/).
|
|
|
|
Integration tests covering plugin initialization, configuration validation,
|
|
tool specifications, handler delegation, client behavior, and health checks.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from plugins.base import BasePlugin, PluginRegistry
|
|
from plugins.wordpress.client import AuthenticationError, ConfigurationError, WordPressClient
|
|
from plugins.wordpress.plugin import WordPressPlugin
|
|
|
|
# --- WordPressClient Tests ---
|
|
|
|
|
|
class TestWordPressClientInit:
|
|
"""Test WordPressClient initialization and validation."""
|
|
|
|
def test_valid_initialization(self):
|
|
"""Should initialize with valid credentials."""
|
|
client = WordPressClient(
|
|
site_url="https://example.com",
|
|
username="admin",
|
|
app_password="xxxx xxxx xxxx xxxx",
|
|
)
|
|
assert client.site_url == "https://example.com"
|
|
assert client.api_base == "https://example.com/wp-json/wp/v2"
|
|
assert client.wc_api_base == "https://example.com/wp-json/wc/v3"
|
|
assert client.username == "admin"
|
|
|
|
def test_trailing_slash_stripped(self):
|
|
"""Should strip trailing slash from site URL."""
|
|
client = WordPressClient(
|
|
site_url="https://example.com/",
|
|
username="admin",
|
|
app_password="xxxx",
|
|
)
|
|
assert client.site_url == "https://example.com"
|
|
assert client.api_base == "https://example.com/wp-json/wp/v2"
|
|
|
|
def test_auth_header_created(self):
|
|
"""Should create proper Basic auth header."""
|
|
client = WordPressClient(
|
|
site_url="https://example.com",
|
|
username="admin",
|
|
app_password="secret123",
|
|
)
|
|
expected_token = base64.b64encode(b"admin:secret123").decode()
|
|
assert client.auth_header == f"Basic {expected_token}"
|
|
|
|
def test_missing_url_raises(self):
|
|
"""Should raise ConfigurationError for empty URL."""
|
|
with pytest.raises(ConfigurationError, match="Site URL is not configured"):
|
|
WordPressClient(site_url="", username="admin", app_password="xxxx")
|
|
|
|
def test_missing_username_raises(self):
|
|
"""Should raise ConfigurationError for empty username."""
|
|
with pytest.raises(ConfigurationError, match="Username is not configured"):
|
|
WordPressClient(site_url="https://example.com", username="", app_password="xxxx")
|
|
|
|
def test_missing_password_raises(self):
|
|
"""Should raise ConfigurationError for empty app password."""
|
|
with pytest.raises(ConfigurationError, match="App password is not configured"):
|
|
WordPressClient(site_url="https://example.com", username="admin", app_password="")
|
|
|
|
def test_none_url_raises(self):
|
|
"""Should raise ConfigurationError for None URL."""
|
|
with pytest.raises(ConfigurationError):
|
|
WordPressClient(site_url=None, username="admin", app_password="xxxx")
|
|
|
|
|
|
class TestWordPressClientErrorParsing:
|
|
"""Test error response parsing."""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
return WordPressClient(
|
|
site_url="https://example.com",
|
|
username="admin",
|
|
app_password="xxxx",
|
|
)
|
|
|
|
def test_parse_json_error(self, client):
|
|
"""Should parse JSON error responses."""
|
|
error_text = json.dumps(
|
|
{"code": "rest_forbidden", "message": "Sorry, you are not allowed."}
|
|
)
|
|
result = client._parse_error_response(403, error_text)
|
|
assert result["error_code"] == "ACCESS_DENIED"
|
|
assert result["status_code"] == 403
|
|
assert result["wp_error_code"] == "rest_forbidden"
|
|
|
|
def test_parse_non_json_error(self, client):
|
|
"""Should handle non-JSON error responses."""
|
|
result = client._parse_error_response(500, "Internal Server Error")
|
|
assert result["error_code"] == "SERVER_ERROR"
|
|
assert result["wp_error_code"] == "unknown_error"
|
|
|
|
def test_parse_auth_error(self, client):
|
|
"""Should provide auth-specific error messages."""
|
|
error_text = json.dumps({"code": "invalid_auth", "message": "Bad credentials"})
|
|
result = client._parse_error_response(401, error_text)
|
|
assert result["error_code"] == "AUTH_FAILED"
|
|
assert "Authentication failed" in result["message"]
|
|
assert "Application Password" in result["message"]
|
|
|
|
def test_parse_woocommerce_auth_error(self, client):
|
|
"""Should provide WooCommerce-specific auth error."""
|
|
error_text = json.dumps({"code": "wc_auth", "message": "No permission"})
|
|
result = client._parse_error_response(401, error_text, use_woocommerce=True)
|
|
assert "WooCommerce" in result["message"]
|
|
assert "manage_woocommerce" in result["message"]
|
|
|
|
def test_parse_404_error(self, client):
|
|
"""Should parse 404 errors correctly."""
|
|
error_text = json.dumps({"code": "rest_no_route", "message": "No route found"})
|
|
result = client._parse_error_response(404, error_text)
|
|
assert result["error_code"] == "NOT_FOUND"
|
|
assert "not found" in result["message"].lower()
|
|
|
|
def test_parse_400_error_with_hints(self, client):
|
|
"""Should provide parameter hints for 400 errors."""
|
|
error_text = json.dumps({"code": "rest_invalid_param", "message": "Invalid param"})
|
|
result = client._parse_error_response(400, error_text)
|
|
assert result["error_code"] == "BAD_REQUEST"
|
|
assert "Hints" in result["message"]
|
|
|
|
def test_raw_response_truncated(self, client):
|
|
"""Should truncate raw response to 500 chars."""
|
|
error_text = "x" * 1000
|
|
result = client._parse_error_response(500, error_text)
|
|
assert len(result["raw_response"]) == 500
|
|
|
|
def test_unknown_status_code(self, client):
|
|
"""Should handle unknown HTTP status codes."""
|
|
result = client._parse_error_response(418, "I'm a teapot")
|
|
assert result["error_code"] == "HTTP_418"
|
|
|
|
|
|
class TestWordPressClientRequest:
|
|
"""Test HTTP request methods."""
|
|
|
|
@pytest.fixture
|
|
def client(self):
|
|
return WordPressClient(
|
|
site_url="https://example.com",
|
|
username="admin",
|
|
app_password="xxxx",
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_filters_none_params(self, client):
|
|
"""Should filter None and empty values from params."""
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 200
|
|
mock_response.json = AsyncMock(return_value={"id": 1})
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.request = MagicMock(
|
|
return_value=AsyncMock(
|
|
__aenter__=AsyncMock(return_value=mock_response),
|
|
__aexit__=AsyncMock(return_value=False),
|
|
)
|
|
)
|
|
|
|
with patch("aiohttp.ClientSession") as mock_cls:
|
|
mock_cls.return_value = AsyncMock(
|
|
__aenter__=AsyncMock(return_value=mock_session),
|
|
__aexit__=AsyncMock(return_value=False),
|
|
)
|
|
|
|
result = await client.request(
|
|
"GET",
|
|
"posts",
|
|
params={"status": "publish", "search": None, "tags": "", "ids": []},
|
|
)
|
|
assert result == {"id": 1}
|
|
|
|
# Verify filtered params
|
|
call_kwargs = mock_session.request.call_args
|
|
filtered_params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
|
|
if filtered_params:
|
|
assert "search" not in filtered_params
|
|
assert "tags" not in filtered_params
|
|
assert "ids" not in filtered_params
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_raises_on_401(self, client):
|
|
"""Should raise AuthenticationError on 401."""
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 401
|
|
mock_response.text = AsyncMock(
|
|
return_value=json.dumps({"code": "invalid_auth", "message": "Bad creds"})
|
|
)
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.request = MagicMock(
|
|
return_value=AsyncMock(
|
|
__aenter__=AsyncMock(return_value=mock_response),
|
|
__aexit__=AsyncMock(return_value=False),
|
|
)
|
|
)
|
|
|
|
with patch("aiohttp.ClientSession") as mock_cls:
|
|
mock_cls.return_value = AsyncMock(
|
|
__aenter__=AsyncMock(return_value=mock_session),
|
|
__aexit__=AsyncMock(return_value=False),
|
|
)
|
|
|
|
with pytest.raises(AuthenticationError):
|
|
await client.request("GET", "posts")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_request_raises_on_500(self, client):
|
|
"""Should raise Exception on 500."""
|
|
mock_response = AsyncMock()
|
|
mock_response.status = 500
|
|
mock_response.text = AsyncMock(return_value="Internal Server Error")
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.request = MagicMock(
|
|
return_value=AsyncMock(
|
|
__aenter__=AsyncMock(return_value=mock_response),
|
|
__aexit__=AsyncMock(return_value=False),
|
|
)
|
|
)
|
|
|
|
with patch("aiohttp.ClientSession") as mock_cls:
|
|
mock_cls.return_value = AsyncMock(
|
|
__aenter__=AsyncMock(return_value=mock_session),
|
|
__aexit__=AsyncMock(return_value=False),
|
|
)
|
|
|
|
with pytest.raises(Exception, match="SERVER_ERROR"):
|
|
await client.request("GET", "posts")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_convenience_method(self, client):
|
|
"""GET method should delegate to request."""
|
|
client.request = AsyncMock(return_value={"posts": []})
|
|
result = await client.get("posts", params={"per_page": 10})
|
|
client.request.assert_called_once_with(
|
|
"GET",
|
|
"posts",
|
|
params={"per_page": 10},
|
|
use_custom_namespace=False,
|
|
use_woocommerce=False,
|
|
)
|
|
assert result == {"posts": []}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_convenience_method(self, client):
|
|
"""POST method should delegate to request."""
|
|
client.request = AsyncMock(return_value={"id": 1})
|
|
result = await client.post("posts", json_data={"title": "Test"})
|
|
client.request.assert_called_once()
|
|
assert result == {"id": 1}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_convenience_method(self, client):
|
|
"""DELETE method should delegate to request."""
|
|
client.request = AsyncMock(return_value={"deleted": True})
|
|
result = await client.delete("posts/1", params={"force": True})
|
|
client.request.assert_called_once()
|
|
assert result == {"deleted": True}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_woocommerce_url(self, client):
|
|
"""WooCommerce requests should use wc/v3 base."""
|
|
client.request = AsyncMock(return_value={"available": True})
|
|
await client.get("products", use_woocommerce=True)
|
|
client.request.assert_called_once_with(
|
|
"GET",
|
|
"products",
|
|
params=None,
|
|
use_custom_namespace=False,
|
|
use_woocommerce=True,
|
|
)
|
|
|
|
|
|
# --- WordPressPlugin Tests ---
|
|
|
|
|
|
class TestWordPressPluginInit:
|
|
"""Test WordPress plugin initialization."""
|
|
|
|
VALID_CONFIG = {
|
|
"url": "https://example.com",
|
|
"username": "admin",
|
|
"app_password": "xxxx xxxx xxxx xxxx",
|
|
}
|
|
|
|
def test_create_with_valid_config(self):
|
|
"""Should initialize with all required config keys."""
|
|
plugin = WordPressPlugin(self.VALID_CONFIG)
|
|
assert plugin.project_id is not None
|
|
assert plugin.client is not None
|
|
assert isinstance(plugin, BasePlugin)
|
|
|
|
def test_handlers_initialized(self):
|
|
"""Should initialize all core handlers."""
|
|
plugin = WordPressPlugin(self.VALID_CONFIG)
|
|
assert plugin.posts is not None
|
|
assert plugin.media is not None
|
|
assert plugin.taxonomy is not None
|
|
assert plugin.comments is not None
|
|
assert plugin.users is not None
|
|
assert plugin.site is not None
|
|
assert plugin.seo is not None
|
|
assert plugin.menus is not None
|
|
|
|
def test_wp_cli_none_without_container(self):
|
|
"""WP-CLI handler should be None without container config."""
|
|
plugin = WordPressPlugin(self.VALID_CONFIG)
|
|
assert plugin.wp_cli is None
|
|
|
|
def test_missing_url_raises(self):
|
|
"""Should raise ValueError for missing URL."""
|
|
config = {"username": "admin", "app_password": "xxxx"}
|
|
with pytest.raises(ValueError, match="Missing required configuration"):
|
|
WordPressPlugin(config)
|
|
|
|
def test_missing_username_raises(self):
|
|
"""Should raise ValueError for missing username."""
|
|
config = {"url": "https://example.com", "app_password": "xxxx"}
|
|
with pytest.raises(ValueError, match="Missing required configuration"):
|
|
WordPressPlugin(config)
|
|
|
|
def test_missing_password_raises(self):
|
|
"""Should raise ValueError for missing app_password."""
|
|
config = {"url": "https://example.com", "username": "admin"}
|
|
with pytest.raises(ValueError, match="Missing required configuration"):
|
|
WordPressPlugin(config)
|
|
|
|
def test_custom_project_id(self):
|
|
"""Should accept custom project_id."""
|
|
plugin = WordPressPlugin(self.VALID_CONFIG, project_id="wp_myblog")
|
|
assert plugin.project_id == "wp_myblog"
|
|
|
|
def test_auto_generated_project_id(self):
|
|
"""Should auto-generate project_id from config."""
|
|
plugin = WordPressPlugin(self.VALID_CONFIG)
|
|
assert plugin.project_id.startswith("wordpress")
|
|
|
|
def test_plugin_name(self):
|
|
"""Should return 'wordpress' as plugin name."""
|
|
assert WordPressPlugin.get_plugin_name() == "wordpress"
|
|
|
|
def test_required_config_keys(self):
|
|
"""Should require url, username, app_password."""
|
|
keys = WordPressPlugin.get_required_config_keys()
|
|
assert "url" in keys
|
|
assert "username" in keys
|
|
assert "app_password" in keys
|
|
|
|
|
|
class TestWordPressToolSpecifications:
|
|
"""Test tool specification generation."""
|
|
|
|
def test_specs_not_empty(self):
|
|
"""Should return non-empty tool specifications."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
assert len(specs) > 0
|
|
|
|
def test_specs_count(self):
|
|
"""Should return at least 65 tool specs (65 documented + possible additions)."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
assert len(specs) >= 65
|
|
|
|
def test_specs_have_required_fields(self):
|
|
"""Each spec should have name, method_name, description, schema, scope."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
for spec in specs:
|
|
assert "name" in spec, "Missing 'name' in spec"
|
|
assert "method_name" in spec, f"Missing 'method_name' in {spec.get('name')}"
|
|
assert "description" in spec, f"Missing 'description' in {spec.get('name')}"
|
|
assert "schema" in spec, f"Missing 'schema' in {spec.get('name')}"
|
|
assert "scope" in spec, f"Missing 'scope' in {spec.get('name')}"
|
|
|
|
def test_specs_scope_values(self):
|
|
"""All scopes should be valid (read, write, admin)."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
valid_scopes = {"read", "write", "admin"}
|
|
for spec in specs:
|
|
assert (
|
|
spec["scope"] in valid_scopes
|
|
), f"Invalid scope '{spec['scope']}' in {spec['name']}"
|
|
|
|
def test_specs_unique_names(self):
|
|
"""All tool names should be unique."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
names = [s["name"] for s in specs]
|
|
assert len(names) == len(
|
|
set(names)
|
|
), f"Duplicate tool names found: {[n for n in names if names.count(n) > 1]}"
|
|
|
|
def test_core_tools_present(self):
|
|
"""Should include key WordPress tools."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
names = {s["name"] for s in specs}
|
|
expected = {"list_posts", "create_post", "get_post", "update_post", "delete_post"}
|
|
assert expected.issubset(names), f"Missing core tools: {expected - names}"
|
|
|
|
def test_media_tools_present(self):
|
|
"""Should include media tools."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
names = {s["name"] for s in specs}
|
|
assert "list_media" in names
|
|
assert "upload_media_from_url" in names
|
|
|
|
def test_taxonomy_tools_present(self):
|
|
"""Should include taxonomy tools."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
names = {s["name"] for s in specs}
|
|
assert "list_categories" in names
|
|
assert "list_tags" in names
|
|
|
|
def test_wp_cli_tools_present(self):
|
|
"""Should include WP-CLI tools."""
|
|
specs = WordPressPlugin.get_tool_specifications()
|
|
names = {s["name"] for s in specs}
|
|
assert "wp_cache_flush" in names
|
|
assert "wp_db_check" in names
|
|
|
|
|
|
class TestWordPressHandlerDelegation:
|
|
"""Test that plugin methods delegate to handlers correctly."""
|
|
|
|
VALID_CONFIG = {
|
|
"url": "https://example.com",
|
|
"username": "admin",
|
|
"app_password": "xxxx xxxx xxxx xxxx",
|
|
}
|
|
|
|
@pytest.fixture
|
|
def plugin(self):
|
|
return WordPressPlugin(self.VALID_CONFIG)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_posts_delegates(self, plugin):
|
|
"""list_posts should delegate to posts handler."""
|
|
plugin.posts.list_posts = AsyncMock(return_value={"posts": []})
|
|
result = await plugin.list_posts(per_page=5)
|
|
plugin.posts.list_posts.assert_called_once_with(per_page=5)
|
|
assert result == {"posts": []}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_post_delegates(self, plugin):
|
|
"""create_post should delegate to posts handler."""
|
|
plugin.posts.create_post = AsyncMock(return_value={"id": 42})
|
|
result = await plugin.create_post(title="Test", content="Hello")
|
|
plugin.posts.create_post.assert_called_once_with(title="Test", content="Hello")
|
|
assert result == {"id": 42}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_media_delegates(self, plugin):
|
|
"""list_media should delegate to media handler."""
|
|
plugin.media.list_media = AsyncMock(return_value=[])
|
|
await plugin.list_media()
|
|
plugin.media.list_media.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_categories_delegates(self, plugin):
|
|
"""list_categories should delegate to taxonomy handler."""
|
|
plugin.taxonomy.list_categories = AsyncMock(return_value=[])
|
|
await plugin.list_categories()
|
|
plugin.taxonomy.list_categories.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_comments_delegates(self, plugin):
|
|
"""list_comments should delegate to comments handler."""
|
|
plugin.comments.list_comments = AsyncMock(return_value=[])
|
|
await plugin.list_comments()
|
|
plugin.comments.list_comments.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wp_cli_not_available(self, plugin):
|
|
"""WP-CLI methods should return error when not configured."""
|
|
result = await plugin.wp_cache_flush()
|
|
assert "error" in result.lower() or "error" in json.loads(result)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_delegates(self, plugin):
|
|
"""health_check should delegate to site handler."""
|
|
plugin.site.health_check = AsyncMock(return_value={"healthy": True})
|
|
result = await plugin.health_check()
|
|
assert result["healthy"] is True
|
|
|
|
|
|
class TestWordPressPluginInfo:
|
|
"""Test plugin info and metadata methods."""
|
|
|
|
VALID_CONFIG = {
|
|
"url": "https://example.com",
|
|
"username": "admin",
|
|
"app_password": "xxxx xxxx xxxx xxxx",
|
|
}
|
|
|
|
def test_get_project_info(self):
|
|
"""Should return structured project info."""
|
|
plugin = WordPressPlugin(self.VALID_CONFIG, project_id="wp_test")
|
|
info = plugin.get_project_info()
|
|
assert info["project_id"] == "wp_test"
|
|
assert info["plugin_type"] == "wordpress"
|
|
assert "url" in info["config_keys"]
|
|
|
|
def test_get_tools_returns_empty(self):
|
|
"""Legacy get_tools should return empty list (Option B architecture)."""
|
|
plugin = WordPressPlugin(self.VALID_CONFIG)
|
|
assert plugin.get_tools() == []
|
|
|
|
|
|
# --- PluginRegistry Tests ---
|
|
|
|
|
|
class TestPluginRegistryWithWordPress:
|
|
"""Test PluginRegistry with WordPress plugin."""
|
|
|
|
def test_register_wordpress(self):
|
|
"""Should register WordPress plugin type."""
|
|
reg = PluginRegistry()
|
|
reg.register("wordpress", WordPressPlugin)
|
|
assert reg.is_registered("wordpress")
|
|
assert "wordpress" in reg.get_registered_types()
|
|
|
|
def test_create_instance(self):
|
|
"""Should create WordPress plugin instance."""
|
|
reg = PluginRegistry()
|
|
reg.register("wordpress", WordPressPlugin)
|
|
config = {
|
|
"url": "https://example.com",
|
|
"username": "admin",
|
|
"app_password": "xxxx",
|
|
}
|
|
instance = reg.create_instance("wordpress", "wp_site1", config)
|
|
assert isinstance(instance, WordPressPlugin)
|
|
assert instance.project_id == "wp_site1"
|
|
|
|
def test_register_non_baseplugin_raises(self):
|
|
"""Should reject classes not inheriting BasePlugin."""
|
|
reg = PluginRegistry()
|
|
with pytest.raises(TypeError, match="must inherit from BasePlugin"):
|
|
reg.register("bad", dict)
|
|
|
|
def test_unknown_type_raises(self):
|
|
"""Should raise KeyError for unknown plugin type."""
|
|
reg = PluginRegistry()
|
|
with pytest.raises(KeyError, match="Unknown plugin type"):
|
|
reg.create_instance("nonexistent", "id1", {})
|