Only enforce group membership when updating annos on behalf of a user

We can't enforce the group based on the current user when that user is an
admin.

h/services/annotation_write.py

@@ -78,11 +78,13 @@ def create_annotation(self, data: dict) -> Annotation:
         return annotation
 
     def update_annotation(
+        # pylint: disable=too-many-arguments
         self,
         annotation: Annotation,
         data: dict,
         update_timestamp: bool = True,
         reindex_tag: str = "storage.update_annotation",
+        enforce_write_permission: bool = True,
     ) -> Annotation:
         """
         Update an annotation and its associated document metadata.
@@ -93,6 +95,8 @@ def update_annotation(
             the annotation.
         :param reindex_tag: Tag used by the reindexing job to identify the
             source of the reindexing request.
+        :param enforce_write_permission: Check that the user has permissions
+            to write to the group the annotation is in
         """
         initial_target_uri = annotation.target_uri
 
@@ -104,7 +108,9 @@ def update_annotation(
         # instead of the one which was present when we loaded the model
         # https://docs.sqlalchemy.org/en/13/faq/sessions.html#i-set-the-foo-id-attribute-on-my-instance-to-7-but-the-foo-attribute-is-still-none-shouldn-t-it-have-loaded-foo-with-id-7
         self._db.expire(annotation, ["group"])
-        self._validate_group(annotation)
+        self._validate_group(
+            annotation, enforce_write_permission=enforce_write_permission
+        )
 
         if (
             document := data.get("document", {})
@@ -142,16 +148,15 @@ def _update_annotation_values(annotation: Annotation, data: dict):
         extra = data.get("extra", {})
         annotation.extra.update(extra)
 
-    def _validate_group(self, annotation: Annotation):
+    def _validate_group(self, annotation: Annotation, enforce_write_permission=True):
         group = annotation.group
         if not group:
             raise ValidationError(
                 "group: " + _(f"Invalid group id {annotation.groupid}")
             )
 
-        # The user must have permission to create an annotation in the group
-        # they've asked to create one in.
-        if not self._has_permission(
+        # The user must have permission to write to the group
+        if enforce_write_permission and not self._has_permission(
             Permission.Group.WRITE, context=GroupContext(annotation.group)
         ):
             raise ValidationError(

h/services/url_migration.py

@@ -79,6 +79,10 @@ def move_annotations(self, annotation_ids, current_uri, new_url_info):
                 # Don't update "edited" timestamp on annotation cards.
                 update_timestamp=False,
                 reindex_tag="URLMigrationService.move_annotations",
+                # This action is taken by the admin user most of the time, who
+                # will not have write permission in the relevant group, so we
+                # disable the check
+                enforce_write_permission=False,
             )
 
             log.info("Moved annotation %s to URL %s", ann.uuid, ann.target_uri)

tests/h/services/annotation_write_test.py

@@ -95,9 +95,12 @@ def test_update_annotation(
                 },
             },
             update_timestamp=True,
+            enforce_write_permission=sentinel.enforce_write_permission,
         )
 
-        _validate_group.assert_called_once_with(annotation)
+        _validate_group.assert_called_once_with(
+            annotation, enforce_write_permission=sentinel.enforce_write_permission
+        )
         update_document_metadata.assert_called_once_with(
             db_session,
             result.target_uri,
@@ -146,6 +149,16 @@ def test__validate_group_with_no_permission(self, svc, annotation, has_permissio
             Permission.Group.WRITE, context=GroupContext(annotation.group)
         )
 
+    def test__validate_group_with_no_permission_and_checking_disabled(
+        self, svc, annotation, has_permission
+    ):
+        has_permission.return_value = False
+
+        # pylint: disable=protected-access
+        svc._validate_group(annotation, enforce_write_permission=False)
+
+        has_permission.assert_not_called()
+
     @pytest.mark.parametrize("enforce_scope", (True, False))
     @pytest.mark.parametrize("matching_scope", (True, False))
     @pytest.mark.parametrize("has_scopes", (True, False))

tests/h/services/url_migration_test.py

@@ -57,6 +57,7 @@ def test_move_annotations_updates_urls(
                 data={"target_uri": "https://example.org"},
                 update_timestamp=False,
                 reindex_tag="URLMigrationService.move_annotations",
+                enforce_write_permission=False,
             )
 
     def test_move_annotations_updates_selectors(
@@ -95,6 +96,7 @@ def test_move_annotations_updates_selectors(
             },
             update_timestamp=False,
             reindex_tag="URLMigrationService.move_annotations",
+            enforce_write_permission=False,
         )
 
     def test_move_annotations_updates_documents(
@@ -123,6 +125,7 @@ def test_move_annotations_updates_documents(
             },
             update_timestamp=False,
             reindex_tag="URLMigrationService.move_annotations",
+            enforce_write_permission=False,
         )
 
     def test_move_annotations_by_url_moves_matching_annotations(
@@ -149,10 +152,11 @@ def test_move_annotations_by_url_moves_matching_annotations(
 
         # First annotation should be moved synchronously.
         annotation_write_service.update_annotation.assert_called_once_with(
-            anns[0],
-            {"target_uri": "https://example.org"},
+            annotation=anns[0],
+            data={"target_uri": "https://example.org"},
             update_timestamp=False,
             reindex_tag="URLMigrationService.move_annotations",
+            enforce_write_permission=False,
         )
         pyramid_request.tm.commit.assert_called_once()
 

tests/h/views/api/bulk/_ndjson_test.py

@@ -34,6 +34,7 @@ def test_it_captures_initial_errors(self):
         def failing_method():
             raise ValueError("Oh no!")
 
+            # pragma: nocover
             yield 1  # pylint: disable=unreachable
 
         with pytest.raises(ValueError):