rsp2k
ddf23d10e4
compose_feedback_services() called import_server() without awaiting it. That method became async in FastMCP 2.14, so on the version this package was built against it composed nothing while logging success. FastMCP 4 removed it outright. Use mount(), which is sync on 2.12 through 4.x. The test suite could not start (pytest-html hook, plugin not installed), so nothing caught this. Add pytest-html, bring examples/ over from the docs repo since the integration tests import them, and fix the tests that had drifted from the code. 407 tests pass on FastMCP 2.12.2, 2.14.3 and 4.0.10. - Declare fastmcp>=2.12.2,<5; CI matrix covers each family plus a non-blocking canary on the newest FastMCP - Add a composition test with real servers; the mocked ones could not see a missing await - Rewrite the README against the actual API and execute its examples - Ignore ruff UP042: StrEnum would change stored enum formatting - Version 2026.09.27
686 lines
27 KiB
Python
686 lines
27 KiB
Python
"""Unit tests for FastMCP Feedback tool mixins."""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, AsyncMock, patch
|
|
from fastmcp import FastMCP
|
|
|
|
from fastmcp_feedback.mixins import SubmissionMixin, RetrievalMixin, ManagementMixin
|
|
from fastmcp_feedback.models import SubmitFeedbackRequest, FeedbackStatus
|
|
from fastmcp_feedback.database import FeedbackDatabase
|
|
from tests._compat import tool_map
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestSubmissionMixin:
|
|
"""Test the SubmissionMixin class."""
|
|
|
|
async def test_mixin_initialization(self, test_database, test_insights):
|
|
"""Test SubmissionMixin initialization."""
|
|
mixin = SubmissionMixin(test_database, test_insights)
|
|
|
|
assert mixin.database == test_database
|
|
assert mixin.insights == test_insights
|
|
assert hasattr(mixin, 'submit_feedback')
|
|
|
|
async def test_submit_feedback_success(self, test_database, test_insights, sample_feedback_data):
|
|
"""Test successful feedback submission."""
|
|
await test_database.initialize()
|
|
mixin = SubmissionMixin(test_database, test_insights)
|
|
|
|
request = SubmitFeedbackRequest(**sample_feedback_data)
|
|
result = await mixin.submit_feedback(request)
|
|
|
|
assert result["success"] is True
|
|
assert "feedback_id" in result
|
|
assert result["message"] == "Feedback submitted successfully"
|
|
|
|
# Verify insights were recorded
|
|
metrics = test_insights.get_metrics("feedback_submitted")
|
|
assert len(metrics) == 1
|
|
assert metrics[0].value["type"] == "bug"
|
|
|
|
await test_database.close()
|
|
|
|
async def test_submit_feedback_validation_error(self, test_database, test_insights):
|
|
"""Test feedback submission with validation error."""
|
|
mixin = SubmissionMixin(test_database, test_insights)
|
|
|
|
# Invalid request with empty title
|
|
invalid_request = {
|
|
"type": "bug",
|
|
"title": "", # Empty title should fail validation
|
|
"description": "Test description",
|
|
"submitter": "test_user"
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="Title cannot be empty"):
|
|
request = SubmitFeedbackRequest(**invalid_request)
|
|
await mixin.submit_feedback(request)
|
|
|
|
async def test_submit_feedback_database_error(self, test_insights, sample_feedback_data):
|
|
"""Test feedback submission with database error."""
|
|
# Create a database with invalid URL to simulate error
|
|
bad_database = FeedbackDatabase("invalid://bad_url")
|
|
mixin = SubmissionMixin(bad_database, test_insights)
|
|
|
|
request = SubmitFeedbackRequest(**sample_feedback_data)
|
|
|
|
# This should handle database connection errors gracefully
|
|
result = await mixin.submit_feedback(request)
|
|
|
|
assert result["success"] is False
|
|
assert "error" in result
|
|
|
|
async def test_submit_feedback_with_insights_disabled(self, test_database, disabled_insights, sample_feedback_data):
|
|
"""Test feedback submission with insights disabled."""
|
|
await test_database.initialize()
|
|
mixin = SubmissionMixin(test_database, disabled_insights)
|
|
|
|
request = SubmitFeedbackRequest(**sample_feedback_data)
|
|
result = await mixin.submit_feedback(request)
|
|
|
|
assert result["success"] is True
|
|
|
|
# No metrics should be recorded when insights disabled
|
|
metrics = disabled_insights.get_metrics("feedback_submitted")
|
|
assert len(metrics) == 0
|
|
|
|
await test_database.close()
|
|
|
|
async def test_tool_registration(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test registering submission tools on FastMCP server."""
|
|
mixin = SubmissionMixin(test_database, test_insights)
|
|
|
|
# Register tools without prefix
|
|
mixin.register_tools(mock_fastmcp_server)
|
|
|
|
# Verify tool was registered
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
assert "submit_feedback" in tool_names
|
|
|
|
async def test_tool_registration_with_prefix(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test registering submission tools with prefix."""
|
|
mixin = SubmissionMixin(test_database, test_insights)
|
|
|
|
# Register tools with prefix
|
|
mixin.register_tools(mock_fastmcp_server, prefix="support", separator="_")
|
|
|
|
# Verify tool was registered with prefix
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
assert "support_submit_feedback" in tool_names
|
|
|
|
async def test_multiple_feedback_submissions(self, test_database, test_insights, multiple_feedback_data):
|
|
"""Test submitting multiple feedback items."""
|
|
await test_database.initialize()
|
|
mixin = SubmissionMixin(test_database, test_insights)
|
|
|
|
feedback_ids = []
|
|
|
|
for data in multiple_feedback_data:
|
|
request = SubmitFeedbackRequest(**data)
|
|
result = await mixin.submit_feedback(request)
|
|
|
|
assert result["success"] is True
|
|
feedback_ids.append(result["feedback_id"])
|
|
|
|
# All feedback IDs should be unique
|
|
assert len(set(feedback_ids)) == len(feedback_ids)
|
|
|
|
# Should have recorded 3 metrics
|
|
metrics = test_insights.get_metrics("feedback_submitted")
|
|
assert len(metrics) == 3
|
|
|
|
await test_database.close()
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestRetrievalMixin:
|
|
"""Test the RetrievalMixin class."""
|
|
|
|
async def test_mixin_initialization(self, test_database, test_insights):
|
|
"""Test RetrievalMixin initialization."""
|
|
mixin = RetrievalMixin(test_database, test_insights)
|
|
|
|
assert mixin.database == test_database
|
|
assert mixin.insights == test_insights
|
|
assert hasattr(mixin, 'list_feedback')
|
|
assert hasattr(mixin, 'get_feedback_statistics')
|
|
|
|
async def test_list_feedback_empty(self, test_database, test_insights):
|
|
"""Test listing feedback when database is empty."""
|
|
await test_database.initialize()
|
|
mixin = RetrievalMixin(test_database, test_insights)
|
|
|
|
result = await mixin.list_feedback()
|
|
|
|
assert result["feedback"] == []
|
|
assert result["total_count"] == 0
|
|
assert result["page"] == 1
|
|
assert result["per_page"] == 10
|
|
|
|
await test_database.close()
|
|
|
|
async def test_list_feedback_with_data(self, populated_database, test_insights):
|
|
"""Test listing feedback with populated database."""
|
|
mixin = RetrievalMixin(populated_database, test_insights)
|
|
|
|
result = await mixin.list_feedback()
|
|
|
|
assert len(result["feedback"]) > 0
|
|
assert result["total_count"] > 0
|
|
|
|
# Verify feedback structure
|
|
feedback_item = result["feedback"][0]
|
|
required_fields = ["id", "type", "title", "description", "status", "created_at"]
|
|
for field in required_fields:
|
|
assert field in feedback_item
|
|
|
|
async def test_list_feedback_with_type_filter(self, populated_database, test_insights):
|
|
"""Test listing feedback with type filter."""
|
|
mixin = RetrievalMixin(populated_database, test_insights)
|
|
|
|
result = await mixin.list_feedback(type_filter="bug")
|
|
|
|
# All returned feedback should be of type "bug"
|
|
for feedback in result["feedback"]:
|
|
assert feedback["type"] == "bug"
|
|
|
|
async def test_list_feedback_with_status_filter(self, populated_database, test_insights):
|
|
"""Test listing feedback with status filter."""
|
|
mixin = RetrievalMixin(populated_database, test_insights)
|
|
|
|
result = await mixin.list_feedback(status_filter="open")
|
|
|
|
# All returned feedback should have "open" status
|
|
for feedback in result["feedback"]:
|
|
assert feedback["status"] == "open"
|
|
|
|
async def test_list_feedback_pagination(self, populated_database, test_insights):
|
|
"""Test feedback listing pagination."""
|
|
mixin = RetrievalMixin(populated_database, test_insights)
|
|
|
|
# Get first page with limit of 2
|
|
page1 = await mixin.list_feedback(page=1, per_page=2)
|
|
|
|
assert len(page1["feedback"]) <= 2
|
|
assert page1["page"] == 1
|
|
assert page1["per_page"] == 2
|
|
|
|
# Get second page
|
|
page2 = await mixin.list_feedback(page=2, per_page=2)
|
|
|
|
assert page2["page"] == 2
|
|
assert page2["per_page"] == 2
|
|
|
|
# Pages should contain different items
|
|
if len(page1["feedback"]) > 0 and len(page2["feedback"]) > 0:
|
|
page1_ids = {item["id"] for item in page1["feedback"]}
|
|
page2_ids = {item["id"] for item in page2["feedback"]}
|
|
assert page1_ids.isdisjoint(page2_ids)
|
|
|
|
async def test_get_feedback_statistics(self, populated_database, test_insights):
|
|
"""Test getting feedback statistics."""
|
|
mixin = RetrievalMixin(populated_database, test_insights)
|
|
|
|
stats = await mixin.get_feedback_statistics()
|
|
|
|
assert "total_count" in stats
|
|
assert "by_type" in stats
|
|
assert "by_status" in stats
|
|
assert "recent_count" in stats
|
|
|
|
assert stats["total_count"] >= 0
|
|
assert isinstance(stats["by_type"], dict)
|
|
assert isinstance(stats["by_status"], dict)
|
|
|
|
async def test_get_feedback_by_id(self, populated_database, test_insights):
|
|
"""Test getting specific feedback by ID."""
|
|
mixin = RetrievalMixin(populated_database, test_insights)
|
|
|
|
# Get list to find valid ID
|
|
feedback_list = await mixin.list_feedback()
|
|
if len(feedback_list["feedback"]) > 0:
|
|
feedback_id = feedback_list["feedback"][0]["id"]
|
|
|
|
feedback = await mixin.get_feedback_by_id(feedback_id)
|
|
|
|
assert feedback is not None
|
|
assert feedback["id"] == feedback_id
|
|
|
|
async def test_get_feedback_by_invalid_id(self, test_database, test_insights):
|
|
"""Test getting feedback with invalid ID."""
|
|
await test_database.initialize()
|
|
mixin = RetrievalMixin(test_database, test_insights)
|
|
|
|
feedback = await mixin.get_feedback_by_id("nonexistent_id")
|
|
|
|
assert feedback is None
|
|
|
|
await test_database.close()
|
|
|
|
async def test_tool_registration(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test registering retrieval tools on FastMCP server."""
|
|
mixin = RetrievalMixin(test_database, test_insights)
|
|
|
|
mixin.register_tools(mock_fastmcp_server)
|
|
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
|
|
expected_tools = ["list_feedback", "get_feedback_statistics"]
|
|
for tool_name in expected_tools:
|
|
assert tool_name in tool_names
|
|
|
|
async def test_insights_tracking(self, test_database, test_insights):
|
|
"""Test that retrieval operations track insights."""
|
|
await test_database.initialize()
|
|
mixin = RetrievalMixin(test_database, test_insights)
|
|
|
|
# List feedback should record metric
|
|
await mixin.list_feedback()
|
|
|
|
metrics = test_insights.get_metrics("feedback_listed")
|
|
assert len(metrics) == 1
|
|
|
|
# Get statistics should record metric
|
|
await mixin.get_feedback_statistics()
|
|
|
|
stats_metrics = test_insights.get_metrics("feedback_statistics_viewed")
|
|
assert len(stats_metrics) == 1
|
|
|
|
await test_database.close()
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestManagementMixin:
|
|
"""Test the ManagementMixin class."""
|
|
|
|
async def test_mixin_initialization(self, test_database, test_insights):
|
|
"""Test ManagementMixin initialization."""
|
|
mixin = ManagementMixin(test_database, test_insights)
|
|
|
|
assert mixin.database == test_database
|
|
assert mixin.insights == test_insights
|
|
assert hasattr(mixin, 'update_feedback_status')
|
|
assert hasattr(mixin, 'delete_feedback')
|
|
|
|
async def test_update_feedback_status_success(self, populated_database, test_insights):
|
|
"""Test successful feedback status update."""
|
|
mixin = ManagementMixin(populated_database, test_insights)
|
|
|
|
# Get a feedback item to update
|
|
retrieval = RetrievalMixin(populated_database, test_insights)
|
|
feedback_list = await retrieval.list_feedback()
|
|
|
|
if len(feedback_list["feedback"]) > 0:
|
|
feedback_id = feedback_list["feedback"][0]["id"]
|
|
|
|
result = await mixin.update_feedback_status(
|
|
feedback_id,
|
|
FeedbackStatus.IN_PROGRESS
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["message"] == "Feedback status updated successfully"
|
|
|
|
# Verify the update
|
|
updated_feedback = await retrieval.get_feedback_by_id(feedback_id)
|
|
assert updated_feedback["status"] == "in_progress"
|
|
|
|
async def test_update_feedback_status_invalid_id(self, test_database, test_insights):
|
|
"""Test updating feedback status with invalid ID."""
|
|
await test_database.initialize()
|
|
mixin = ManagementMixin(test_database, test_insights)
|
|
|
|
result = await mixin.update_feedback_status(
|
|
"99999", # Use a valid integer format that won't exist
|
|
FeedbackStatus.RESOLVED
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "not found" in result["error"].lower()
|
|
|
|
await test_database.close()
|
|
|
|
async def test_update_feedback_status_with_note(self, populated_database, test_insights):
|
|
"""Test updating feedback status with a note."""
|
|
mixin = ManagementMixin(populated_database, test_insights)
|
|
|
|
# Get feedback item
|
|
retrieval = RetrievalMixin(populated_database, test_insights)
|
|
feedback_list = await retrieval.list_feedback()
|
|
|
|
if len(feedback_list["feedback"]) > 0:
|
|
feedback_id = feedback_list["feedback"][0]["id"]
|
|
|
|
result = await mixin.update_feedback_status(
|
|
feedback_id,
|
|
FeedbackStatus.RESOLVED,
|
|
note="Fixed in version 1.2.3"
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
# Verify note was recorded in insights
|
|
metrics = test_insights.get_metrics("feedback_status_updated")
|
|
assert len(metrics) >= 1
|
|
# Check that the note information is recorded (has_note field should be True)
|
|
assert any(m.value.get("has_note") is True for m in metrics)
|
|
|
|
async def test_delete_feedback_success(self, populated_database, test_insights):
|
|
"""Test successful feedback deletion."""
|
|
mixin = ManagementMixin(populated_database, test_insights)
|
|
retrieval = RetrievalMixin(populated_database, test_insights)
|
|
|
|
# Get feedback item to delete
|
|
feedback_list = await retrieval.list_feedback()
|
|
initial_count = len(feedback_list["feedback"])
|
|
|
|
if initial_count > 0:
|
|
feedback_id = feedback_list["feedback"][0]["id"]
|
|
|
|
result = await mixin.delete_feedback(feedback_id)
|
|
|
|
assert result["success"] is True
|
|
assert result["message"] == "Feedback deleted successfully"
|
|
|
|
# Verify deletion
|
|
updated_list = await retrieval.list_feedback()
|
|
assert len(updated_list["feedback"]) == initial_count - 1
|
|
|
|
# Verify feedback is gone
|
|
deleted_feedback = await retrieval.get_feedback_by_id(feedback_id)
|
|
assert deleted_feedback is None
|
|
|
|
async def test_delete_feedback_invalid_id(self, test_database, test_insights):
|
|
"""Test deleting feedback with invalid ID."""
|
|
await test_database.initialize()
|
|
mixin = ManagementMixin(test_database, test_insights)
|
|
|
|
result = await mixin.delete_feedback("99999") # Use valid integer format
|
|
|
|
assert result["success"] is False
|
|
assert "not found" in result["error"].lower()
|
|
|
|
await test_database.close()
|
|
|
|
async def test_bulk_status_update(self, populated_database, test_insights):
|
|
"""Test bulk status updates."""
|
|
mixin = ManagementMixin(populated_database, test_insights)
|
|
retrieval = RetrievalMixin(populated_database, test_insights)
|
|
|
|
# Get multiple feedback IDs
|
|
feedback_list = await retrieval.list_feedback()
|
|
feedback_ids = [item["id"] for item in feedback_list["feedback"]]
|
|
|
|
if len(feedback_ids) >= 2:
|
|
result = await mixin.bulk_update_status(
|
|
feedback_ids[:2],
|
|
FeedbackStatus.CLOSED
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["updated_count"] == 2
|
|
|
|
# Verify updates
|
|
for feedback_id in feedback_ids[:2]:
|
|
feedback = await retrieval.get_feedback_by_id(feedback_id)
|
|
assert feedback["status"] == "closed"
|
|
|
|
async def test_tool_registration(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test registering management tools on FastMCP server."""
|
|
mixin = ManagementMixin(test_database, test_insights)
|
|
|
|
mixin.register_tools(mock_fastmcp_server)
|
|
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
|
|
expected_tools = ["update_feedback_status", "delete_feedback"]
|
|
for tool_name in expected_tools:
|
|
assert tool_name in tool_names
|
|
|
|
async def test_insights_tracking_for_management(self, populated_database, test_insights):
|
|
"""Test insights tracking for management operations."""
|
|
mixin = ManagementMixin(populated_database, test_insights)
|
|
retrieval = RetrievalMixin(populated_database, test_insights)
|
|
|
|
# Get feedback to manage
|
|
feedback_list = await retrieval.list_feedback()
|
|
|
|
if len(feedback_list["feedback"]) > 0:
|
|
feedback_id = feedback_list["feedback"][0]["id"]
|
|
|
|
# Update status - should record metric
|
|
await mixin.update_feedback_status(feedback_id, FeedbackStatus.IN_PROGRESS)
|
|
|
|
status_metrics = test_insights.get_metrics("feedback_status_updated")
|
|
assert len(status_metrics) >= 1
|
|
|
|
# Delete feedback - should record metric
|
|
await mixin.delete_feedback(feedback_id)
|
|
|
|
delete_metrics = test_insights.get_metrics("feedback_deleted")
|
|
assert len(delete_metrics) >= 1
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestMixinIntegration:
|
|
"""Test integration between different mixins."""
|
|
|
|
async def test_full_feedback_lifecycle(self, test_database, test_insights, sample_feedback_data):
|
|
"""Test complete feedback lifecycle using all mixins."""
|
|
await test_database.initialize()
|
|
|
|
# Initialize all mixins
|
|
submission = SubmissionMixin(test_database, test_insights)
|
|
retrieval = RetrievalMixin(test_database, test_insights)
|
|
management = ManagementMixin(test_database, test_insights)
|
|
|
|
# 1. Submit feedback
|
|
request = SubmitFeedbackRequest(**sample_feedback_data)
|
|
submit_result = await submission.submit_feedback(request)
|
|
|
|
assert submit_result["success"] is True
|
|
feedback_id = submit_result["feedback_id"]
|
|
|
|
# 2. Retrieve feedback
|
|
feedback = await retrieval.get_feedback_by_id(feedback_id)
|
|
assert feedback is not None
|
|
assert feedback["status"] == "open"
|
|
|
|
# 3. Update status
|
|
update_result = await management.update_feedback_status(
|
|
feedback_id,
|
|
FeedbackStatus.IN_PROGRESS
|
|
)
|
|
assert update_result["success"] is True
|
|
|
|
# 4. Verify update
|
|
updated_feedback = await retrieval.get_feedback_by_id(feedback_id)
|
|
assert updated_feedback["status"] == "in_progress"
|
|
|
|
# 5. Get statistics
|
|
stats = await retrieval.get_feedback_statistics()
|
|
assert stats["total_count"] >= 1
|
|
assert "IN_PROGRESS" in stats["by_status"]
|
|
|
|
# 6. Final status update
|
|
final_result = await management.update_feedback_status(
|
|
feedback_id,
|
|
FeedbackStatus.RESOLVED
|
|
)
|
|
assert final_result["success"] is True
|
|
|
|
await test_database.close()
|
|
|
|
async def test_mixin_server_composition(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test registering all mixins on the same server."""
|
|
# Initialize all mixins
|
|
submission = SubmissionMixin(test_database, test_insights)
|
|
retrieval = RetrievalMixin(test_database, test_insights)
|
|
management = ManagementMixin(test_database, test_insights)
|
|
|
|
# Register all tools
|
|
submission.register_tools(mock_fastmcp_server)
|
|
retrieval.register_tools(mock_fastmcp_server)
|
|
management.register_tools(mock_fastmcp_server)
|
|
|
|
# Verify all tools are registered
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
|
|
expected_tools = [
|
|
"submit_feedback",
|
|
"list_feedback",
|
|
"get_feedback_statistics",
|
|
"update_feedback_status",
|
|
"delete_feedback"
|
|
]
|
|
|
|
for tool_name in expected_tools:
|
|
assert tool_name in tool_names
|
|
|
|
async def test_mixin_prefix_isolation(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test mixin tools with different prefixes don't conflict."""
|
|
submission = SubmissionMixin(test_database, test_insights)
|
|
retrieval = RetrievalMixin(test_database, test_insights)
|
|
|
|
# Register with different prefixes
|
|
submission.register_tools(mock_fastmcp_server, prefix="admin")
|
|
retrieval.register_tools(mock_fastmcp_server, prefix="public")
|
|
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
|
|
assert "admin_submit_feedback" in tool_names
|
|
assert "public_list_feedback" in tool_names
|
|
assert "submit_feedback" not in tool_names # No unprefixed versions
|
|
assert "list_feedback" not in tool_names
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestMixinErrorHandling:
|
|
"""Test error handling across all mixins."""
|
|
|
|
async def test_database_error_handling(self, test_insights):
|
|
"""Test mixin behavior with database errors."""
|
|
# Use uninitialized database to trigger errors
|
|
bad_database = Mock()
|
|
bad_database.session = Mock(side_effect=Exception("Database error"))
|
|
|
|
submission = SubmissionMixin(bad_database, test_insights)
|
|
|
|
request = SubmitFeedbackRequest(
|
|
type="bug",
|
|
title="Test feedback",
|
|
description="Test description",
|
|
submitter="test_user"
|
|
)
|
|
|
|
result = await submission.submit_feedback(request)
|
|
|
|
assert result["success"] is False
|
|
assert "error" in result
|
|
|
|
async def test_insights_error_handling(self, test_database):
|
|
"""Test mixin behavior with insights errors."""
|
|
await test_database.initialize()
|
|
|
|
# Mock insights to raise errors for specific calls
|
|
bad_insights = Mock()
|
|
def failing_record_metric(metric_name, value):
|
|
if metric_name == "feedback_submitted":
|
|
raise Exception("Insights error")
|
|
# Allow tool_used metrics to succeed to avoid cascade failures
|
|
return None
|
|
|
|
bad_insights.record_metric = Mock(side_effect=failing_record_metric)
|
|
bad_insights.enabled = True
|
|
|
|
submission = SubmissionMixin(test_database, bad_insights)
|
|
|
|
request = SubmitFeedbackRequest(
|
|
type="bug",
|
|
title="Test feedback",
|
|
description="Test description",
|
|
submitter="test_user"
|
|
)
|
|
|
|
# Current implementation propagates insights errors, so submission should fail
|
|
result = await submission.submit_feedback(request)
|
|
|
|
# The implementation currently doesn't handle insights errors gracefully
|
|
assert result["success"] is False
|
|
assert "error" in result
|
|
|
|
await test_database.close()
|
|
|
|
async def test_concurrent_operations(self, test_database, test_insights, multiple_feedback_data):
|
|
"""Test concurrent operations across mixins."""
|
|
await test_database.initialize()
|
|
|
|
submission = SubmissionMixin(test_database, test_insights)
|
|
retrieval = RetrievalMixin(test_database, test_insights)
|
|
|
|
import asyncio
|
|
|
|
# Submit multiple feedback concurrently
|
|
tasks = []
|
|
for data in multiple_feedback_data:
|
|
request = SubmitFeedbackRequest(**data)
|
|
task = submission.submit_feedback(request)
|
|
tasks.append(task)
|
|
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
# All submissions should succeed
|
|
for result in results:
|
|
assert result["success"] is True
|
|
|
|
# All should be retrievable
|
|
feedback_list = await retrieval.list_feedback()
|
|
assert len(feedback_list["feedback"]) == len(multiple_feedback_data)
|
|
|
|
await test_database.close()
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
class TestMixinConfiguration:
|
|
"""Test mixin configuration options."""
|
|
|
|
async def test_mixin_with_different_separators(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test mixin tool registration with different separators."""
|
|
submission = SubmissionMixin(test_database, test_insights)
|
|
|
|
# Register with dash separator
|
|
submission.register_tools(mock_fastmcp_server, prefix="api", separator="-")
|
|
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
|
|
assert "api-submit_feedback" in tool_names
|
|
|
|
async def test_mixin_selective_tool_registration(self, mock_fastmcp_server, test_database, test_insights):
|
|
"""Test mixin tool registration (selective registration not directly supported)."""
|
|
submission = SubmissionMixin(test_database, test_insights)
|
|
|
|
# MCPMixin doesn't support selective tool registration with 'tools' parameter
|
|
# Test the standard registration instead
|
|
submission.register_tools(mock_fastmcp_server)
|
|
|
|
tools = await tool_map(mock_fastmcp_server)
|
|
tool_names = list(tools.keys())
|
|
|
|
# SubmissionMixin should only register its tools
|
|
assert "submit_feedback" in tool_names
|
|
# Should not have retrieval tools
|
|
assert "list_feedback" not in tool_names |