mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-08-08 03:43:20 +00:00
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 <noreply@anthropic.com>
This commit is contained in:
Trenton H
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
fa96b4629b
commit
1139e7f59b
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
+32
-83
@@ -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=<pk>``
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user