Skip to content

Commit

Permalink
Remove the storage method for creating annotations
Browse files Browse the repository at this point in the history
  • Loading branch information
Jon Betts committed May 4, 2023
1 parent da0f308 commit 85e6b8d
Show file tree
Hide file tree
Showing 2 changed files with 1 addition and 201 deletions.
69 changes: 0 additions & 69 deletions h/storage.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,81 +21,12 @@

from h import models, schemas
from h.models.document import update_document_metadata
from h.security import Permission
from h.services.annotation_read import AnnotationReadService
from h.traversal.group import GroupContext
from h.util.group_scope import url_in_scope
from h.util.uri import normalize as normalize_uri

_ = i18n.TranslationStringFactory(__package__)


def create_annotation(request, data):
"""
Create an annotation from already-validated data.
:param request: the request object
:param data: an annotation data dict that has already been validated by
:py:class:`h.schemas.annotation.CreateAnnotationSchema`
:returns: the created and flushed annotation
"""
document_data = data.pop("document", {})

annotation_read: AnnotationReadService = request.find_service(AnnotationReadService)

# Replies must have the same group as their parent.
if data["references"]:
root_annotation_id = data["references"][0]

if root_annotation := annotation_read.get_annotation_by_id(root_annotation_id):
data["groupid"] = root_annotation.groupid
else:
raise schemas.ValidationError(
"references.0: "
+ _("Annotation {id} does not exist").format(id=root_annotation_id)
)

# Create the annotation and enable relationship loading so we can access
# the group, even though we've not added this to the session yet
annotation = models.Annotation(**data)
request.db.enable_relationship_loading(annotation)

group = annotation.group
if not group:
raise schemas.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 request.has_permission(Permission.Group.WRITE, context=GroupContext(group)):
raise schemas.ValidationError(
"group: " + _("You may not create annotations in the specified group!")
)

_validate_group_scope(group, data["target_uri"])

annotation.created = annotation.updated = datetime.utcnow()
annotation.document = update_document_metadata(
request.db,
annotation.target_uri,
document_data["document_meta_dicts"],
document_data["document_uri_dicts"],
created=annotation.created,
updated=annotation.updated,
)

request.db.add(annotation)
request.db.flush()

request.find_service( # pylint: disable=protected-access
name="search_index"
)._queue.add_by_id(annotation.id, tag="storage.create_annotation", schedule_in=60)

return annotation


def update_annotation(
request,
id_,
Expand Down
133 changes: 1 addition & 132 deletions tests/h/storage_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,16 +1,13 @@
from datetime import datetime as datetime_
from datetime import timedelta
from unittest.mock import create_autospec, sentinel
from unittest.mock import sentinel

import pytest
import sqlalchemy as sa
from h_matchers import Any

from h import storage
from h.models.document import Document, DocumentURI
from h.schemas import ValidationError
from h.security import Permission
from h.traversal.group import GroupContext

pytestmark = pytest.mark.usefixtures("search_index")

Expand Down Expand Up @@ -87,134 +84,6 @@ def test_expand_uri_document_uris(self, db_session, normalized, expected_uris):
assert uris == expected_uris


@pytest.mark.usefixtures("annotation_read_service")
class TestCreateAnnotation:
def test_it(self, pyramid_request, annotation_data, datetime):
annotation = storage.create_annotation(pyramid_request, annotation_data)

for param, value in annotation_data.items():
assert getattr(annotation, param) == value

assert annotation.created == datetime.utcnow.return_value
assert annotation.updated == datetime.utcnow.return_value

assert sa.inspect(annotation).persistent # We saved it to the DB

def test_it_validates_the_group_scope(
self, pyramid_request, annotation_data, group, _validate_group_scope
):
storage.create_annotation(pyramid_request, annotation_data)

_validate_group_scope.assert_called_once_with(
group, annotation_data["target_uri"]
)

def test_it_adds_document_metadata(
self, pyramid_request, annotation_data, update_document_metadata, datetime
):
annotation_data["document"] = {
"document_meta_dicts": sentinel.document_meta_dicts,
"document_uri_dicts": sentinel.document_uri_dicts,
}

annotation = storage.create_annotation(pyramid_request, annotation_data)

update_document_metadata.assert_called_once_with(
pyramid_request.db,
annotation_data["target_uri"],
sentinel.document_meta_dicts,
sentinel.document_uri_dicts,
created=datetime.utcnow.return_value,
updated=datetime.utcnow.return_value,
)
assert annotation.document == update_document_metadata.return_value

def test_it_queues_the_search_index(
self, pyramid_request, annotation_data, search_index
):
annotation = storage.create_annotation(pyramid_request, annotation_data)

search_index._queue.add_by_id.assert_called_once_with( # pylint:disable=protected-access
annotation.id, tag="storage.create_annotation", schedule_in=60
)

def test_it_sets_the_group_to_match_the_parent_for_replies(
self,
pyramid_request,
annotation_data,
factories,
other_group,
annotation_read_service,
):
parent_annotation = factories.Annotation(group=other_group)
annotation_data["references"] = [parent_annotation.id]
annotation_read_service.get_annotation_by_id.return_value = parent_annotation

annotation = storage.create_annotation(pyramid_request, annotation_data)

assert annotation.groupid
assert annotation.group == parent_annotation.group

def test_it_raises_if_parent_annotation_does_not_exist(
self, pyramid_request, annotation_data, annotation_read_service
):
annotation_read_service.get_annotation_by_id.return_value = None

annotation_data["references"] = ["MISSING_ID"]

with pytest.raises(ValidationError):
storage.create_annotation(pyramid_request, annotation_data)

def test_it_raises_if_the_group_doesnt_exist(
self, pyramid_request, annotation_data
):
annotation_data["groupid"] = "MISSING_ID"

with pytest.raises(ValidationError):
storage.create_annotation(pyramid_request, annotation_data)

def test_it_raises_if_write_permission_is_missing(
self, pyramid_request, annotation_data, has_permission
):
has_permission.return_value = False

with pytest.raises(ValidationError):
storage.create_annotation(pyramid_request, annotation_data)

has_permission.assert_called_once_with(
Permission.Group.WRITE, context=Any.instance_of(GroupContext)
)

def test_it_does_not_crash_if_target_selectors_is_empty(
self, pyramid_request, annotation_data
):
# Page notes have [] for target_selectors.
annotation_data["target_selectors"] = []

storage.create_annotation(pyramid_request, annotation_data)

@pytest.mark.xfail(reason="This test passed before due to over fixturing")
def test_it_does_not_crash_if_no_text_or_tags(
self, pyramid_request, annotation_data
):
# Highlights have no text or tags.
annotation_data["text"] = annotation_data["tags"] = ""

# ValueError: Attribute 'tags' does not accept objects of type <class 'str'>
# So what should this be? None?
storage.create_annotation(pyramid_request, annotation_data)

@pytest.fixture
def other_group(self, factories):
# Set an authority_provided_id so our group_id is not None
return factories.OpenGroup(authority_provided_id="other_group_auth_id")

@pytest.fixture
def has_permission(self, pyramid_request):
pyramid_request.has_permission = create_autospec(pyramid_request.has_permission)
return pyramid_request.has_permission


class TestUpdateAnnotation:
def test_it(self, pyramid_request, annotation, annotation_data, datetime):
result = storage.update_annotation(
Expand Down

0 comments on commit 85e6b8d

Please sign in to comment.