From 1139e7f59bf71377e1b92f846d6c9b7554211cbb Mon Sep 17 00:00:00 2001 From: Trenton H <797416+stumpylog@users.noreply.github.com> Date: Mon, 13 Apr 2026 14:28:18 -0700 Subject: [PATCH] feat: rewrite versioning.py to operate on DocumentVersion Replace the old root_document FK navigation with DocumentVersion queryset lookups. Add DocumentVersionFactory to factories.py. Rewrite test_version_conditionals.py to test the new API using pytest style with factories. Co-Authored-By: Claude Sonnet 4.6 --- src/documents/tests/factories.py | 11 ++ .../tests/test_version_conditionals.py | 159 +++++++++--------- src/documents/versioning.py | 115 ++++--------- 3 files changed, 120 insertions(+), 165 deletions(-) diff --git a/src/documents/tests/factories.py b/src/documents/tests/factories.py index b0fd68428..7bdc7341c 100644 --- a/src/documents/tests/factories.py +++ b/src/documents/tests/factories.py @@ -10,6 +10,7 @@ from factory.django import DjangoModelFactory from documents.models import Correspondent from documents.models import Document from documents.models import DocumentType +from documents.models import DocumentVersion from documents.models import MatchingModel from documents.models import StoragePath from documents.models import Tag @@ -65,3 +66,13 @@ class DocumentFactory(DjangoModelFactory): correspondent = None document_type = None storage_path = None + + +class DocumentVersionFactory(DjangoModelFactory): + class Meta: + model = DocumentVersion + + document = factory.SubFactory(DocumentFactory) + version_number = factory.Sequence(lambda n: n + 1) + checksum = factory.Faker("sha256") + mime_type = "application/pdf" diff --git a/src/documents/tests/test_version_conditionals.py b/src/documents/tests/test_version_conditionals.py index fd24a2a51..e91b4deeb 100644 --- a/src/documents/tests/test_version_conditionals.py +++ b/src/documents/tests/test_version_conditionals.py @@ -1,91 +1,86 @@ +from __future__ import annotations + from types import SimpleNamespace -from unittest import mock -from django.test import TestCase +import pytest -from documents.conditionals import metadata_etag -from documents.conditionals import preview_etag -from documents.conditionals import thumbnail_last_modified -from documents.models import Document -from documents.tests.utils import DirectoriesMixin -from documents.versioning import resolve_effective_document_by_pk +from documents.tests.factories import DocumentFactory +from documents.tests.factories import DocumentVersionFactory +from documents.versioning import VersionResolutionError +from documents.versioning import get_latest_version +from documents.versioning import get_version_by_pk +from documents.versioning import resolve_requested_version -class TestConditionals(DirectoriesMixin, TestCase): - def test_metadata_etag_uses_latest_version_for_root_request(self) -> None: - root = Document.objects.create( - title="root", - checksum="root-checksum", - archive_checksum="root-archive", - mime_type="application/pdf", - ) - latest = Document.objects.create( - title="v1", - checksum="version-checksum", - archive_checksum="version-archive", - mime_type="application/pdf", - root_document=root, - ) +@pytest.mark.django_db +class TestGetLatestVersion: + def test_returns_highest_version_number(self) -> None: + doc = DocumentFactory() + DocumentVersionFactory(document=doc, version_number=1) + DocumentVersionFactory(document=doc, version_number=2) + v3 = DocumentVersionFactory(document=doc, version_number=3) + result = get_latest_version(doc) + assert result is not None + assert result.pk == v3.pk + + def test_returns_none_when_no_versions(self) -> None: + doc = DocumentFactory() + assert get_latest_version(doc) is None + + +@pytest.mark.django_db +class TestGetVersionByPk: + def test_returns_version_belonging_to_document(self) -> None: + doc = DocumentFactory() + v = DocumentVersionFactory(document=doc, version_number=1) + result = get_version_by_pk(doc, v.pk) + assert result is not None + assert result.pk == v.pk + + def test_returns_none_for_unrelated_version(self) -> None: + doc_a = DocumentFactory() + doc_b = DocumentFactory() + v_b = DocumentVersionFactory(document=doc_b, version_number=1) + assert get_version_by_pk(doc_a, v_b.pk) is None + + def test_returns_none_for_nonexistent_pk(self) -> None: + doc = DocumentFactory() + assert get_version_by_pk(doc, 999999) is None + + +@pytest.mark.django_db +class TestResolveRequestedVersion: + def test_no_version_param_returns_latest(self) -> None: + doc = DocumentFactory() + DocumentVersionFactory(document=doc, version_number=1) + v2 = DocumentVersionFactory(document=doc, version_number=2) request = SimpleNamespace(query_params={}) + result = resolve_requested_version(doc, request) + assert result.version is not None + assert result.version.pk == v2.pk + assert result.error is None - self.assertEqual(metadata_etag(request, root.id), latest.checksum) - self.assertEqual(preview_etag(request, root.id), latest.archive_checksum) + def test_explicit_version_param_returns_that_version(self) -> None: + doc = DocumentFactory() + v1 = DocumentVersionFactory(document=doc, version_number=1) + DocumentVersionFactory(document=doc, version_number=2) + request = SimpleNamespace(query_params={"version": str(v1.pk)}) + result = resolve_requested_version(doc, request) + assert result.version is not None + assert result.version.pk == v1.pk - def test_resolve_effective_doc_returns_none_for_invalid_or_unrelated_version( - self, - ) -> None: - root = Document.objects.create( - title="root", - checksum="root", - mime_type="application/pdf", - ) - other_root = Document.objects.create( - title="other", - checksum="other", - mime_type="application/pdf", - ) - other_version = Document.objects.create( - title="other-v1", - checksum="other-v1", - mime_type="application/pdf", - root_document=other_root, - ) + def test_invalid_version_param_returns_error(self) -> None: + doc = DocumentFactory() + request = SimpleNamespace(query_params={"version": "notanint"}) + result = resolve_requested_version(doc, request) + assert result.version is None + assert result.error == VersionResolutionError.INVALID - invalid_request = SimpleNamespace(query_params={"version": "not-a-number"}) - unrelated_request = SimpleNamespace( - query_params={"version": str(other_version.id)}, - ) - - self.assertIsNone( - resolve_effective_document_by_pk(root.id, invalid_request).document, - ) - self.assertIsNone( - resolve_effective_document_by_pk(root.id, unrelated_request).document, - ) - - def test_thumbnail_last_modified_uses_effective_document_for_cache_key( - self, - ) -> None: - root = Document.objects.create( - title="root", - checksum="root", - mime_type="application/pdf", - ) - latest = Document.objects.create( - title="v2", - checksum="v2", - mime_type="application/pdf", - root_document=root, - ) - latest.thumbnail_path.parent.mkdir(parents=True, exist_ok=True) - latest.thumbnail_path.write_bytes(b"thumb") - - request = SimpleNamespace(query_params={}) - with mock.patch( - "documents.conditionals.get_thumbnail_modified_key", - return_value="thumb-modified-key", - ) as get_thumb_key: - result = thumbnail_last_modified(request, root.id) - - self.assertIsNotNone(result) - get_thumb_key.assert_called_once_with(latest.id) + def test_unrelated_version_id_returns_not_found(self) -> None: + doc_a = DocumentFactory() + doc_b = DocumentFactory() + v_b = DocumentVersionFactory(document=doc_b, version_number=1) + request = SimpleNamespace(query_params={"version": str(v_b.pk)}) + result = resolve_requested_version(doc_a, request) + assert result.version is None + assert result.error == VersionResolutionError.NOT_FOUND diff --git a/src/documents/versioning.py b/src/documents/versioning.py index 844a5a136..efd73cb2a 100644 --- a/src/documents/versioning.py +++ b/src/documents/versioning.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING from typing import Any from documents.models import Document +from documents.models import DocumentVersion if TYPE_CHECKING: from django.http import HttpRequest @@ -18,107 +19,55 @@ class VersionResolutionError(StrEnum): @dataclass(frozen=True, slots=True) class VersionResolution: - document: Document | None + version: DocumentVersion | None error: VersionResolutionError | None = None -def _document_manager(*, include_deleted: bool) -> Any: - return Document.global_objects if include_deleted else Document.objects - - def get_request_version_param(request: HttpRequest) -> str | None: if hasattr(request, "query_params"): return request.query_params.get("version") return None -def get_root_document(doc: Document, *, include_deleted: bool = False) -> Document: - # Use root_document_id to avoid a query when this is already a root. - # If root_document isn't available, fall back to the document itself. - if doc.root_document_id is None: - return doc - if doc.root_document is not None: - return doc.root_document - - manager = _document_manager(include_deleted=include_deleted) - root_doc = manager.only("id").filter(id=doc.root_document_id).first() - return root_doc or doc +def get_latest_version(doc: Document) -> DocumentVersion | None: + """Return the highest-version_number DocumentVersion for doc, or None.""" + return ( + DocumentVersion.objects.filter(document=doc).order_by("-version_number").first() + ) -def get_latest_version_for_root( - root_doc: Document, - *, - include_deleted: bool = False, -) -> Document: - manager = _document_manager(include_deleted=include_deleted) - latest = manager.filter(root_document=root_doc).order_by("-id").first() - return latest or root_doc +def get_version_by_pk(doc: Document, version_pk: int) -> DocumentVersion | None: + """Return the DocumentVersion with the given pk if it belongs to doc.""" + return DocumentVersion.objects.filter(pk=version_pk, document=doc).first() -def resolve_requested_version_for_root( - root_doc: Document, +def resolve_requested_version( + doc: Document, request: Any, - *, - include_deleted: bool = False, ) -> VersionResolution: + """ + Resolve the DocumentVersion to serve based on the optional ``?version=`` + query parameter. + + - No parameter: return the latest version. + - Parameter present: validate and return that specific version. + """ version_param = get_request_version_param(request) if not version_param: - return VersionResolution( - document=get_latest_version_for_root( - root_doc, - include_deleted=include_deleted, - ), - ) + latest = get_latest_version(doc) + if latest is None: + return VersionResolution( + version=None, + error=VersionResolutionError.NOT_FOUND, + ) + return VersionResolution(version=latest) try: - version_id = int(version_param) + version_pk = int(version_param) except (TypeError, ValueError): - return VersionResolution(document=None, error=VersionResolutionError.INVALID) + return VersionResolution(version=None, error=VersionResolutionError.INVALID) - manager = _document_manager(include_deleted=include_deleted) - candidate = manager.only("id", "root_document_id").filter(id=version_id).first() - if candidate is None: - return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND) - if candidate.id != root_doc.id and candidate.root_document_id != root_doc.id: - return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND) - return VersionResolution(document=candidate) - - -def resolve_effective_document( - request_doc: Document, - request: Any, - *, - include_deleted: bool = False, -) -> VersionResolution: - root_doc = get_root_document(request_doc, include_deleted=include_deleted) - if get_request_version_param(request) is not None: - return resolve_requested_version_for_root( - root_doc, - request, - include_deleted=include_deleted, - ) - if request_doc.root_document_id is None: - return VersionResolution( - document=get_latest_version_for_root( - root_doc, - include_deleted=include_deleted, - ), - ) - return VersionResolution(document=request_doc) - - -def resolve_effective_document_by_pk( - pk: int, - request: Any, - *, - include_deleted: bool = False, -) -> VersionResolution: - manager = _document_manager(include_deleted=include_deleted) - request_doc = manager.only("id", "root_document_id").filter(pk=pk).first() - if request_doc is None: - return VersionResolution(document=None, error=VersionResolutionError.NOT_FOUND) - return resolve_effective_document( - request_doc, - request, - include_deleted=include_deleted, - ) + version = get_version_by_pk(doc, version_pk) + if version is None: + return VersionResolution(version=None, error=VersionResolutionError.NOT_FOUND) + return VersionResolution(version=version)