From d94ce6918618a09702c0fda66ab2676a52c3e9be Mon Sep 17 00:00:00 2001 From: Elim Pizza Date: Fri, 5 Jun 2026 13:46:58 -0300 Subject: [PATCH] Add CheckpointService to resolve active checkpoints --- h/services/__init__.py | 4 + h/services/checkpoint.py | 46 ++++++++++++ tests/unit/h/services/checkpoint_test.py | 94 ++++++++++++++++++++++++ 3 files changed, 144 insertions(+) create mode 100644 h/services/checkpoint.py create mode 100644 tests/unit/h/services/checkpoint_test.py diff --git a/h/services/__init__.py b/h/services/__init__.py index 22f16cbe1ff..6b5061117ba 100644 --- a/h/services/__init__.py +++ b/h/services/__init__.py @@ -11,6 +11,7 @@ BulkGroupService, BulkLMSStatsService, ) +from h.services.checkpoint import CheckpointService from h.services.email import EmailService from h.services.http import HTTPService from h.services.job_queue import JobQueueService @@ -51,6 +52,9 @@ def includeme(config): # pragma: no cover # noqa: PLR0915 config.register_service_factory( "h.services.annotation_write.service_factory", iface=AnnotationWriteService ) + config.register_service_factory( + "h.services.checkpoint.factory", iface=CheckpointService + ) config.register_service_factory("h.services.mention.factory", iface=MentionService) config.register_service_factory( "h.services.notification.factory", iface=NotificationService diff --git a/h/services/checkpoint.py b/h/services/checkpoint.py new file mode 100644 index 00000000000..74c5a79728b --- /dev/null +++ b/h/services/checkpoint.py @@ -0,0 +1,46 @@ +from datetime import datetime + +from sqlalchemy import or_, select + +from h.models import Checkpoint, Document + + +class CheckpointService: + """Resolve Hide & Reveal checkpoints for annotation-search authorization.""" + + def __init__(self, db): + self.db = db + + def active_checkpoint(self, group_id: int, uri: str) -> Checkpoint | None: + """ + Return an active (unrevealed) checkpoint for `(group_id, uri)`, or None. + + The `uri` is resolved to its Document(s) the same way the search layer + resolves the request's `uri` param, so the checkpoint lookup matches the + annotations the search will return even when the same document is + addressed by an equivalent URI (e.g. a PDF fingerprint). + + A checkpoint is "active" (still hiding annotations) when its reveal_date + has not yet passed: it is NULL (never revealed) or in the future. + """ + document_ids = [doc.id for doc in Document.find_by_uris(self.db, [uri])] + if not document_ids: + return None + + return self.db.scalar( + select(Checkpoint) + .where(Checkpoint.group_id == group_id) + .where(Checkpoint.document_id.in_(document_ids)) + .where( + or_( + Checkpoint.reveal_date.is_(None), + Checkpoint.reveal_date > datetime.utcnow(), # noqa: DTZ003 + ) + ) + .limit(1) + ) + + +def factory(_context, request) -> CheckpointService: + """Return a CheckpointService instance for the passed context and request.""" + return CheckpointService(db=request.db) diff --git a/tests/unit/h/services/checkpoint_test.py b/tests/unit/h/services/checkpoint_test.py new file mode 100644 index 00000000000..de5240c0eb5 --- /dev/null +++ b/tests/unit/h/services/checkpoint_test.py @@ -0,0 +1,94 @@ +from datetime import datetime, timedelta +from unittest import mock + +import pytest + +from h.services.checkpoint import CheckpointService, factory + + +class TestActiveCheckpoint: + def test_it_returns_an_unrevealed_checkpoint(self, svc, group, document): + checkpoint = self.checkpoint(group, document, reveal_date=None) + + assert svc.active_checkpoint(group.id, "http://example.com/page") == checkpoint + + def test_it_returns_a_checkpoint_with_a_future_reveal_date( + self, svc, group, document + ): + checkpoint = self.checkpoint( + group, + document, + reveal_date=datetime.utcnow() + timedelta(days=1), # noqa: DTZ003 + ) + + assert svc.active_checkpoint(group.id, "http://example.com/page") == checkpoint + + def test_it_returns_None_when_the_checkpoint_is_revealed( + self, svc, group, document + ): + self.checkpoint( + group, + document, + reveal_date=datetime.utcnow() - timedelta(days=1), # noqa: DTZ003 + ) + + assert svc.active_checkpoint(group.id, "http://example.com/page") is None + + @pytest.mark.usefixtures("document") + def test_it_returns_None_when_there_is_no_checkpoint(self, svc, group): + assert svc.active_checkpoint(group.id, "http://example.com/page") is None + + def test_it_returns_None_for_a_different_group( + self, svc, group, document, factories + ): + self.checkpoint(group, document, reveal_date=None) + other_group = factories.Group() + + assert svc.active_checkpoint(other_group.id, "http://example.com/page") is None + + def test_it_resolves_the_uri_to_the_document(self, svc, group, document, factories): + # A second URI on the same document (e.g. a PDF fingerprint) must + # resolve to the same checkpoint. + factories.DocumentURI(document=document, uri="urn:x-pdf:the-fingerprint") + checkpoint = self.checkpoint(group, document, reveal_date=None) + + assert ( + svc.active_checkpoint(group.id, "urn:x-pdf:the-fingerprint") == checkpoint + ) + + def test_it_returns_None_for_an_unknown_uri(self, svc, group, document): + self.checkpoint(group, document, reveal_date=None) + + assert svc.active_checkpoint(group.id, "http://example.com/other") is None + + def checkpoint(self, group, document, reveal_date): + return self.factories.Checkpoint( + group=group, document=document, reveal_date=reveal_date + ) + + @pytest.fixture(autouse=True) + def _factories(self, factories): + self.factories = factories + + @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 + + +class TestFactory: + def test_it(self, pyramid_request): + svc = factory(mock.sentinel.context, pyramid_request) + + assert isinstance(svc, CheckpointService) + assert svc.db == pyramid_request.db + + +@pytest.fixture +def svc(db_session): + return CheckpointService(db=db_session)