Compare commits

..
Author SHA1 Message Date
shamoon 44f7ab021c Fix: use filtered documents for statistics ASN max 2026-11-07 07:25:07 -08:00
5 changed files with 3 additions and 121 deletions

No files matched your search

-12
View File
@@ -31,7 +31,6 @@ from guardian.core import ObjectPermissionChecker
from guardian.models import GroupObjectPermission
from guardian.models import UserObjectPermission
from rest_framework.permissions import BasePermission
from rest_framework.permissions import DjangoModelPermissions
from rest_framework.permissions import DjangoObjectPermissions
from documents.models import Document
@@ -64,17 +63,6 @@ class PermissionGranteesUpdate(TypedDict, total=False):
PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGranteesUpdate]
class PaperlessModelPermissions(DjangoModelPermissions):
"""Require global model permissions, including view permissions for reads."""
perms_map = {
**DjangoModelPermissions.perms_map,
"GET": ["%(app_label)s.view_%(model_name)s"],
"OPTIONS": ["%(app_label)s.view_%(model_name)s"],
"HEAD": ["%(app_label)s.view_%(model_name)s"],
}
class PaperlessObjectPermissions(DjangoObjectPermissions):
"""
A permissions backend that checks for object-level permissions
@@ -15,7 +15,6 @@ from paperless.models import ColorConvertChoices
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.factories import UserFactory
from paperless_testing.http import read_streaming_response
from paperless_testing.permissions import grant_global
class TestApiAppConfig(DirectoriesMixin, APITestCase):
@@ -27,67 +26,6 @@ class TestApiAppConfig(DirectoriesMixin, APITestCase):
user = UserFactory(username="temp_admin", superuser=True)
self.client.force_authenticate(user=user)
def test_api_read_config_requires_view_permission(self) -> None:
for permissions in (
(),
("change_applicationconfiguration",),
("view_document",),
):
user = UserFactory()
grant_global(user, *permissions)
self.client.force_authenticate(user=user)
for endpoint in (self.ENDPOINT, f"{self.ENDPOINT}1/"):
for method in ("get", "head", "options"):
with self.subTest(
permissions=permissions,
endpoint=endpoint,
method=method,
):
response = getattr(self.client, method)(endpoint)
self.assertEqual(
response.status_code,
status.HTTP_403_FORBIDDEN,
)
def test_api_read_config_with_view_permission(self) -> None:
user = UserFactory()
grant_global(user, "view_applicationconfiguration")
self.client.force_authenticate(user=user)
for endpoint in (self.ENDPOINT, f"{self.ENDPOINT}1/"):
for method in ("get", "head", "options"):
with self.subTest(endpoint=endpoint, method=method):
response = getattr(self.client, method)(endpoint)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_api_view_config_permission_does_not_allow_writes(self) -> None:
user = UserFactory()
grant_global(user, "view_applicationconfiguration")
self.client.force_authenticate(user=user)
for method in ("put", "patch", "delete"):
with self.subTest(method=method):
response = getattr(self.client, method)(f"{self.ENDPOINT}1/")
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_api_change_config_permission_allows_updates(self) -> None:
user = UserFactory()
grant_global(user, "change_applicationconfiguration")
self.client.force_authenticate(user=user)
response = self.client.patch(
f"{self.ENDPOINT}1/",
{"app_title": "Updated title"},
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(
ApplicationConfiguration.objects.get().app_title,
"Updated title",
)
def test_api_get_config(self) -> None:
"""
GIVEN:
-44
View File
@@ -1862,7 +1862,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
checksum="A",
mime_type="application/pdf",
content="abc",
archive_serial_number=42,
)
Document.objects.create(
title="none2",
@@ -1906,7 +1905,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
self.assertEqual(response.data["correspondent_count"], 2)
self.assertEqual(response.data["document_type_count"], 1)
self.assertEqual(response.data["storage_path_count"], 2)
self.assertEqual(response.data["current_asn"], 42)
def test_statistics_excludes_document_versions(self) -> None:
root = Document.objects.create(
@@ -1914,7 +1912,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
checksum="A",
mime_type="application/pdf",
content="root",
archive_serial_number=42,
)
version = Document.objects.create(
title="version",
@@ -1923,7 +1920,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
content="version",
root_document=root,
version_index=1,
archive_serial_number=99,
)
tag_inbox = Tag.objects.create(name="t1", is_inbox_tag=True)
version.tags.add(tag_inbox)
@@ -1934,7 +1930,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
self.assertEqual(response.data["documents_total"], 1)
self.assertEqual(response.data["documents_inbox"], 0)
self.assertEqual(response.data["character_count"], 4)
self.assertEqual(response.data["current_asn"], 42)
self.assertEqual(
response.data["document_file_type_counts"][0]["mime_type_count"],
1,
@@ -1948,45 +1943,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
self.assertEqual(response.data["documents_inbox"], None)
self.assertEqual(response.data["inbox_tags"], None)
def test_statistics_current_asn_respects_document_permissions(self) -> None:
viewer = UserFactory()
grant_global(viewer, "view_document")
DocumentFactory(owner=viewer, archive_serial_number=12)
shared = DocumentFactory(owner=self.user, archive_serial_number=23)
grant_object(viewer, shared, "view_document")
DocumentFactory(owner=self.user, archive_serial_number=99)
self.client.force_authenticate(user=viewer)
response = self.client.get("/api/statistics/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["current_asn"], 23)
def test_statistics_current_asn_without_visible_documents(self) -> None:
DocumentFactory(owner=self.user, archive_serial_number=99)
for permissions in ((), ("view_document",)):
viewer = UserFactory()
grant_global(viewer, *permissions)
self.client.force_authenticate(user=viewer)
with self.subTest(permissions=permissions):
response = self.client.get("/api/statistics/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["current_asn"], 0)
def test_statistics_current_asn_with_global_statistics_permission(self) -> None:
viewer = UserFactory()
grant_global(viewer, "view_global_statistics")
DocumentFactory(owner=self.user, archive_serial_number=99)
self.client.force_authenticate(user=viewer)
response = self.client.get("/api/statistics/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["current_asn"], 99)
def test_statistics_multiple_users(self) -> None:
"""
GIVEN:
+1 -1
View File
@@ -4078,7 +4078,7 @@ class StatisticsView(GenericAPIView[Any]):
for row in mime_type_stats
]
current_asn = Document.objects.aggregate(
current_asn = documents.aggregate(
Max("archive_serial_number", default=0),
).get(
"archive_serial_number__max",
+2 -2
View File
@@ -32,13 +32,13 @@ from rest_framework.fields import BooleanField
from rest_framework.filters import OrderingFilter
from rest_framework.generics import GenericAPIView
from rest_framework.pagination import PageNumberPagination
from rest_framework.permissions import DjangoModelPermissions
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.throttling import ScopedRateThrottle
from rest_framework.viewsets import ModelViewSet
from documents.models import PaperlessTask
from documents.permissions import PaperlessModelPermissions
from documents.permissions import PaperlessObjectPermissions
from documents.tasks import llmindex_index
from paperless.filters import GroupFilterSet
@@ -414,7 +414,7 @@ class ApplicationConfigurationViewSet(ModelViewSet[ApplicationConfiguration]):
queryset = ApplicationConfiguration.objects
serializer_class = ApplicationConfigurationSerializer
permission_classes = (IsAuthenticated, PaperlessModelPermissions)
permission_classes = (IsAuthenticated, DjangoModelPermissions)
@extend_schema(exclude=True)
def create(self, request, *args, **kwargs):