rsp2k
1213b1587e
get_feedback_statistics and get_database_stats grouped the raw columns, which hold enum names (BUG, OPEN) on SQLite and Postgres, while list_feedback returns values (bug, open). Both now count through count_feedback_by and return values. The identity resolver ran after the call, so record_event inside a tool never had user_sub, caller_kind or client_id. It now runs once before the tool; the call scope carries the result to record_event and the call's own record reuses it. Importing the package called logging.basicConfig at INFO. It now adds a NullHandler to the fastmcp_feedback logger; setup_logging() stays as an opt-in. PACKAGE_INFO is read from the installed metadata. psycopg2-binary joins the dev and test extras so the feedback tools can be tested on Postgres.
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 |