From 4eaf3a7d869cbee52d5b06012834a9035bdc6fbe Mon Sep 17 00:00:00 2001 From: Elim Pizza Date: Mon, 8 Jun 2026 12:04:20 -0300 Subject: [PATCH] Hide annotations behind active checkpoints in search --- h/search/core.py | 1 + h/search/query.py | 64 ++++++++++ h/services/checkpoint.py | 83 ++++++++++++- tests/unit/h/search/conftest.py | 8 ++ tests/unit/h/search/query_test.py | 141 +++++++++++++++++++++++ tests/unit/h/services/checkpoint_test.py | 107 +++++++++++++++++ 6 files changed, 403 insertions(+), 1 deletion(-) diff --git a/h/search/core.py b/h/search/core.py index d4c5dfd5f63..09f80789853 100644 --- a/h/search/core.py +++ b/h/search/core.py @@ -53,6 +53,7 @@ def __init__( query.UserFilter(), query.NIPSAFilter(request), query.GroupAndModerationFilter(request), + query.HideRevealFilter(request), query.AnyMatcher(), query.TagsMatcher(), query.UriCombinedWildcardFilter( diff --git a/h/search/query.py b/h/search/query.py index fc767c7242b..2b4bb20e3b9 100644 --- a/h/search/query.py +++ b/h/search/query.py @@ -8,6 +8,7 @@ from h.models import Group, User from h.search.util import add_default_scheme, wildcard_uri_is_valid from h.security import Identity, Permission, identity_permits +from h.services.checkpoint import CheckpointService from h.traversal import GroupContext from h.util import uri from h.util.uri import build_scope_key, parse_uri_versions @@ -263,6 +264,69 @@ def __call__(self, search, params): return search.filter(Q("bool", should=query_clauses)) +class HideRevealFilter: + """ + Hide annotations covered by an active Hide & Reveal checkpoint. + + Until a checkpoint's reveal_date passes, a student must not see their peers' + annotations on the checkpointed document. The scope is resolved from the + requesting user's own memberships (see CheckpointService.hidden_scopes). An + instructor sees everything, while everyone else sees only their own + annotations, instructor notes, and instructor replies to them. + + This is a no-op for the common case of a user with no active checkpoints. + """ + + def __init__(self, request): + self.userid = request.authenticated_userid + self.checkpoint_service = request.find_service(CheckpointService) + self._user = request.user + + def __call__(self, search, params): # noqa: ARG002 + for scope in self.checkpoint_service.hidden_scopes(self._user): + search = search.exclude(self._hidden_query(scope)) + return search + + def _hidden_query(self, scope): + in_scope = Q( + "bool", + must=[ + Q("term", group=scope.group_pubid), + Q("bool", should=self._uri_queries(scope.uris), minimum_should_match=1), + ], + ) + + visible = Q( + "bool", + minimum_should_match=1, + should=[ + # user's own annotations. + Q("term", user_raw=self.userid), + # Instructor notes and instructor replies to user. + Q( + "bool", + must=[Q("terms", user_raw=scope.instructor_userids)], + minimum_should_match=1, + should=[ + Q("bool", must_not=[Q("exists", field="references")]), + Q("terms", references=scope.own_annotation_ids), + ], + ), + ], + ) + + # Hide annotations that are in scope but not in the visible set. + return Q("bool", must=[in_scope], must_not=[visible]) + + @staticmethod + def _uri_queries(uris): + queries = [] + for normalized in uris: + queries.append(Q("term", **{"target.scope": normalized})) + queries.append(Q("wildcard", **{"target.scope": f"{normalized}__v*"})) + return queries + + class UriCombinedWildcardFilter: """ A filter that selects only annotations where the uri matches. diff --git a/h/services/checkpoint.py b/h/services/checkpoint.py index 74c5a79728b..1cd4383e169 100644 --- a/h/services/checkpoint.py +++ b/h/services/checkpoint.py @@ -1,8 +1,27 @@ +from dataclasses import dataclass from datetime import datetime from sqlalchemy import or_, select -from h.models import Checkpoint, Document +from h.models import ( + Annotation, + Checkpoint, + Document, + DocumentURI, + GroupMembership, + User, +) +from h.models.group import LMSRole + + +@dataclass +class HiddenScope: + """A (group, document) under an active checkpoint, with its visibility data.""" + + group_pubid: str + uris: list[str] + instructor_userids: list[str] + own_annotation_ids: list[str] class CheckpointService: @@ -40,6 +59,68 @@ def active_checkpoint(self, group_id: int, uri: str) -> Checkpoint | None: .limit(1) ) + def hidden_scopes(self, user: User | None) -> list[HiddenScope]: + """ + Return the scopes whose annotations must be hidden from user. + + An empty list (the common case: a user with no active checkpoints) means + search behaves normally. + """ + if user is None: + return [] + + checkpoints = self.db.scalars( + select(Checkpoint) + .join(GroupMembership, GroupMembership.group_id == Checkpoint.group_id) + .where(GroupMembership.user_id == user.id) + .where( + or_( + GroupMembership.lms_role.is_(None), + GroupMembership.lms_role != LMSRole.LMS_INSTRUCTOR.value, + ) + ) + .where( + or_( + Checkpoint.reveal_date.is_(None), + Checkpoint.reveal_date > datetime.utcnow(), # noqa: DTZ003 + ) + ) + ).all() + + return [self._hidden_scope(user, checkpoint) for checkpoint in checkpoints] + + def _hidden_scope(self, user: User, checkpoint: Checkpoint) -> HiddenScope: + group_pubid = checkpoint.group.pubid + + uris = self.db.scalars( + select(DocumentURI.uri_normalized).where( + DocumentURI.document_id == checkpoint.document_id + ) + ).all() + + # User.userid is a hybrid that compiles to a tuple, so it can't be + # SELECTed directly: load the users and read it in Python. + instructors = self.db.scalars( + select(User) + .join(GroupMembership, GroupMembership.user_id == User.id) + .where(GroupMembership.group_id == checkpoint.group_id) + .where(GroupMembership.lms_role == LMSRole.LMS_INSTRUCTOR.value) + ).all() + instructor_userids = [instructor.userid for instructor in instructors] + + own_annotation_ids = self.db.scalars( + select(Annotation.id) + .where(Annotation.userid == user.userid) + .where(Annotation.groupid == group_pubid) + ).all() + + return HiddenScope( + group_pubid=group_pubid, + uris=list(uris), + instructor_userids=instructor_userids, + own_annotation_ids=list(own_annotation_ids), + ) + def factory(_context, request) -> CheckpointService: """Return a CheckpointService instance for the passed context and request.""" diff --git a/tests/unit/h/search/conftest.py b/tests/unit/h/search/conftest.py index 56fefe5725a..31d5a04861e 100644 --- a/tests/unit/h/search/conftest.py +++ b/tests/unit/h/search/conftest.py @@ -3,6 +3,7 @@ import pytest from h.models import Group +from h.services.checkpoint import CheckpointService from h.services.group import GroupService from h.services.search_index import SearchIndexService @@ -12,6 +13,13 @@ def world_group(db_session): return db_session.query(Group).filter_by(pubid="__world__").one() +@pytest.fixture(autouse=True) +def checkpoint_service(pyramid_config, db_session): + service = CheckpointService(db_session) + pyramid_config.register_service(service, iface=CheckpointService) + return service + + @pytest.fixture def group_service(pyramid_config, world_group): group_service = mock.create_autospec(GroupService, instance=True, spec_set=True) diff --git a/tests/unit/h/search/query_test.py b/tests/unit/h/search/query_test.py index c979592f5d9..e699b65ec1c 100644 --- a/tests/unit/h/search/query_test.py +++ b/tests/unit/h/search/query_test.py @@ -5,7 +5,9 @@ import pytest from webob.multidict import MultiDict +from h.models import GroupMembership from h.models.annotation import ModerationStatus +from h.models.group import LMSRole from h.search import Search, query from h.security.identity import Identity from h.security.permissions import Permission @@ -1173,6 +1175,145 @@ def test_it_returns_annotation_counts_by_user( assert result.aggregations["users"] == expected +class TestHideRevealFilter: + def test_a_student_sees_only_their_visible_annotations( + self, search, Annotation, make_own, other_student, instructor + ): + # The student's own annotation is persisted (not just indexed) so the + # resolver can find it when matching instructor replies. + own = make_own() + instructor_note = Annotation(userid=instructor.userid) + instructor_reply_to_me = Annotation( + userid=instructor.userid, references=[own.id] + ) + peer = Annotation(userid=other_student.userid) + Annotation(userid=instructor.userid, references=[peer.id]) + + result = search.run(MultiDict({})) + + assert sorted(result.annotation_ids) == sorted( + [own.id, instructor_note.id, instructor_reply_to_me.id] + ) + + def test_it_does_not_hide_annotations_on_other_documents( + self, search, Annotation, other_student + ): + elsewhere = Annotation( + userid=other_student.userid, target_uri="http://example.com/other" + ) + + result = search.run(MultiDict({})) + + assert elsewhere.id in result.annotation_ids + + def test_it_does_not_hide_annotations_in_other_groups( + self, search, Annotation, other_student + ): + other = Annotation(userid=other_student.userid, groupid="__world__") + + result = search.run(MultiDict({})) + + assert other.id in result.annotation_ids + + @pytest.mark.usefixtures("revealed_checkpoint") + def test_a_revealed_checkpoint_hides_nothing( + self, search, Annotation, other_student + ): + peer = Annotation(userid=other_student.userid) + + result = search.run(MultiDict({})) + + assert peer.id in result.annotation_ids + + @pytest.mark.usefixtures("as_instructor") + def test_an_instructor_sees_everything(self, search, Annotation, other_student): + peer = Annotation(userid=other_student.userid) + + result = search.run(MultiDict({})) + + assert peer.id in result.annotation_ids + + def membership(self, db_session, group, user, lms_role): + db_session.add(GroupMembership(user=user, group=group, lms_role=lms_role.value)) + db_session.flush() + + @pytest.fixture + def search(self, search, pyramid_request): + search.append_modifier(query.HideRevealFilter(pyramid_request)) + return search + + @pytest.fixture(autouse=True) + def group(self, factories): + return factories.Group() + + @pytest.fixture(autouse=True) + def document(self, factories): + document = factories.Document() + factories.DocumentURI(document=document, uri="http://example.com/page") + return document + + @pytest.fixture(autouse=True) + def checkpoint(self, factories, group, document): + return factories.Checkpoint(group=group, document=document, reveal_date=None) + + @pytest.fixture + def revealed_checkpoint(self, checkpoint): + checkpoint.reveal_date = datetime.datetime(2000, 1, 1) # noqa: DTZ001 + return checkpoint + + @pytest.fixture + def student(self, factories, group, db_session): + student = factories.User() + self.membership(db_session, group, student, LMSRole.LMS_STUDENT) + return student + + @pytest.fixture + def other_student(self, factories, group, db_session): + other_student = factories.User() + self.membership(db_session, group, other_student, LMSRole.LMS_STUDENT) + return other_student + + @pytest.fixture + def instructor(self, factories, group, db_session): + instructor = factories.User() + self.membership(db_session, group, instructor, LMSRole.LMS_INSTRUCTOR) + return instructor + + @pytest.fixture(autouse=True) + def as_student(self, pyramid_request, pyramid_config, student): + pyramid_request.user = student + pyramid_config.testing_securitypolicy(student.userid) + + @pytest.fixture + def as_instructor(self, pyramid_request, pyramid_config, instructor): + pyramid_request.user = instructor + pyramid_config.testing_securitypolicy(instructor.userid) + + @pytest.fixture + def Annotation(self, Annotation, group): + def HideRevealAnnotation(**kwargs): + kwargs.setdefault("groupid", group.pubid) + kwargs.setdefault("target_uri", "http://example.com/page") + kwargs.setdefault("shared", True) + return Annotation(**kwargs) + + return HideRevealAnnotation + + @pytest.fixture + def make_own(self, factories, index_annotations, group, student): + def make_own(): + annotation = factories.Annotation( + userid=student.userid, + groupid=group.pubid, + target_uri="http://example.com/page", + shared=True, + ) + index_annotations(annotation) + return annotation + + return make_own + + @pytest.fixture def search(pyramid_request, group_service): # noqa: ARG001 search = Search(pyramid_request) diff --git a/tests/unit/h/services/checkpoint_test.py b/tests/unit/h/services/checkpoint_test.py index de5240c0eb5..ee633f55ac7 100644 --- a/tests/unit/h/services/checkpoint_test.py +++ b/tests/unit/h/services/checkpoint_test.py @@ -3,7 +3,10 @@ import pytest +from h.models import GroupMembership +from h.models.group import LMSRole from h.services.checkpoint import CheckpointService, factory +from h.util.uri import normalize as uri_normalize class TestActiveCheckpoint: @@ -81,6 +84,110 @@ def document(self, factories): return document +class TestHiddenScopes: + def test_it_returns_empty_for_a_none_user(self, svc): + assert svc.hidden_scopes(None) == [] + + def test_it_returns_empty_when_the_user_has_no_checkpoints(self, svc, user): + assert svc.hidden_scopes(user) == [] + + @pytest.mark.usefixtures("active_checkpoint") + def test_it_returns_empty_when_the_user_is_not_a_member(self, svc, user): + assert svc.hidden_scopes(user) == [] + + @pytest.mark.usefixtures("active_checkpoint", "instructor_membership") + def test_it_returns_empty_for_an_instructor(self, svc, user): + assert svc.hidden_scopes(user) == [] + + @pytest.mark.usefixtures("revealed_checkpoint", "student_membership") + def test_it_returns_empty_when_the_checkpoint_is_revealed(self, svc, user): + assert svc.hidden_scopes(user) == [] + + @pytest.mark.usefixtures("active_checkpoint", "student_membership") + def test_it_returns_a_scope_for_a_student(self, svc, user, group): + [scope] = svc.hidden_scopes(user) + + assert scope.group_pubid == group.pubid + # The scope carries the normalized URIs, matching ES `target.scope`. + assert scope.uris == [uri_normalize("http://example.com/page")] + + @pytest.mark.usefixtures("active_checkpoint", "null_role_membership") + def test_it_restricts_members_with_no_lms_role(self, svc, user): + # Default-deny: anyone not explicitly an instructor is restricted. + assert len(svc.hidden_scopes(user)) == 1 + + @pytest.mark.usefixtures("active_checkpoint", "student_membership") + def test_it_collects_instructor_userids(self, svc, user, group, factories): + instructor = factories.User() + self.membership(group, instructor, LMSRole.LMS_INSTRUCTOR) + + [scope] = svc.hidden_scopes(user) + + assert scope.instructor_userids == [instructor.userid] + + @pytest.mark.usefixtures("active_checkpoint", "student_membership") + def test_it_collects_the_users_own_annotation_ids( + self, svc, user, group, factories + ): + own = factories.Annotation(userid=user.userid, groupid=group.pubid) + # An annotation by someone else in the group is not "own". + factories.Annotation(groupid=group.pubid) + + [scope] = svc.hidden_scopes(user) + + assert scope.own_annotation_ids == [own.id] + + def membership(self, group, user, lms_role): + membership = GroupMembership( + user=user, group=group, lms_role=lms_role.value if lms_role else None + ) + self.db.add(membership) + self.db.flush() + return membership + + @pytest.fixture(autouse=True) + def _db(self, db_session): + self.db = db_session + + @pytest.fixture + def user(self, factories): + return factories.User() + + @pytest.fixture + def group(self, factories): + return factories.Group() + + @pytest.fixture + def document(self, factories): + document = factories.Document() + factories.DocumentURI(document=document, uri="http://example.com/page") + return document + + @pytest.fixture + def active_checkpoint(self, factories, group, document): + return factories.Checkpoint(group=group, document=document, reveal_date=None) + + @pytest.fixture + def revealed_checkpoint(self, factories, group, document): + return factories.Checkpoint( + group=group, + document=document, + reveal_date=datetime.utcnow() - timedelta(days=1), # noqa: DTZ003 + ) + + @pytest.fixture + def student_membership(self, group, user): + return self.membership(group, user, LMSRole.LMS_STUDENT) + + @pytest.fixture + def instructor_membership(self, group, user): + return self.membership(group, user, LMSRole.LMS_INSTRUCTOR) + + @pytest.fixture + def null_role_membership(self, group, user): + return self.membership(group, user, None) + + class TestFactory: def test_it(self, pyramid_request): svc = factory(mock.sentinel.context, pyramid_request)