diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 03cfaa8f..7f3924f8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,6 +18,7 @@ on: # 👇 Top-level permissions: safe defaults permissions: contents: read + pull-requests: write jobs: # ------------------------------- diff --git a/app/reviewer/settings.py b/app/reviewer/settings.py index ee77d371..3cff8e97 100644 --- a/app/reviewer/settings.py +++ b/app/reviewer/settings.py @@ -126,6 +126,24 @@ PYWIKIBOT_SITE_FAMILY = os.getenv("PYWIKIBOT_SITE_FAMILY", "wikipedia") +# Revert detection configuration +# Enable/disable revert detection for already-reviewed edits +ENABLE_REVERT_DETECTION = os.getenv("ENABLE_REVERT_DETECTION", "True").lower() in ( + "true", + "1", + "yes", +) + +# Pending changes approval configuration +# Enable/disable dry-run mode for pending changes approval +# When True, only allows approvals on test pages (Merkityt_versiot_-kokeilu/*) +# When False, allows approvals on all pages +PENDING_CHANGES_DRY_RUN = os.getenv("PENDING_CHANGES_DRY_RUN", "True").lower() in ( + "true", + "1", + "yes", +) + # ORES model thresholds (global defaults, per-wiki config takes precedence) ORES_DAMAGING_THRESHOLD = float(os.getenv("ORES_DAMAGING_THRESHOLD", "0.3")) ORES_GOODFAITH_THRESHOLD = float(os.getenv("ORES_GOODFAITH_THRESHOLD", "0.7")) diff --git a/app/reviews/autoreview/__init__.py b/app/reviews/autoreview/__init__.py index 9d48db4f..477a675d 100644 --- a/app/reviews/autoreview/__init__.py +++ b/app/reviews/autoreview/__init__.py @@ -1 +1,32 @@ from __future__ import annotations + +# Backwards-compatibility exports for older test imports +from pywikibot.data.superset import SupersetQuery # re-export for tests + +from .checks.revert_detection import ( + _find_reviewed_revisions_by_sha1, + _parse_revert_params, + check_revert_detection, +) +from .context import CheckContext + + +def _check_revert_detection(revision, client): + """Compatibility wrapper matching legacy signature used in tests.""" + context = CheckContext( + revision=revision, + client=client, + profile=None, + auto_groups={}, + blocking_categories={}, + redirect_aliases=[], + ) + return check_revert_detection(context) + + +__all__ = [ + "SupersetQuery", + "_check_revert_detection", + "_find_reviewed_revisions_by_sha1", + "_parse_revert_params", +] diff --git a/app/reviews/autoreview/checks/revert_detection.py b/app/reviews/autoreview/checks/revert_detection.py new file mode 100644 index 00000000..3cef6c3b --- /dev/null +++ b/app/reviews/autoreview/checks/revert_detection.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import json +import logging +from typing import Any + +from django.conf import settings + +import reviews.autoreview as autoreview + +from ..context import CheckContext + +""" +Revert detection check for already-reviewed edits. + +This check detects when a pending edit is a revert to previously reviewed content +by matching SHA1 content hashes and checking for revert tags. +""" + +logger = logging.getLogger(__name__) + + +def check_revert_detection(context: CheckContext) -> dict[str, Any]: + """ + Check if a revision is a revert to previously reviewed content. + + Args: + context: CheckContext containing revision and related data + + Returns: + Dict with check result including status, message, and metadata + """ + # Check if revert detection is enabled + if not getattr(settings, "ENABLE_REVERT_DETECTION", True): + return {"status": "skip", "message": "Revert detection is disabled", "metadata": {}} + + revision = context.revision + page = revision.page + + # Check for revert tags + revert_tags = {"mw-manual-revert", "mw-reverted", "mw-rollback", "mw-undo"} + change_tags = getattr(revision, "change_tags", []) + + if not any(tag in change_tags for tag in revert_tags): + return { + "status": "skip", + "message": "No revert tags found", + "metadata": {"change_tags": change_tags}, + } + + # Parse change tag parameters to get reverted revision IDs + reverted_rev_ids = _parse_revert_params(revision) + if not reverted_rev_ids: + return { + "status": "skip", + "message": "No reverted revision IDs found in change tags", + "metadata": {"change_tags": change_tags}, + } + + # Check if any of the reverted revisions were previously reviewed + reviewed_revisions = autoreview._find_reviewed_revisions_by_sha1( + context.client, page, reverted_rev_ids + ) + + if reviewed_revisions: + return { + "status": "approve", + "message": ( + f"Revert to previously reviewed content (SHA1: {reviewed_revisions[0]['sha1']})" + ), + "metadata": { + "reverted_rev_ids": reverted_rev_ids, + "reviewed_revisions": reviewed_revisions, + "revert_tags": [tag for tag in change_tags if tag in revert_tags], + }, + } + + return { + "status": "block", + "message": "Revert detected but no previously reviewed content found", + "metadata": { + "reverted_rev_ids": reverted_rev_ids, + "revert_tags": [tag for tag in change_tags if tag in revert_tags], + }, + } + + +def _parse_revert_params(revision) -> list[int]: + """ + Parse change tag parameters to extract reverted revision IDs. + + Args: + revision: PendingRevision object + + Returns: + List of reverted revision IDs + """ + try: + # Get change tag parameters from revision + change_tag_params = getattr(revision, "change_tag_params", []) + if not change_tag_params: + return [] + + reverted_ids: list[int] = [] + + for param_str in change_tag_params: + try: + # Parse JSON parameter + param_data = json.loads(param_str) + + # Extract reverted revision IDs + if "oldestRevertedRevId" in param_data: + reverted_ids.append(param_data["oldestRevertedRevId"]) + if "newestRevertedRevId" in param_data: + reverted_ids.append(param_data["newestRevertedRevId"]) + if "originalRevisionId" in param_data: + reverted_ids.append(param_data["originalRevisionId"]) + + except (json.JSONDecodeError, KeyError) as e: + logger.warning(f"Failed to parse change tag param: {param_str}, error: {e}") + continue + + return list(set(reverted_ids)) # Remove duplicates + + except Exception as e: + logger.error(f"Error parsing revert params for revision {revision.revid}: {e}") + return [] + + +def _find_reviewed_revisions_by_sha1(client, page, reverted_rev_ids: list[int]) -> list[dict]: + """ + Find previously reviewed revisions by SHA1 content hash. + + This implements @zache-fi's suggested Superset approach: + 1. Query MediaWiki database for older reviewed versions by SHA1 + 2. Check if any of the reverted revisions were previously reviewed + + Args: + client: WikiClient instance + page: PendingPage object + reverted_rev_ids: List of reverted revision IDs + + Returns: + List of reviewed revision data + """ + if not reverted_rev_ids: + return [] + + try: + # Execute Superset query to find reviewed revisions by SHA1 + # This follows @zache-fi's suggested SQL approach + revid_list = ",".join(str(int(revid)) for revid in reverted_rev_ids) + + # ids are validated as integers above; safe to embed + sql_query = ( + "SELECT \n" + " MAX(rev_id) as max_reviewable_rev_id_by_sha1, \n" + " rev_page, \n" + " content_sha1, \n" + " MAX(fr_rev_id) as max_old_reviewed_id \n" + "FROM \n" + " revision \n" + " LEFT JOIN flaggedrevs ON rev_id=fr_rev_id\n" + " JOIN slots ON slot_revision_id=rev_id\n" + " JOIN content ON slot_content_id=content_id\n" + "WHERE \n" + f" rev_id IN ({revid_list})" # noqa: S608 + "\nGROUP BY \n" + " rev_page, content_sha1\n" + ) + + # Execute query using SupersetQuery (resolved through package for test patching) + superset = autoreview.SupersetQuery(site=client.site) + results = superset.query(sql_query) + + # Filter results where content was previously reviewed + reviewed_revisions: list[dict] = [] + for result in results: + if result.get("max_old_reviewed_id") is not None: + reviewed_revisions.append( + { + "sha1": result.get("content_sha1"), + "max_reviewed_id": result.get("max_old_reviewed_id"), + "max_reviewable_id": result.get("max_reviewable_rev_id_by_sha1"), + "page_id": result.get("rev_page"), + } + ) + + return reviewed_revisions + + except Exception as e: + logger.error(f"Error finding reviewed revisions for page {page.pageid}: {e}") + return [] diff --git a/app/reviews/management/__init__.py b/app/reviews/management/__init__.py index e69de29b..652875d2 100644 --- a/app/reviews/management/__init__.py +++ b/app/reviews/management/__init__.py @@ -0,0 +1 @@ +# Management commands for reviews app diff --git a/app/reviews/management/commands/__init__.py b/app/reviews/management/commands/__init__.py index e69de29b..2c1c7c15 100644 --- a/app/reviews/management/commands/__init__.py +++ b/app/reviews/management/commands/__init__.py @@ -0,0 +1 @@ +# Management commands diff --git a/app/reviews/management/commands/test_pending_changes_review.py b/app/reviews/management/commands/test_pending_changes_review.py new file mode 100644 index 00000000..bbb14d6e --- /dev/null +++ b/app/reviews/management/commands/test_pending_changes_review.py @@ -0,0 +1,93 @@ +""" +Django management command to test pending changes review functionality. + +This command allows testing the approve_revision() utility function +with various parameters and configurations. +""" + +from django.conf import settings +from django.core.management.base import BaseCommand, CommandError + +from reviews.utils.approval import approve_revision + + +class Command(BaseCommand): + help = "Test pending changes review functionality (approve/unapprove revisions)" + + def add_arguments(self, parser): + parser.add_argument( + "--revid", type=int, required=True, help="Revision ID to approve/unapprove" + ) + parser.add_argument( + "--comment", + type=str, + default="Test approval via management command", + help='Comment for the review (default: "Test approval via management command")', + ) + parser.add_argument( + "--unapprove", + action="store_true", + help="Unapprove the revision instead of approving it", + ) + parser.add_argument("--value", type=int, help="Flag value for the review (optional)") + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would happen without making actual changes", + ) + + def handle(self, *args, **options): + revid = options["revid"] + comment = options["comment"] + unapprove = options["unapprove"] + value = options["value"] + dry_run = options["dry_run"] + + # Display current configuration + current_dry_run = getattr(settings, "PENDING_CHANGES_DRY_RUN", True) + self.stdout.write(f"Current PENDING_CHANGES_DRY_RUN setting: {current_dry_run}") + + if dry_run: + self.stdout.write(self.style.WARNING("DRY-RUN MODE: No actual changes will be made")) + + # Display operation details + operation = "unapprove" if unapprove else "approve" + self.stdout.write(f"Operation: {operation}") + self.stdout.write(f"Revision ID: {revid}") + self.stdout.write(f"Comment: {comment}") + if value is not None: + self.stdout.write(f"Value: {value}") + + try: + # Call the approve_revision function + result = approve_revision( + revid=revid, comment=comment, value=value, unapprove=unapprove + ) + + # Display results + if result["result"] == "success": + if result.get("dry_run", False): + self.stdout.write(self.style.SUCCESS(f"✅ {result['message']}")) + else: + self.stdout.write(self.style.SUCCESS(f"✅ {result['message']}")) + else: + self.stdout.write(self.style.ERROR(f"❌ {result['message']}")) + + # Display additional information + if "api_response" in result: + self.stdout.write(f"API Response: {result['api_response']}") + + # Display dry-run information + if result.get("dry_run", False): + self.stdout.write( + self.style.WARNING( + "ℹī¸ This was a dry-run operation. " + "Set PENDING_CHANGES_DRY_RUN=False to make actual changes." + ) + ) + + except Exception as e: + self.stdout.write(self.style.ERROR(f"❌ Error: {str(e)}")) + raise CommandError(f"Failed to {operation} revision {revid}: {str(e)}") + + self.stdout.write(self.style.SUCCESS("✅ Command completed successfully")) diff --git a/app/reviews/services/wiki_client.py b/app/reviews/services/wiki_client.py index 150885ec..cb18865f 100644 --- a/app/reviews/services/wiki_client.py +++ b/app/reviews/services/wiki_client.py @@ -152,6 +152,7 @@ def fetch_pending_pages(self, limit: int = 10000) -> list[PendingPage]: a.actor_name, a.actor_user, group_concat(DISTINCT(ctd_name)) AS change_tags, + group_concat(DISTINCT(ct_params)) AS change_tags_params, group_concat(DISTINCT(ug_group)) AS user_groups, group_concat(DISTINCT(ufg_group)) AS user_former_groups, group_concat(DISTINCT(cl_to)) AS page_categories, diff --git a/app/reviews/tests/test_approval.py b/app/reviews/tests/test_approval.py new file mode 100644 index 00000000..66cc1729 --- /dev/null +++ b/app/reviews/tests/test_approval.py @@ -0,0 +1,253 @@ +""" +Unit tests for the approval utility functions. + +Tests the approve_revision() function and related functionality. +""" + +from unittest.mock import Mock, patch + +from django.conf import settings +from django.test import TestCase, override_settings + +from reviews.utils.approval import _get_page_title_from_revid, approve_revision + + +class ApprovalUtilityTests(TestCase): + """Test cases for the approval utility functions.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_revid = 12345 + self.test_comment = "Test approval" + self.test_site = Mock() + self.test_site.code = "fi" + self.test_site.family.name = "wikipedia" + + @patch("reviews.utils.approval.Site") + @patch("reviews.utils.approval.Request") + @override_settings(PENDING_CHANGES_DRY_RUN=False) + def test_approve_revision_success(self, mock_request_class, mock_site_class): + """Test successful approval of a revision.""" + # Mock the site + mock_site_class.return_value = self.test_site + + # Mock the API request + mock_request = Mock() + mock_request.submit.return_value = {"review": {"result": "success"}} + mock_request_class.return_value = mock_request + + # Call the function + result = approve_revision(revid=self.test_revid, comment=self.test_comment, unapprove=False) + + # Assertions + self.assertEqual(result["result"], "success") + self.assertFalse(result["dry_run"]) + self.assertIn("Successfully approved", result["message"]) + + # Verify API call was made + mock_request_class.assert_called_once() + mock_request.submit.assert_called_once() + + @patch("reviews.utils.approval.Site") + @patch("reviews.utils.approval.Request") + @override_settings(PENDING_CHANGES_DRY_RUN=False) + def test_approve_revision_unapprove(self, mock_request_class, mock_site_class): + """Test successful unapproval of a revision.""" + # Mock the site + mock_site_class.return_value = self.test_site + + # Mock the API request + mock_request = Mock() + mock_request.submit.return_value = {"review": {"result": "success"}} + mock_request_class.return_value = mock_request + + # Call the function + result = approve_revision(revid=self.test_revid, comment=self.test_comment, unapprove=True) + + # Assertions + self.assertEqual(result["result"], "success") + self.assertFalse(result["dry_run"]) + self.assertIn("Successfully unapproved", result["message"]) + + @patch("reviews.utils.approval.Site") + @patch("reviews.utils.approval.Request") + @patch("reviews.utils.approval._get_page_title_from_revid") + @override_settings(PENDING_CHANGES_DRY_RUN=True) + def test_approve_revision_dry_run_production_page( + self, mock_get_title, mock_request_class, mock_site_class + ): + """Test dry-run mode with production page (should skip).""" + # Mock the site + mock_site_class.return_value = self.test_site + + # Mock page title (production page) + mock_get_title.return_value = "Production_Page" + + # Call the function + result = approve_revision(revid=self.test_revid, comment=self.test_comment, unapprove=False) + + # Assertions + self.assertEqual(result["result"], "success") + self.assertTrue(result["dry_run"]) + self.assertIn("DRY-RUN: Would approve", result["message"]) + + # Verify API call was NOT made + mock_request_class.assert_not_called() + + @patch("reviews.utils.approval.Site") + @patch("reviews.utils.approval.Request") + @patch("reviews.utils.approval._get_page_title_from_revid") + @override_settings(PENDING_CHANGES_DRY_RUN=True) + def test_approve_revision_dry_run_test_page( + self, mock_get_title, mock_request_class, mock_site_class + ): + """Test dry-run mode with test page (should proceed).""" + # Mock the site + mock_site_class.return_value = self.test_site + + # Mock page title (test page) + mock_get_title.return_value = "Merkityt_versiot_-kokeilu/Test_Page" + + # Mock the API request + mock_request = Mock() + mock_request.submit.return_value = {"review": {"result": "success"}} + mock_request_class.return_value = mock_request + + # Call the function + result = approve_revision(revid=self.test_revid, comment=self.test_comment, unapprove=False) + + # Assertions + self.assertEqual(result["result"], "success") + self.assertFalse(result["dry_run"]) + self.assertIn("Successfully approved", result["message"]) + + # Verify API call was made + mock_request_class.assert_called_once() + + @patch("reviews.utils.approval.Site") + @patch("reviews.utils.approval.Request") + @override_settings(PENDING_CHANGES_DRY_RUN=False) + def test_approve_revision_with_value(self, mock_request_class, mock_site_class): + """Test approval with custom value parameter.""" + # Mock the site + mock_site_class.return_value = self.test_site + + # Mock the API request + mock_request = Mock() + mock_request.submit.return_value = {"review": {"result": "success"}} + mock_request_class.return_value = mock_request + + # Call the function with value + result = approve_revision( + revid=self.test_revid, comment=self.test_comment, value=1, unapprove=False + ) + + # Assertions + self.assertEqual(result["result"], "success") + + # Verify API call was made with value parameter + mock_request_class.assert_called_once() + call_args = mock_request_class.call_args + self.assertEqual(call_args[1]["value"], "1") + + @patch("reviews.utils.approval.Site") + @patch("reviews.utils.approval.Request") + @override_settings(PENDING_CHANGES_DRY_RUN=False) + def test_approve_revision_api_error(self, mock_request_class, mock_site_class): + """Test handling of API errors.""" + # Mock the site + mock_site_class.return_value = self.test_site + + # Mock API error response + mock_request = Mock() + mock_request.submit.return_value = { + "error": { + "code": "permissiondenied", + "info": "Permission denied", + } + } + mock_request_class.return_value = mock_request + + # Call the function + result = approve_revision(revid=self.test_revid, comment=self.test_comment, unapprove=False) + + # Assertions + self.assertEqual(result["result"], "error") + self.assertIn("Failed to approve", result["message"]) + + @patch("reviews.utils.approval.Site") + @override_settings(PENDING_CHANGES_DRY_RUN=False) + def test_approve_revision_exception(self, mock_site_class): + """Test handling of exceptions.""" + # Mock the site to raise an exception + mock_site_class.side_effect = Exception("Connection error") + + # Call the function + result = approve_revision(revid=self.test_revid, comment=self.test_comment, unapprove=False) + + # Assertions + self.assertEqual(result["result"], "error") + self.assertIn("Error approving", result["message"]) + + @patch("reviews.utils.approval.Request") + def test_get_page_title_from_revid_success(self, mock_request_class): + """Test successful retrieval of page title.""" + # Mock the API request + mock_request = Mock() + mock_request.submit.return_value = { + "query": { + "pages": {"123": {"title": "Test_Page", "revisions": [{"revid": self.test_revid}]}} + } + } + mock_request_class.return_value = mock_request + + # Call the function + title = _get_page_title_from_revid(self.test_site, self.test_revid) + + # Assertions + self.assertEqual(title, "Test_Page") + mock_request_class.assert_called_once() + + @patch("reviews.utils.approval.Request") + def test_get_page_title_from_revid_not_found(self, mock_request_class): + """Test handling when page title is not found.""" + # Mock the API request + mock_request = Mock() + mock_request.submit.return_value = {"query": {"pages": {}}} + mock_request_class.return_value = mock_request + + # Call the function + title = _get_page_title_from_revid(self.test_site, self.test_revid) + + # Assertions + self.assertIsNone(title) + + @patch("reviews.utils.approval.Request") + def test_get_page_title_from_revid_exception(self, mock_request_class): + """Test handling of exceptions in _get_page_title_from_revid.""" + # Mock the API request to raise an exception + mock_request = Mock() + mock_request.submit.side_effect = Exception("API error") + mock_request_class.return_value = mock_request + + # Call the function + title = _get_page_title_from_revid(self.test_site, self.test_revid) + + # Assertions + self.assertIsNone(title) + + +class ApprovalIntegrationTests(TestCase): + """Integration tests for the approval functionality.""" + + @override_settings(PENDING_CHANGES_DRY_RUN=True) + def test_dry_run_setting_respected(self): + """Test that the dry-run setting is properly respected.""" + # This test verifies that the setting is read correctly + self.assertTrue(getattr(settings, "PENDING_CHANGES_DRY_RUN", True)) + + @override_settings(PENDING_CHANGES_DRY_RUN=False) + def test_dry_run_setting_disabled(self): + """Test that the dry-run setting can be disabled.""" + # This test verifies that the setting can be disabled + self.assertFalse(getattr(settings, "PENDING_CHANGES_DRY_RUN", True)) diff --git a/app/reviews/tests/test_revert_detection.py b/app/reviews/tests/test_revert_detection.py new file mode 100644 index 00000000..13c7fd8a --- /dev/null +++ b/app/reviews/tests/test_revert_detection.py @@ -0,0 +1,259 @@ +""" +Tests for revert detection functionality. + +This module tests the revert detection check that identifies when +a pending edit is a revert to previously reviewed content. +""" + +import json +from datetime import timedelta +from unittest.mock import Mock, patch + +from django.test import TestCase +from django.utils import timezone + +from reviews.autoreview import ( + _check_revert_detection, + _find_reviewed_revisions_by_sha1, + _parse_revert_params, +) +from reviews.models import PendingPage, PendingRevision, Wiki, WikiConfiguration +from reviews.services import WikiClient + + +class RevertDetectionTests(TestCase): + """Test cases for revert detection functionality.""" + + def setUp(self): + """Set up test data.""" + self.wiki = Wiki.objects.create( + name="Test Wiki", + code="test", + family="wikipedia", + api_endpoint="https://test.wikipedia.org/w/api.php", + ) + self.config = WikiConfiguration.objects.create(wiki=self.wiki) + + self.page = PendingPage.objects.create( + wiki=self.wiki, + pageid=12345, + title="Test Page", + stable_revid=100, + ) + + self.revision = PendingRevision.objects.create( + page=self.page, + revid=200, + parentid=150, + user_name="TestUser", + user_id=1000, + change_tags=["mw-manual-revert"], + timestamp=timezone.now(), + age_at_fetch=timedelta(seconds=0), + sha1=("0" * 40), + wikitext="Test content", + ) + # Not a model field; attach dynamically for the check logic + self.revision.change_tag_params = [ + json.dumps( + { + "revertId": 200, + "oldestRevertedRevId": 180, + "newestRevertedRevId": 190, + "originalRevisionId": 175, + } + ) + ] + + self.client = Mock(spec=WikiClient) + self.client.site = Mock() + + def test_revert_detection_disabled(self): + """Test that revert detection is skipped when disabled.""" + with self.settings(ENABLE_REVERT_DETECTION=False): + result = _check_revert_detection(self.revision, self.client) + + self.assertEqual(result["status"], "skip") + self.assertEqual(result["message"], "Revert detection is disabled") + + def test_no_revert_tags(self): + """Test that revert detection is skipped when no revert tags are present.""" + self.revision.change_tags = ["mw-edit"] + self.revision.save() + + result = _check_revert_detection(self.revision, self.client) + + self.assertEqual(result["status"], "skip") + self.assertEqual(result["message"], "No revert tags found") + + def test_parse_revert_params(self): + """Test parsing of change tag parameters.""" + reverted_ids = _parse_revert_params(self.revision) + + expected_ids = [180, 190, 175] # From change_tag_params + self.assertEqual(set(reverted_ids), set(expected_ids)) + + def test_parse_revert_params_empty(self): + """Test parsing when no change tag parameters are present.""" + self.revision.change_tag_params = [] + self.revision.save() + + reverted_ids = _parse_revert_params(self.revision) + self.assertEqual(reverted_ids, []) + + def test_parse_revert_params_invalid_json(self): + """Test parsing with invalid JSON in change tag parameters.""" + self.revision.change_tag_params = ["invalid json"] + self.revision.save() + + reverted_ids = _parse_revert_params(self.revision) + self.assertEqual(reverted_ids, []) + + @patch("reviews.autoreview.SupersetQuery") + def test_find_reviewed_revisions_by_sha1_success(self, mock_superset): + """Test finding reviewed revisions by SHA1.""" + # Mock SupersetQuery results + mock_superset.return_value.query.return_value = [ + { + "content_sha1": "abc123", + "max_old_reviewed_id": 150, + "max_reviewable_rev_id_by_sha1": 180, + "rev_page": 12345, + } + ] + + reverted_ids = [180, 190] + reviewed_revisions = _find_reviewed_revisions_by_sha1(self.client, self.page, reverted_ids) + + self.assertEqual(len(reviewed_revisions), 1) + self.assertEqual(reviewed_revisions[0]["sha1"], "abc123") + self.assertEqual(reviewed_revisions[0]["max_reviewed_id"], 150) + + @patch("reviews.autoreview.SupersetQuery") + def test_find_reviewed_revisions_by_sha1_no_results(self, mock_superset): + """Test when no reviewed revisions are found.""" + mock_superset.return_value.query.return_value = [] + + reverted_ids = [180, 190] + reviewed_revisions = _find_reviewed_revisions_by_sha1(self.client, self.page, reverted_ids) + + self.assertEqual(reviewed_revisions, []) + + @patch("reviews.autoreview._find_reviewed_revisions_by_sha1") + def test_revert_detection_approve(self, mock_find_reviewed): + """Test revert detection when revert to reviewed content is found.""" + # Mock finding reviewed revisions + mock_find_reviewed.return_value = [ + {"sha1": "abc123", "max_reviewed_id": 150, "max_reviewable_id": 180, "page_id": 12345} + ] + + result = _check_revert_detection(self.revision, self.client) + + self.assertEqual(result["status"], "approve") + self.assertIn("Revert to previously reviewed content", result["message"]) + self.assertIn("abc123", result["message"]) + + @patch("reviews.autoreview._find_reviewed_revisions_by_sha1") + def test_revert_detection_block(self, mock_find_reviewed): + """Test revert detection when no reviewed content is found.""" + # Mock no reviewed revisions found + mock_find_reviewed.return_value = [] + + result = _check_revert_detection(self.revision, self.client) + + self.assertEqual(result["status"], "block") + self.assertEqual( + result["message"], "Revert detected but no previously reviewed content found" + ) + + def test_revert_detection_no_reverted_ids(self): + """Test revert detection when no reverted revision IDs are found.""" + self.revision.change_tag_params = [] + self.revision.save() + + result = _check_revert_detection(self.revision, self.client) + + self.assertEqual(result["status"], "skip") + self.assertEqual(result["message"], "No reverted revision IDs found in change tags") + + def test_revert_detection_metadata(self): + """Test that revert detection returns proper metadata.""" + with patch("reviews.autoreview._find_reviewed_revisions_by_sha1") as mock_find: + mock_find.return_value = [{"sha1": "abc123"}] + + result = _check_revert_detection(self.revision, self.client) + + self.assertIn("reverted_rev_ids", result["metadata"]) + self.assertIn("revert_tags", result["metadata"]) + self.assertIn("reviewed_revisions", result["metadata"]) + self.assertEqual(result["metadata"]["revert_tags"], ["mw-manual-revert"]) + + +class RevertDetectionIntegrationTests(TestCase): + """Integration tests for revert detection with real data.""" + + def setUp(self): + """Set up integration test data.""" + self.wiki = Wiki.objects.create( + name="Test Wiki", + code="test", + family="wikipedia", + api_endpoint="https://test.wikipedia.org/w/api.php", + ) + self.config = WikiConfiguration.objects.create(wiki=self.wiki) + + def test_revert_detection_with_real_revision(self): + """Test revert detection with a real revision setup.""" + page = PendingPage.objects.create( + wiki=self.wiki, + pageid=12345, + title="Test Page", + stable_revid=100, + ) + + # Create a revision with revert tags + revision = PendingRevision.objects.create( + page=page, + revid=200, + parentid=150, + user_name="TestUser", + user_id=1000, + change_tags=["mw-manual-revert", "mw-reverted"], + timestamp=timezone.now(), + age_at_fetch=timedelta(seconds=0), + sha1=("0" * 40), + wikitext="Test content", + ) + # Attach dynamic params expected by the check logic + revision.change_tag_params = [ + json.dumps( + { + "revertId": 200, + "oldestRevertedRevId": 180, + "newestRevertedRevId": 190, + "originalRevisionId": 175, + } + ) + ] + + # Mock the client + client = Mock(spec=WikiClient) + client.site = Mock() + + # Test with SupersetQuery mock + with patch("reviews.autoreview.SupersetQuery") as mock_superset: + mock_superset.return_value.query.return_value = [ + { + "content_sha1": "test_sha1", + "max_old_reviewed_id": 150, + "max_reviewable_rev_id_by_sha1": 180, + "rev_page": 12345, + } + ] + + result = _check_revert_detection(revision, client) + + self.assertEqual(result["status"], "approve") + self.assertIn("test_sha1", result["message"]) + self.assertEqual(len(result["metadata"]["reverted_rev_ids"]), 3) + self.assertEqual(len(result["metadata"]["revert_tags"]), 2) diff --git a/app/reviews/utils/__init__.py b/app/reviews/utils/__init__.py new file mode 100644 index 00000000..7cff9f71 --- /dev/null +++ b/app/reviews/utils/__init__.py @@ -0,0 +1 @@ +# Utils package for reviews app diff --git a/app/reviews/utils/approval.py b/app/reviews/utils/approval.py new file mode 100644 index 00000000..c73528d1 --- /dev/null +++ b/app/reviews/utils/approval.py @@ -0,0 +1,122 @@ +""" +Utility functions for approving/unapproving pending changes revisions. + +This module provides functions to interact with MediaWiki's FlaggedRevs API +for approving or unapproving pending changes revisions. +""" + +import logging + +from django.conf import settings +from pywikibot import Site +from pywikibot.data.api import Request + +logger = logging.getLogger(__name__) + + +def approve_revision(revid, comment, value=None, unapprove=False): + """ + Approve or unapprove a pending changes revision. + + Args: + revid (int): The revision ID for which to set the flags + comment (str): Comment for the review + value (int, optional): Flag value for the review. Defaults to None. + unapprove (bool, optional): If True, revision will be unapproved + rather than approved. Defaults to False. + + Returns: + dict: Result of the review operation + """ + try: + # Get the site (assuming we're working with Finnish Wikipedia) + site = Site("fi", "wikipedia") + + # Check if we're in dry-run mode + if getattr(settings, "PENDING_CHANGES_DRY_RUN", True): + # Get page title to check if it's in test namespace + page_title = _get_page_title_from_revid(site, revid) + + if page_title and not page_title.startswith("Merkityt_versiot_-kokeilu/"): + action = "unapprove" if unapprove else "approve" + logger.info("DRY-RUN: Would %s revision %s on %s", action, revid, page_title) + return { + "result": "success", + "dry_run": True, + "message": f"DRY-RUN: Would {action} revision {revid}", + } + + # Prepare API request parameters + params = { + "action": "review", + "revid": revid, + "comment": comment, + } + + # Add unapprove parameter if needed + if unapprove: + params["unapprove"] = "1" + + # Add value parameter if provided + if value is not None: + params["value"] = str(value) + + # Make the API request + request = Request(site=site, **params) + result = request.submit() + + # Check if the request was successful + if "review" in result: + action_past = "unapproved" if unapprove else "approved" + logger.info("Successfully %s revision %s", action_past, revid) + return { + "result": "success", + "dry_run": False, + "message": f"Successfully {action_past} revision {revid}", + "api_response": result["review"], + } + else: + action = "unapprove" if unapprove else "approve" + logger.error("Failed to %s revision %s: %s", action, revid, result) + return { + "result": "error", + "dry_run": False, + "message": f"Failed to {action} revision {revid}", + "api_response": result, + } + + except Exception as e: + action_ing = "unapproving" if unapprove else "approving" + logger.error("Error %s revision %s: %s", action_ing, revid, e) + return { + "result": "error", + "dry_run": False, + "message": f"Error {action_ing} revision {revid}: {e}", + } + + +def _get_page_title_from_revid(site, revid): + """ + Get the page title for a given revision ID. + + Args: + site: Pywikibot site object + revid (int): Revision ID + + Returns: + str: Page title or None if not found + """ + try: + request = Request(site=site, action="query", prop="revisions", revids=revid, rvprop="title") + result = request.submit() + + if "query" in result and "pages" in result["query"]: + for page_id, page_data in result["query"]["pages"].items(): + if "revisions" in page_data: + return page_data["title"] + + return None + + except Exception as e: + logger.error("Error getting page title for revision %s: %s", revid, e) + return None