Compare commits

...
13 changed files with 240 additions and 13 deletions

No files matched your search

+6 -6
View File
@@ -476,13 +476,13 @@ def update_filename_and_move_files(
return
with FileLock(settings.MEDIA_LOCK):
try:
# If this was waiting for the lock, the filename or archive_filename
# of this document may have been updated. This happens if multiple updates
# get queued from the UI for the same document
# So freshen up the data before doing anything
instance.refresh_from_db()
# If this was waiting for the lock, the filename or archive_filename
# of this document may have been updated. This happens if multiple updates
# get queued from the UI for the same document
# So freshen up the data before doing anything
instance.refresh_from_db()
try:
old_filename = instance.filename
old_source_path = instance.source_path
move_original = False
+4 -1
View File
@@ -44,7 +44,10 @@ def localize_date(value: date | datetime | str, format: str, locale: str) -> str
def format_datetime(value: str | datetime, format: str) -> str:
if isinstance(value, str):
value = parse_date(value)
parsed = parse_datetime(value) or parse_date(value)
if parsed is None:
raise ValueError(f"Invalid date or datetime string: {value!r}")
return parsed.strftime(format=format)
return value.strftime(format=format)
+47
View File
@@ -539,3 +539,50 @@ class TestSystemStatus(APITestCase):
self.assertEqual(response.data["tasks"]["summary"]["success_count"], 1)
self.assertEqual(response.data["tasks"]["summary"]["failure_count"], 1)
self.assertEqual(response.data["tasks"]["summary"]["pending_count"], 1)
def test_system_status_reports_database_error(self) -> None:
"""
GIVEN:
- Reading the migration state fails
WHEN:
- The system status is requested
THEN:
- The response reports the database error rather than a server error
"""
self.client.force_login(self.user)
with mock.patch(
"documents.views.MigrationLoader",
side_effect=Exception("database is down"),
):
response = self.client.get(self.ENDPOINT)
self.assertEqual(response.status_code, status.HTTP_200_OK)
database = response.data["database"]
self.assertEqual(database["status"], "ERROR")
self.assertIsNotNone(database["error"])
self.assertEqual(
database["migration_status"],
{"latest_migration": None, "unapplied_migrations": []},
)
def test_system_status_migration_query_error(self) -> None:
"""
GIVEN:
- The migration graph loads but reading the applied migrations fails
WHEN:
- The system status is requested
THEN:
- No migrations are reported as unapplied, since the state is unknown
"""
self.client.force_login(self.user)
with mock.patch("documents.views.MigrationRecorder") as recorder:
recorder.Migration.objects.all.side_effect = Exception("query failed")
response = self.client.get(self.ENDPOINT)
self.assertEqual(response.status_code, status.HTTP_200_OK)
database = response.data["database"]
self.assertEqual(database["status"], "ERROR")
self.assertEqual(
database["migration_status"],
{"latest_migration": None, "unapplied_migrations": []},
)
+26
View File
@@ -31,6 +31,7 @@ from documents.serialisers import DocumentSerializer
from documents.tasks import empty_trash
from paperless_testing.assertions import FileSystemAssertsMixin
from paperless_testing.dirs import DirectoriesMixin
from paperless_testing.dirs import PaperlessDirs
from paperless_testing.factories import DocumentFactory
from paperless_testing.factories import UserFactory
@@ -2134,3 +2135,28 @@ class TestPathDateLocalization:
with override_settings(FILENAME_FORMAT=filename_format):
filename = generate_filename(document)
assert filename == Path(expected_filename)
@pytest.mark.django_db
def test_refresh_failure_surfaces_database_error(
paperless_dirs: PaperlessDirs,
) -> None:
"""
GIVEN:
- A document with a stored filename
WHEN:
- Saving it triggers a filename update and refreshing it from the database fails
THEN:
- The database error is raised, not an unbound variable error from the rollback
"""
document = DocumentFactory(filename="some.pdf")
with (
mock.patch.object(
Document,
"refresh_from_db",
side_effect=DatabaseError("database is locked"),
),
pytest.raises(DatabaseError, match="database is locked"),
):
document.save()
+21
View File
@@ -4,6 +4,7 @@ from typing import Literal
import pytest
from documents.templating.filters import format_datetime
from documents.templating.filters import localize_date
@@ -294,3 +295,23 @@ class TestDateLocalization:
Tests `localize_date` with `date` string across different locales and formats.
"""
assert localize_date(value, format_style, locale_str) == expected_output
class TestFormatDatetime:
@pytest.mark.parametrize(
("value", "expected"),
[
pytest.param("2026-09-21", "2026-09-21 00:00", id="date"),
pytest.param(
"2026-09-21T13:45:00+00:00",
"2026-09-21 13:45",
id="datetime",
),
],
)
def test_parses_iso_strings(self, value: str, expected: str) -> None:
assert format_datetime(value, "%Y-%m-%d %H:%M") == expected
def test_rejects_unparseable_string(self) -> None:
with pytest.raises(ValueError, match="not a date"):
format_datetime("not a date", "%Y")
+10 -3
View File
@@ -5267,7 +5267,9 @@ class CustomFieldViewSet(PermissionsAwareDocumentCountMixin, ModelViewSet[Custom
"migration_status": inline_serializer(
name="MigrationStatus",
fields={
"latest_migration": serializers.CharField(),
"latest_migration": serializers.CharField(
allow_null=True,
),
"unapplied_migrations": serializers.ListSerializer(
child=serializers.CharField(),
),
@@ -5342,6 +5344,8 @@ class SystemStatusView(PassUserMixin):
db_conn = connections["default"]
db_url = str(db_conn.settings_dict["NAME"])
db_error = None
all_migrations: list[str] = []
applied_migrations: list[str] = []
try:
db_conn.ensure_connection()
@@ -5352,7 +5356,8 @@ class SystemStatusView(PassUserMixin):
f"{m.app}.{m.name}"
for m in MigrationRecorder.Migration.objects.all().order_by("id")
]
except Exception as e: # pragma: no cover
except Exception as e:
all_migrations = []
applied_migrations = []
db_status = "ERROR"
logger.exception(
@@ -5548,7 +5553,9 @@ class SystemStatusView(PassUserMixin):
"status": db_status,
"error": db_error,
"migration_status": {
"latest_migration": applied_migrations[-1],
"latest_migration": (
applied_migrations[-1] if applied_migrations else None
),
"unapplied_migrations": [
m for m in all_migrations if m not in applied_migrations
],
+6 -1
View File
@@ -461,9 +461,14 @@ def _parse_paperless_url():
global CSRF_TRUSTED_ORIGINS, CORS_ALLOWED_ORIGINS, ALLOWED_HOSTS
url = os.getenv("PAPERLESS_URL")
if url:
hostname = urlparse(url).hostname
if not hostname:
raise ImproperlyConfigured(
f"PAPERLESS_URL {url!r} has no hostname, expected a URL like https://example.com",
)
CSRF_TRUSTED_ORIGINS.append(url)
CORS_ALLOWED_ORIGINS.append(url)
ALLOWED_HOSTS.append(urlparse(url).hostname)
ALLOWED_HOSTS.append(hostname)
return url
@@ -166,3 +166,22 @@ class TestPaperlessURLSettings(TestCase):
self.assertIn(url, settings.CSRF_TRUSTED_ORIGINS)
self.assertIn(url, settings.CORS_ALLOWED_ORIGINS)
@pytest.mark.parametrize("url", ["example.com", "https://", "/just/a/path"])
def test_paperless_url_without_hostname_is_rejected(
monkeypatch: pytest.MonkeyPatch,
url: str,
) -> None:
"""
GIVEN:
- PAPERLESS_URL is set to a value with no hostname
WHEN:
- The URL is parsed
THEN:
- A configuration error is raised instead of adding None to ALLOWED_HOSTS
"""
monkeypatch.setenv("PAPERLESS_URL", url)
with pytest.raises(ImproperlyConfigured, match="no hostname"):
_parse_paperless_url()
+2 -2
View File
@@ -169,9 +169,9 @@ class TagMailAction(BaseMailAction):
# The custom tag should look like "apple:<color>"
if "apple:" in parameter.lower():
_, self.color = parameter.split(":")
self.color = self.color.strip()
self.color = self.color.strip().lower()
if self.color.lower() not in APPLE_MAIL_TAG_COLORS:
if self.color not in APPLE_MAIL_TAG_COLORS:
raise MailError("Not a valid AppleMail tag color.")
self.keyword = None
+6
View File
@@ -89,6 +89,9 @@ class PaperlessMailOAuth2Manager:
"""
logger = logging.getLogger("paperless_mail")
logger.debug(f"Attempting to refresh oauth token for account {account}")
if account.refresh_token is None:
logger.error(f"Account {account} has no refresh token to use")
return False
try:
result: OAuth2Token
if account.account_type == MailAccount.MailAccountType.GMAIL_OAUTH:
@@ -103,6 +106,9 @@ class PaperlessMailOAuth2Manager:
refresh_token=account.refresh_token,
),
)
else:
logger.error(f"Account {account} does not use OAuth")
return False
if "refresh_token" in result:
# Outlook returns a new refresh token on refresh, Gmail does not
account.refresh_token = result["refresh_token"]
+27
View File
@@ -23,6 +23,7 @@ from rest_framework.test import APITestCase
from documents.models import Correspondent
from documents.models import MatchingModel
from paperless_mail import tasks
from paperless_mail.mail import APPLE_MAIL_TAG_COLORS
from paperless_mail.mail import MailAccountHandler
from paperless_mail.mail import MailError
from paperless_mail.mail import TagMailAction
@@ -1873,6 +1874,32 @@ class TestErrorCallback:
assert processed_mails.get().owner == owner
class TestTagMailActionColor:
@pytest.mark.parametrize(
"parameter",
["apple:green", "apple:Green", "APPLE: GREEN "],
)
def test_color_case_is_ignored(self, parameter: str) -> None:
"""
GIVEN:
- An Apple Mail tag parameter in any letter case
WHEN:
- The action tags a message
THEN:
- The lowercase color's flag bits are set on the message
"""
action = TagMailAction(parameter, supports_gmail_labels=False)
mailbox = mock.Mock()
action.post_consume(mailbox, "1", parameter)
assert mailbox.flag.call_args_list[1] == mock.call(
"1",
APPLE_MAIL_TAG_COLORS["green"],
value=True,
)
class TestManagementCommand(TestCase):
@mock.patch(
"paperless_mail.management.commands.mail_fetcher.tasks.process_mail_accounts",
@@ -264,6 +264,28 @@ class TestMailOAuth(
MailAccount.objects.filter(imap_server="outlook.office365.com").exists(),
)
def test_oauth_callback_view_unrecognized_scope(self) -> None:
"""
GIVEN:
- A valid OAuth state and a callback scope that is not Google's
WHEN:
- OAuth callback is called
THEN:
- 400 is returned instead of a server error and no account is created
"""
session = self.client.session
session["oauth_state"] = "test_state"
session.save()
for scope in ("", "openid", "https://example.com/other"):
with self.subTest(scope=scope):
response = self.client.get(
"/api/oauth/callback/",
{"code": "test_code", "scope": scope, "state": "test_state"},
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertFalse(MailAccount.objects.exists())
@mock.patch("paperless_mail.mail.get_mailbox")
@mock.patch(
"httpx_oauth.oauth2.BaseOAuth2.refresh_token",
@@ -366,3 +388,39 @@ class TestMailOAuth(
f"Failed to refresh oauth token for account {mail_account}: test_error",
cm.output[0],
)
def test_refresh_account_oauth_token_non_oauth_account(self) -> None:
"""
GIVEN:
- A mail account which does not use OAuth
WHEN:
- A token refresh is requested
THEN:
- False is returned rather than an unbound variable error
"""
account = MailAccountFactory(
account_type=MailAccount.MailAccountType.IMAP,
refresh_token="token",
)
self.assertFalse(
PaperlessMailOAuth2Manager().refresh_account_oauth_token(account),
)
def test_refresh_account_oauth_token_without_refresh_token(self) -> None:
"""
GIVEN:
- An OAuth mail account with no refresh token stored
WHEN:
- A token refresh is requested
THEN:
- False is returned
"""
account = MailAccountFactory(
account_type=MailAccount.MailAccountType.GMAIL_OAUTH,
refresh_token=None,
)
self.assertFalse(
PaperlessMailOAuth2Manager().refresh_account_oauth_token(account),
)
+8
View File
@@ -319,6 +319,14 @@ class OauthCallbackView(GenericAPIView[Any]):
result = oauth_manager.get_outlook_access_token(code)
else:
logger.error(
f"Invalid oauth callback request: unrecognized scope {scope!r}",
)
return HttpResponseBadRequest(
"Invalid request, see logs for more detail",
)
access_token = result["access_token"]
refresh_token = result["refresh_token"]
expires_in = result["expires_in"]