From f02a4b82e7f73403719474f327b105d8728e8934 Mon Sep 17 00:00:00 2001 From: Chintan Joshi Date: Tue, 31 Mar 2026 17:03:41 +0000 Subject: [PATCH] fix: clear the debt --- forum/ai_moderation.py | 422 ++++++++++++++++-------------------- forum/settings/common.py | 16 ++ tests/test_ai_moderation.py | 133 ++++-------- 3 files changed, 241 insertions(+), 330 deletions(-) diff --git a/forum/ai_moderation.py b/forum/ai_moderation.py index 0f7a1816..7d29ac2c 100644 --- a/forum/ai_moderation.py +++ b/forum/ai_moderation.py @@ -97,192 +97,162 @@ def create_moderation_audit_log( log.error(f"Failed to create database audit log: {db_error}") -class AIModerationService: +def moderate_and_flag_spam( + content: str, + content_instance: Any, + course_id: Optional[str] = None, + backend: Optional[Any] = None, +) -> Dict[str, Any]: """ - Service for AI-based content moderation. + Moderate content and flag as spam if detected. + + This function checks content using AI moderation service and takes appropriate actions + based on the settings. Settings and API calls are only accessed after checking if + AI moderation is enabled via waffle flags. + + Main workflow: + 1. Check if AI moderation is enabled (waffle flag) + 2. Make API request to AI moderation service + 3. If spam detected: + a. Flag content as spam and abuse + b. Optionally soft-delete if auto-delete is enabled + c. Create audit log - Waffle Flag "discussion.enable_ai_moderation" controls whether AI moderation is active. + Args: + content: The text content to moderate + content_instance: The content model instance (Thread or Comment) + course_id: Optional course ID for waffle flag checking + backend: Backend instance for database operations - XPERT AI Moderation API is used to classify content as spam or not spam. + Returns: + Dictionary with moderation results and actions taken + + TODO: + - Add content check for images """ + # Initialize result with default values + result = { + "is_spam": False, + "reasoning": "AI moderation disabled or unavailable", + "classification": "not_spam", + "actions_taken": ["no_action"], + "flagged": False, + } + + # ============================================================================ + # STEP 1: Check if AI moderation is enabled (waffle flag check) + # ============================================================================ + # pylint: disable=import-outside-toplevel + from forum.toggles import ( + is_ai_moderation_enabled, + is_ai_auto_delete_spam_enabled, + ) - def __init__(self): # type: ignore[no-untyped-def] - """Initialize the AI moderation service.""" - self.api_url = getattr(settings, "AI_MODERATION_API_URL", None) - self.client_id = getattr(settings, "AI_MODERATION_CLIENT_ID", None) - self.system_message = getattr(settings, "AI_MODERATION_SYSTEM_MESSAGE", None) - self.connection_timeout = getattr( - settings, "AI_MODERATION_CONNECTION_TIMEOUT", 30 - ) # seconds - self.read_timeout = getattr( - settings, "AI_MODERATION_READ_TIMEOUT", 30 - ) # seconds - self.ai_moderation_user_id = getattr(settings, "AI_MODERATION_USER_ID", None) - - def _make_api_request(self, content: str) -> Optional[Dict[str, Any]]: - """ - Make API request to XPert Service. - - Args: - content: The text content to moderate - - Returns: - Dictionary with 'reasoning' and 'classification' keys, or None if failed - """ - if not self.api_url: - log.error("AI_MODERATION_API_URL setting is not configured") - return None - - headers = { - "accept": "*/*", - "accept-language": "en-US,en;q=0.9", - "content-type": "application/json", - "user-agent": "Mozilla/5.0 (compatible; edX-Forum-AI-Moderation/1.0)", - } + course_key = CourseKey.from_string(course_id) if course_id else None + if not is_ai_moderation_enabled(course_key): # type: ignore[no-untyped-call] + return result - payload = { - "messages": [{"role": "user", "content": content}], - "client_id": self.client_id, - "system_message": self.system_message, - } + # ============================================================================ + # STEP 2: Make API request to AI moderation service + # ============================================================================ + api_url = settings.AI_MODERATION_API_URL + if not api_url: + log.error("AI_MODERATION_API_URL setting is not configured") + result["reasoning"] = "AI moderation API not configured" + return result - try: - response = requests.post( - self.api_url, - headers=headers, - json=payload, - timeout=(self.connection_timeout, self.read_timeout), - ) - response.raise_for_status() - - response_data = response.json() - # Validate response data structure - if not isinstance(response_data, list): - log.error( - f"Expected list response from XPert API, got {type(response_data)}" - ) - return None - - if len(response_data) == 0: - log.error("Empty response list from XPert API") - return None - - if not isinstance(response_data[0], dict): - log.error( - f"Expected dict in response list, got {type(response_data[0])}" - ) - return None - - assistant_content = response_data[0].get("content", "") - # Parse the JSON content from the assistant response - try: - moderation_result = json.loads(assistant_content) - # full API response for audit purposes - moderation_result["full_api_response"] = response_data - return moderation_result - except json.JSONDecodeError as e: - log.error(f"Failed to parse AI moderation response JSON: {e}") - return None - except ( - requests.RequestException, - requests.Timeout, - requests.ConnectionError, - ) as e: - log.error(f"AI moderation API request failed: {e}") - return None - - def moderate_and_flag_content( - self, - content: str, - content_instance: Any, - course_id: Optional[str] = None, - backend: Optional[Any] = None, - ) -> Dict[str, Any]: - """ - Moderate content and flag as spam and flag abuse if detected. - - Args: - content: The text content to check - content_instance: The content model instance (Thread or Comment) - course_id: Optional course ID for waffle flag checking - backend: Backend instance for database operations - - Returns: - Dictionary with moderation results and actions taken - """ - result = { - "is_spam": False, - "reasoning": "AI moderation disabled or unavailable", - "classification": "not_spam", - "actions_taken": ["no_action"], - "flagged": False, - } - # Check if AI moderation is enabled - # pylint: disable=import-outside-toplevel - from forum.toggles import ( - is_ai_moderation_enabled, - is_ai_auto_delete_spam_enabled, + headers = { + "accept": "*/*", + "accept-language": "en-US,en;q=0.9", + "content-type": "application/json", + "user-agent": "Mozilla/5.0 (compatible; edX-Forum-AI-Moderation/1.0)", + } + + payload = { + "messages": [{"role": "user", "content": content}], + "client_id": settings.AI_MODERATION_CLIENT_ID, + "system_message": settings.AI_MODERATION_SYSTEM_MESSAGE, + } + + try: + response = requests.post( + api_url, + headers=headers, + json=payload, + timeout=( + settings.AI_MODERATION_CONNECTION_TIMEOUT, + settings.AI_MODERATION_READ_TIMEOUT, + ), ) + response.raise_for_status() + + response_data = response.json() + + # Validate response data structure + if not isinstance(response_data, list): + log.error( + f"Expected list response from XPert API, got {type(response_data)}" + ) + result["reasoning"] = "Invalid API response format" + return result - course_key = CourseKey.from_string(course_id) if course_id else None - if not is_ai_moderation_enabled(course_key): # type: ignore[no-untyped-call] + if len(response_data) == 0: + log.error("Empty response list from XPert API") + result["reasoning"] = "Empty API response" return result - # Make API request - moderation_result = self._make_api_request(content) + if not isinstance(response_data[0], dict): + log.error( + f"Expected dict in response list, got {type(response_data[0])}" + ) + result["reasoning"] = "Invalid API response structure" + return result - if moderation_result is None: - result["reasoning"] = "AI moderation API failed" - log.warning("AI moderation API failed") + assistant_content = response_data[0].get("content", "") + + # Parse the JSON content from the assistant response + try: + moderation_result = json.loads(assistant_content) + # Include full API response for audit purposes + moderation_result["full_api_response"] = response_data + except json.JSONDecodeError as e: + log.error(f"Failed to parse AI moderation response JSON: {e}") + result["reasoning"] = "Failed to parse API response" return result + + except (requests.RequestException, requests.Timeout, requests.ConnectionError) as e: + log.error(f"AI moderation API request failed: {e}") + result["reasoning"] = "AI moderation API failed" + return result - classification = moderation_result.get("classification", "not_spam") - reasoning = moderation_result.get("reasoning", "No reasoning provided") - is_spam = classification in ["spam", "spam_or_scam"] - - result.update( - { - "is_spam": is_spam, - "reasoning": reasoning, - "classification": classification, - "moderation_result": moderation_result, - } - ) + # ============================================================================ + # STEP 3: Process moderation result and determine if content is spam + # ============================================================================ + classification = moderation_result.get("classification", "not_spam") + reasoning = moderation_result.get("reasoning", "No reasoning provided") + is_spam = classification in ["spam", "spam_or_scam"] - if is_spam: - # Flag content as spam and abuse first - try: - content_instance["is_spam"] = True - - self._mark_as_spam_and_moderate(content_instance, backend) - result["actions_taken"] = ["flagged"] - result["flagged"] = True - except (AttributeError, ValueError, TypeError) as e: - log.error(f"Failed to flag content as spam: {e}") - result["actions_taken"] = ["no_action"] - - # Only attempt deletion if flagging succeeded - if is_ai_auto_delete_spam_enabled(course_key) and result["flagged"]: # type: ignore[no-untyped-call] - try: - self._delete_content(content_instance) - result["actions_taken"] = result["actions_taken"] + ["soft_deleted"] # type: ignore[operator] - except (ForumV2RequestError, ObjectDoesNotExist, ValidationError) as e: - log.error(f"Failed to delete content after flagging: {e}") - else: - result["actions_taken"] = ["no_action"] - - # Only create audit log for spam content (or API failures, handled above) - if is_spam: - create_moderation_audit_log( - content_instance, - moderation_result, - result["actions_taken"], # type: ignore[arg-type] - _get_author_from_content(content_instance), - ) + result.update( + { + "is_spam": is_spam, + "reasoning": reasoning, + "classification": classification, + "moderation_result": moderation_result, + } + ) + + # If not spam, we're done + if not is_spam: + result["actions_taken"] = ["no_action"] return result - def _mark_as_spam_and_moderate(self, content_instance: Any, backend: Any) -> None: - """Flag content as abuse using backend methods.""" + # ============================================================================ + # STEP 4: Flag content as spam and abuse + # ============================================================================ + try: + content_instance["is_spam"] = True + + # Flag content using backend methods content_id = str(content_instance.get("_id")) content_type = str(content_instance.get("_type")) extra_data = { @@ -290,72 +260,56 @@ def _mark_as_spam_and_moderate(self, content_instance: Any, backend: Any) -> Non "CommentThread" if content_type == "CommentThread" else "Comment" ) } - if not self.ai_moderation_user_id: + + ai_moderation_user_id = settings.AI_MODERATION_USER_ID + if not ai_moderation_user_id: raise ValueError("AI_MODERATION_USER_ID setting is not configured.") + backend.flag_content_as_spam(content_type, content_id) - backend.flag_as_abuse(str(self.ai_moderation_user_id), content_id, **extra_data) - - def _delete_content(self, content_instance: Any) -> None: - """ - Soft delete content using API layer delete functions. - - Uses the API layer which handles all business logic including: - - Content validation - - Soft deletion - - Stats updates - - Subscription cleanup (for threads) - - Anonymous content handling - - Args: - content_instance: Dict containing content data including _id, _type, and course_id - """ - # Import here to avoid circular dependency (api modules import from ai_moderation) - # pylint: disable=import-outside-toplevel,cyclic-import - from forum.api.comments import delete_comment - from forum.api.threads import delete_thread - - content_id = str(content_instance.get("_id")) - content_type = str(content_instance.get("_type")) - course_id = content_instance.get("course_id") - deleted_by = ( - str(self.ai_moderation_user_id) if self.ai_moderation_user_id else None - ) - - # Use API layer functions which handle all business logic - # Exceptions propagate to caller for proper error handling - if content_type == "CommentThread": - delete_thread(content_id, course_id=course_id, deleted_by=deleted_by) - log.info(f"AI Moderation Deleted CommentThread: {content_id}") - elif content_type == "Comment": - delete_comment(content_id, course_id=course_id, deleted_by=deleted_by) - log.info(f"AI Moderation Deleted Comment: {content_id}") - - -# Global instance -ai_moderation_service = AIModerationService() # type: ignore[no-untyped-call] - - -def moderate_and_flag_spam( - content: str, - content_instance: Any, - course_id: Optional[str] = None, - backend: Optional[Any] = None, -) -> Dict[str, Any]: - """ - Moderate content and flag as spam if detected. - - Args: - content: The text content to moderate - content_instance: The content model instance - course_id: Optional course ID for waffle flag checking - backend: Backend instance for database operations - - Returns: - Dictionary with moderation results and actions taken + backend.flag_as_abuse(str(ai_moderation_user_id), content_id, **extra_data) + + result["actions_taken"] = ["flagged"] + result["flagged"] = True + + except (AttributeError, ValueError, TypeError) as e: + log.error(f"Failed to flag content as spam: {e}") + result["actions_taken"] = ["no_action"] + # If flagging failed, skip deletion and audit log + return result - TODO:- - - Add content check for images - """ - return ai_moderation_service.moderate_and_flag_content( - content, content_instance, course_id, backend + # ============================================================================ + # STEP 5: Optionally soft-delete content if auto-delete is enabled + # ============================================================================ + if is_ai_auto_delete_spam_enabled(course_key): # type: ignore[no-untyped-call] + try: + # Import here to avoid circular dependency + # pylint: disable=import-outside-toplevel,cyclic-import + from forum.api.comments import delete_comment + from forum.api.threads import delete_thread + + deleted_by = str(ai_moderation_user_id) if ai_moderation_user_id else None + + # Use API layer functions which handle all business logic + if content_type == "CommentThread": + delete_thread(content_id, course_id=course_id, deleted_by=deleted_by) + log.info(f"AI Moderation Deleted CommentThread: {content_id}") + elif content_type == "Comment": + delete_comment(content_id, course_id=course_id, deleted_by=deleted_by) + log.info(f"AI Moderation Deleted Comment: {content_id}") + + result["actions_taken"] = result["actions_taken"] + ["soft_deleted"] # type: ignore[operator] + + except (ForumV2RequestError, ObjectDoesNotExist, ValidationError) as e: + log.error(f"Failed to delete content after flagging: {e}") + + # ============================================================================ + # STEP 6: Create audit log for spam content + # ============================================================================ + create_moderation_audit_log( + content_instance, + moderation_result, + result["actions_taken"], # type: ignore[arg-type] + _get_author_from_content(content_instance), ) + + return result diff --git a/forum/settings/common.py b/forum/settings/common.py index dbecd9cd..e1725fae 100644 --- a/forum/settings/common.py +++ b/forum/settings/common.py @@ -44,3 +44,19 @@ def plugin_settings(settings: Any) -> None: # Timezone-awareness is required for mysql fields settings.USE_TZ = getattr(settings, "USE_TZ", True) + + # AI Moderation settings + settings.AI_MODERATION_API_URL = getattr(settings, "AI_MODERATION_API_URL", None) + settings.AI_MODERATION_CLIENT_ID = getattr( + settings, "AI_MODERATION_CLIENT_ID", None + ) + settings.AI_MODERATION_SYSTEM_MESSAGE = getattr( + settings, "AI_MODERATION_SYSTEM_MESSAGE", None + ) + settings.AI_MODERATION_CONNECTION_TIMEOUT = getattr( + settings, "AI_MODERATION_CONNECTION_TIMEOUT", 30 + ) + settings.AI_MODERATION_READ_TIMEOUT = getattr( + settings, "AI_MODERATION_READ_TIMEOUT", 30 + ) + settings.AI_MODERATION_USER_ID = getattr(settings, "AI_MODERATION_USER_ID", None) diff --git a/tests/test_ai_moderation.py b/tests/test_ai_moderation.py index 508e6f2c..7c7a0f4a 100644 --- a/tests/test_ai_moderation.py +++ b/tests/test_ai_moderation.py @@ -7,7 +7,7 @@ import pytest from django.contrib.auth import get_user_model -from forum.ai_moderation import AIModerationService, moderate_and_flag_spam +from forum.ai_moderation import moderate_and_flag_spam from forum.backends.mysql.models import ModerationAuditLog from forum.utils import ForumV2RequestError @@ -65,14 +65,6 @@ def mock_waffle_flags() -> Any: yield {"enabled": mock_enabled, "auto_delete": mock_auto_delete} -@pytest.fixture -def ai_service( - mock_ai_moderation_settings: Any, # pylint: disable=redefined-outer-name,unused-argument -) -> AIModerationService: - """Create an AI moderation service instance.""" - return AIModerationService() # type: ignore[no-untyped-call] - - @pytest.fixture def sample_thread_content() -> dict[str, Any]: """Create sample thread content for testing.""" @@ -106,7 +98,7 @@ class TestAIModerationAutoDelete: # pylint: disable=redefined-outer-name,unused def test_auto_delete_triggered_when_enabled( self, - ai_service: AIModerationService, + mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_thread_content: dict[str, Any], ) -> None: @@ -122,11 +114,11 @@ def test_auto_delete_triggered_when_enabled( backend = Mock() - with patch("requests.post", return_value=mock_response), patch.object( - ai_service, "_delete_content" - ) as mock_delete: + with patch("requests.post", return_value=mock_response), patch( + "forum.ai_moderation.delete_thread" + ) as mock_delete_thread: - result = ai_service.moderate_and_flag_content( + result = moderate_and_flag_spam( "spam content", sample_thread_content, course_id="course-v1:edX+DemoX+Demo", @@ -134,7 +126,7 @@ def test_auto_delete_triggered_when_enabled( ) # Verify auto-delete was called - mock_delete.assert_called_once_with(sample_thread_content) + mock_delete_thread.assert_called_once() # Verify actions_taken includes both flagged and soft_deleted assert "flagged" in result["actions_taken"] @@ -143,7 +135,7 @@ def test_auto_delete_triggered_when_enabled( def test_auto_delete_not_triggered_when_disabled( self, - ai_service: AIModerationService, + mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_thread_content: dict[str, Any], ) -> None: @@ -162,11 +154,11 @@ def test_auto_delete_not_triggered_when_disabled( backend = Mock() - with patch("requests.post", return_value=mock_response), patch.object( - ai_service, "_delete_content" - ) as mock_delete: + with patch("requests.post", return_value=mock_response), patch( + "forum.ai_moderation.delete_thread" + ) as mock_delete_thread: - result = ai_service.moderate_and_flag_content( + result = moderate_and_flag_spam( "spam content", sample_thread_content, course_id="course-v1:edX+DemoX+Demo", @@ -174,7 +166,7 @@ def test_auto_delete_not_triggered_when_disabled( ) # Verify auto-delete was NOT called - mock_delete.assert_not_called() + mock_delete_thread.assert_not_called() # Verify actions_taken includes only flagged assert "flagged" in result["actions_taken"] @@ -183,7 +175,7 @@ def test_auto_delete_not_triggered_when_disabled( def test_auto_delete_not_triggered_for_non_spam( self, - ai_service: AIModerationService, + mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_thread_content: dict[str, Any], ) -> None: @@ -201,11 +193,11 @@ def test_auto_delete_not_triggered_for_non_spam( backend = Mock() - with patch("requests.post", return_value=mock_response), patch.object( - ai_service, "_delete_content" - ) as mock_delete: + with patch("requests.post", return_value=mock_response), patch( + "forum.ai_moderation.delete_thread" + ) as mock_delete_thread: - result = ai_service.moderate_and_flag_content( + result = moderate_and_flag_spam( "legitimate content", sample_thread_content, course_id="course-v1:edX+DemoX+Demo", @@ -213,7 +205,7 @@ def test_auto_delete_not_triggered_for_non_spam( ) # Verify auto-delete was NOT called - mock_delete.assert_not_called() + mock_delete_thread.assert_not_called() # Verify no actions taken assert result["actions_taken"] == ["no_action"] @@ -221,7 +213,7 @@ def test_auto_delete_not_triggered_for_non_spam( def test_actions_taken_reflects_flagged_only_when_delete_disabled( self, - ai_service: AIModerationService, + mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_comment_content: dict[str, Any], ) -> None: @@ -240,7 +232,7 @@ def test_actions_taken_reflects_flagged_only_when_delete_disabled( backend = Mock() with patch("requests.post", return_value=mock_response): - result = ai_service.moderate_and_flag_content( + result = moderate_and_flag_spam( "spam content", sample_comment_content, course_id="course-v1:edX+DemoX+Demo", @@ -252,7 +244,7 @@ def test_actions_taken_reflects_flagged_only_when_delete_disabled( def test_actions_taken_reflects_both_when_delete_enabled( self, - ai_service: AIModerationService, + mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_comment_content: dict[str, Any], ) -> None: @@ -267,11 +259,11 @@ def test_actions_taken_reflects_both_when_delete_enabled( backend = Mock() - with patch("requests.post", return_value=mock_response), patch.object( - ai_service, "_delete_content" + with patch("requests.post", return_value=mock_response), patch( + "forum.ai_moderation.delete_comment" ): - result = ai_service.moderate_and_flag_content( + result = moderate_and_flag_spam( "spam content", sample_comment_content, course_id="course-v1:edX+DemoX+Demo", @@ -288,7 +280,7 @@ class TestAIModerationErrorHandling: # pylint: disable=redefined-outer-name,unu def test_deletion_failure_after_successful_flagging( self, - ai_service: AIModerationService, + mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_thread_content: dict[str, Any], ) -> None: @@ -303,13 +295,12 @@ def test_deletion_failure_after_successful_flagging( backend = Mock() - with patch("requests.post", return_value=mock_response), patch.object( - ai_service, - "_delete_content", + with patch("requests.post", return_value=mock_response), patch( + "forum.ai_moderation.delete_thread", side_effect=ForumV2RequestError("Delete failed"), ): - result = ai_service.moderate_and_flag_content( + result = moderate_and_flag_spam( "spam content", sample_thread_content, course_id="course-v1:edX+DemoX+Demo", @@ -324,7 +315,7 @@ def test_deletion_failure_after_successful_flagging( def test_flagging_failure_prevents_deletion( self, - ai_service: AIModerationService, + mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_thread_content: dict[str, Any], ) -> None: @@ -340,11 +331,11 @@ def test_flagging_failure_prevents_deletion( backend = Mock() backend.flag_content_as_spam.side_effect = ValueError("Flag failed") - with patch("requests.post", return_value=mock_response), patch.object( - ai_service, "_delete_content" - ) as mock_delete: + with patch("requests.post", return_value=mock_response), patch( + "forum.ai_moderation.delete_thread" + ) as mock_delete_thread: - result = ai_service.moderate_and_flag_content( + result = moderate_and_flag_spam( "spam content", sample_thread_content, course_id="course-v1:edX+DemoX+Demo", @@ -352,57 +343,10 @@ def test_flagging_failure_prevents_deletion( ) # Delete should not be called if flagging fails - mock_delete.assert_not_called() + mock_delete_thread.assert_not_called() assert result["actions_taken"] == ["no_action"] -class TestDeleteContentMethod: # pylint: disable=redefined-outer-name,protected-access - """Tests for the _delete_content method.""" - - def test_delete_thread_calls_api_correctly( - self, - ai_service: AIModerationService, - sample_thread_content: dict[str, Any], - ) -> None: - """Test that deleting a thread calls the API layer correctly.""" - with patch("forum.api.threads.delete_thread") as mock_delete_thread: - ai_service._delete_content(sample_thread_content) - - mock_delete_thread.assert_called_once_with( - "thread123", - course_id="course-v1:edX+DemoX+Demo", - deleted_by="999", - ) - - def test_delete_comment_calls_api_correctly( - self, - ai_service: AIModerationService, - sample_comment_content: dict[str, Any], - ) -> None: - """Test that deleting a comment calls the API layer correctly.""" - with patch("forum.api.comments.delete_comment") as mock_delete_comment: - ai_service._delete_content(sample_comment_content) - - mock_delete_comment.assert_called_once_with( - "comment456", - course_id="course-v1:edX+DemoX+Demo", - deleted_by="999", - ) - - def test_delete_handles_api_errors( - self, - ai_service: AIModerationService, - sample_thread_content: dict[str, Any], - ) -> None: - """Test that deletion errors propagate to caller.""" - with patch("forum.api.threads.delete_thread") as mock_delete_thread: - mock_delete_thread.side_effect = ForumV2RequestError("API Error") - - # Should raise exception to caller - with pytest.raises(ForumV2RequestError): - ai_service._delete_content(sample_thread_content) - - class TestModerateAndFlagSpamFunction: # pylint: disable=redefined-outer-name """Tests for the module-level moderate_and_flag_spam function.""" @@ -422,12 +366,10 @@ def test_moderate_and_flag_spam_with_auto_delete( # pylint: disable=unused-argu ] backend = Mock() - # Create instance with mocked settings already active - test_service: AIModerationService = AIModerationService() # type: ignore[no-untyped-call] with patch("requests.post", return_value=mock_response), patch( "forum.api.threads.delete_thread" - ), patch("forum.ai_moderation.ai_moderation_service", test_service): + ): result = moderate_and_flag_spam( "spam content", @@ -446,7 +388,6 @@ class TestAuditLogging: # pylint: disable=redefined-outer-name,unused-argument def test_audit_log_created_for_auto_deleted_content( self, - ai_service: AIModerationService, mock_ai_moderation_settings: Any, mock_waffle_flags: dict[str, Mock], sample_thread_content: dict[str, Any], @@ -468,7 +409,7 @@ def test_audit_log_created_for_auto_deleted_content( "forum.api.threads.delete_thread" ): - ai_service.moderate_and_flag_content( + moderate_and_flag_spam( "spam content", sample_thread_content, course_id="course-v1:edX+DemoX+Demo",