mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-02-23 03:46:24 +00:00
Compare commits
5 Commits
chore/typi
...
feature-da
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3d7c3c15b9 | ||
|
|
1675bd3981 | ||
|
|
b79edfd271 | ||
|
|
8f0a06c2da | ||
|
|
3ceb62f098 |
@@ -202,3 +202,51 @@ def audit_log_check(app_configs, **kwargs):
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
@register()
|
||||
def check_deprecated_db_settings(app_configs, **kwargs) -> list[Warning]:
|
||||
"""Check for deprecated database environment variables.
|
||||
|
||||
Detects legacy advanced options that should be migrated to
|
||||
PAPERLESS_DB_OPTIONS.
|
||||
|
||||
Returns:
|
||||
List of Django Warning instances for any deprecated vars found.
|
||||
"""
|
||||
deprecated_vars = {
|
||||
"PAPERLESS_DB_TIMEOUT": "timeout (or connect_timeout for Postgres/MariaDB)",
|
||||
"PAPERLESS_DB_POOLSIZE": "pool.min_size,pool.max_size",
|
||||
"PAPERLESS_DBSSLMODE": "sslmode (or ssl_mode for MariaDB)",
|
||||
"PAPERLESS_DBSSLROOTCERT": "sslrootcert (or ssl.ca for MariaDB)",
|
||||
"PAPERLESS_DBSSLCERT": "sslcert (or ssl.cert for MariaDB)",
|
||||
"PAPERLESS_DBSSLKEY": "sslkey (or ssl.key for MariaDB)",
|
||||
}
|
||||
|
||||
found_vars = []
|
||||
for var_name in deprecated_vars:
|
||||
if os.getenv(var_name):
|
||||
found_vars.append(var_name)
|
||||
|
||||
if not found_vars:
|
||||
return []
|
||||
|
||||
# Build migration example
|
||||
examples = []
|
||||
for var in found_vars:
|
||||
examples.append(f"{var} -> PAPERLESS_DB_OPTIONS={deprecated_vars[var]}=<value>")
|
||||
|
||||
return [
|
||||
Warning(
|
||||
"Deprecated database environment variables detected",
|
||||
# TODO: Need to check this URL
|
||||
hint=(
|
||||
f"Found: {', '.join(found_vars)}. "
|
||||
"These will be removed in v3.2. "
|
||||
"Migrate to PAPERLESS_DB_OPTIONS instead. "
|
||||
f"Examples: {'; '.join(examples[:3])}. "
|
||||
"See https://docs.paperless-ngx.com/migration/"
|
||||
),
|
||||
id="paperless.W001",
|
||||
),
|
||||
]
|
||||
|
||||
@@ -17,6 +17,8 @@ from dateparser.languages.loader import LocaleDataLoader
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from paperless.settings.custom import parse_db_settings
|
||||
|
||||
logger = logging.getLogger("paperless.settings")
|
||||
|
||||
# Tap paperless.conf if it's available
|
||||
@@ -722,83 +724,8 @@ EMAIL_CERTIFICATE_FILE = __get_optional_path("PAPERLESS_EMAIL_CERTIFICATE_LOCATI
|
||||
###############################################################################
|
||||
# Database #
|
||||
###############################################################################
|
||||
def _parse_db_settings() -> dict:
|
||||
databases = {
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": DATA_DIR / "db.sqlite3",
|
||||
"OPTIONS": {},
|
||||
},
|
||||
}
|
||||
if os.getenv("PAPERLESS_DBHOST"):
|
||||
# Have sqlite available as a second option for management commands
|
||||
# This is important when migrating to/from sqlite
|
||||
databases["sqlite"] = databases["default"].copy()
|
||||
|
||||
databases["default"] = {
|
||||
"HOST": os.getenv("PAPERLESS_DBHOST"),
|
||||
"NAME": os.getenv("PAPERLESS_DBNAME", "paperless"),
|
||||
"USER": os.getenv("PAPERLESS_DBUSER", "paperless"),
|
||||
"PASSWORD": os.getenv("PAPERLESS_DBPASS", "paperless"),
|
||||
"OPTIONS": {},
|
||||
}
|
||||
if os.getenv("PAPERLESS_DBPORT"):
|
||||
databases["default"]["PORT"] = os.getenv("PAPERLESS_DBPORT")
|
||||
|
||||
# Leave room for future extensibility
|
||||
if os.getenv("PAPERLESS_DBENGINE") == "mariadb":
|
||||
engine = "django.db.backends.mysql"
|
||||
# Contrary to Postgres, Django does not natively support connection pooling for MariaDB.
|
||||
# However, since MariaDB uses threads instead of forks, establishing connections is significantly faster
|
||||
# compared to PostgreSQL, so the lack of pooling is not an issue
|
||||
options = {
|
||||
"read_default_file": "/etc/mysql/my.cnf",
|
||||
"charset": "utf8mb4",
|
||||
"ssl_mode": os.getenv("PAPERLESS_DBSSLMODE", "PREFERRED"),
|
||||
"ssl": {
|
||||
"ca": os.getenv("PAPERLESS_DBSSLROOTCERT", None),
|
||||
"cert": os.getenv("PAPERLESS_DBSSLCERT", None),
|
||||
"key": os.getenv("PAPERLESS_DBSSLKEY", None),
|
||||
},
|
||||
}
|
||||
|
||||
else: # Default to PostgresDB
|
||||
engine = "django.db.backends.postgresql"
|
||||
options = {
|
||||
"sslmode": os.getenv("PAPERLESS_DBSSLMODE", "prefer"),
|
||||
"sslrootcert": os.getenv("PAPERLESS_DBSSLROOTCERT", None),
|
||||
"sslcert": os.getenv("PAPERLESS_DBSSLCERT", None),
|
||||
"sslkey": os.getenv("PAPERLESS_DBSSLKEY", None),
|
||||
}
|
||||
if int(os.getenv("PAPERLESS_DB_POOLSIZE", 0)) > 0:
|
||||
options.update(
|
||||
{
|
||||
"pool": {
|
||||
"min_size": 1,
|
||||
"max_size": int(os.getenv("PAPERLESS_DB_POOLSIZE")),
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
databases["default"]["ENGINE"] = engine
|
||||
databases["default"]["OPTIONS"].update(options)
|
||||
|
||||
if os.getenv("PAPERLESS_DB_TIMEOUT") is not None:
|
||||
if databases["default"]["ENGINE"] == "django.db.backends.sqlite3":
|
||||
databases["default"]["OPTIONS"].update(
|
||||
{"timeout": int(os.getenv("PAPERLESS_DB_TIMEOUT"))},
|
||||
)
|
||||
else:
|
||||
databases["default"]["OPTIONS"].update(
|
||||
{"connect_timeout": int(os.getenv("PAPERLESS_DB_TIMEOUT"))},
|
||||
)
|
||||
databases["sqlite"]["OPTIONS"].update(
|
||||
{"timeout": int(os.getenv("PAPERLESS_DB_TIMEOUT"))},
|
||||
)
|
||||
return databases
|
||||
|
||||
|
||||
DATABASES = _parse_db_settings()
|
||||
DATABASES = parse_db_settings(DATA_DIR)
|
||||
|
||||
if os.getenv("PAPERLESS_DBENGINE") == "mariadb":
|
||||
# Silence Django error on old MariaDB versions.
|
||||
125
src/paperless/settings/custom.py
Normal file
125
src/paperless/settings/custom.py
Normal file
@@ -0,0 +1,125 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import TypeAlias
|
||||
|
||||
from paperless.settings.parsers import get_choice_from_env
|
||||
from paperless.settings.parsers import get_int_from_env
|
||||
from paperless.settings.parsers import parse_dict_from_str
|
||||
|
||||
# ENGINE/NAME/HOST/USER/PASSWORD (str), PORT (int), OPTIONS (dict)
|
||||
DatabaseConfig: TypeAlias = dict[str, str | int | dict[str, str | int | dict | None]]
|
||||
|
||||
|
||||
def parse_db_settings(data_dir: Path) -> dict[str, DatabaseConfig]:
|
||||
"""Parse database settings from environment variables.
|
||||
|
||||
Core connection variables (no deprecation):
|
||||
- PAPERLESS_DBENGINE (sqlite/postgresql/mariadb)
|
||||
- PAPERLESS_DBHOST, PAPERLESS_DBPORT
|
||||
- PAPERLESS_DBNAME, PAPERLESS_DBUSER, PAPERLESS_DBPASS
|
||||
|
||||
Advanced options can be set via:
|
||||
- Legacy individual env vars (deprecated in v3.0, removed in v3.2)
|
||||
- PAPERLESS_DB_OPTIONS (recommended v3+ approach)
|
||||
|
||||
Args:
|
||||
data_dir: The data directory path for SQLite database location.
|
||||
|
||||
Returns:
|
||||
A databases dict suitable for Django DATABASES setting.
|
||||
"""
|
||||
engine = get_choice_from_env(
|
||||
"PAPERLESS_DBENGINE",
|
||||
{"sqlite", "postgresql", "mariadb"},
|
||||
default="sqlite",
|
||||
)
|
||||
|
||||
match engine:
|
||||
case "sqlite":
|
||||
db_config = {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": str((data_dir / "db.sqlite3").resolve()),
|
||||
}
|
||||
base_options = {}
|
||||
|
||||
case "postgresql":
|
||||
db_config = {
|
||||
"ENGINE": "django.db.backends.postgresql",
|
||||
"HOST": os.getenv("PAPERLESS_DBHOST"),
|
||||
"NAME": os.getenv("PAPERLESS_DBNAME", "paperless"),
|
||||
"USER": os.getenv("PAPERLESS_DBUSER", "paperless"),
|
||||
"PASSWORD": os.getenv("PAPERLESS_DBPASS", "paperless"),
|
||||
}
|
||||
|
||||
base_options = {
|
||||
"sslmode": os.getenv("PAPERLESS_DBSSLMODE", "prefer"),
|
||||
"sslrootcert": os.getenv("PAPERLESS_DBSSLROOTCERT"),
|
||||
"sslcert": os.getenv("PAPERLESS_DBSSLCERT"),
|
||||
"sslkey": os.getenv("PAPERLESS_DBSSLKEY"),
|
||||
}
|
||||
|
||||
if (pool_size := get_int_from_env("PAPERLESS_DB_POOLSIZE")) is not None:
|
||||
base_options["pool"] = {
|
||||
"min_size": 1,
|
||||
"max_size": pool_size,
|
||||
}
|
||||
|
||||
case "mariadb":
|
||||
db_config = {
|
||||
"ENGINE": "django.db.backends.mysql",
|
||||
"HOST": os.getenv("PAPERLESS_DBHOST"),
|
||||
"NAME": os.getenv("PAPERLESS_DBNAME", "paperless"),
|
||||
"USER": os.getenv("PAPERLESS_DBUSER", "paperless"),
|
||||
"PASSWORD": os.getenv("PAPERLESS_DBPASS", "paperless"),
|
||||
}
|
||||
|
||||
base_options = {
|
||||
"read_default_file": "/etc/mysql/my.cnf",
|
||||
"charset": "utf8mb4",
|
||||
"collation": "utf8mb4_unicode_ci",
|
||||
"ssl_mode": os.getenv("PAPERLESS_DBSSLMODE", "PREFERRED"),
|
||||
"ssl": {
|
||||
"ca": os.getenv("PAPERLESS_DBSSLROOTCERT"),
|
||||
"cert": os.getenv("PAPERLESS_DBSSLCERT"),
|
||||
"key": os.getenv("PAPERLESS_DBSSLKEY"),
|
||||
},
|
||||
}
|
||||
|
||||
# Handle port setting for external databases
|
||||
if (
|
||||
engine in ("postgresql", "mariadb")
|
||||
and (port := get_int_from_env("PAPERLESS_DBPORT")) is not None
|
||||
):
|
||||
db_config["PORT"] = port
|
||||
|
||||
# Handle timeout setting (common across all engines, different key names)
|
||||
if (timeout := get_int_from_env("PAPERLESS_DB_TIMEOUT")) is not None:
|
||||
timeout_key = "timeout" if engine == "sqlite" else "connect_timeout"
|
||||
base_options[timeout_key] = timeout
|
||||
|
||||
# Apply PAPERLESS_DB_OPTIONS overrides
|
||||
db_config["OPTIONS"] = parse_dict_from_str(
|
||||
os.getenv("PAPERLESS_DB_OPTIONS"),
|
||||
defaults=base_options,
|
||||
type_map={
|
||||
# SQLite options
|
||||
"timeout": int,
|
||||
# Postgres/MariaDB options
|
||||
"connect_timeout": int,
|
||||
"pool.min_size": int,
|
||||
"pool.max_size": int,
|
||||
},
|
||||
)
|
||||
|
||||
databases = {"default": db_config}
|
||||
|
||||
# Add SQLite fallback for PostgreSQL/MariaDB
|
||||
# TODO: Is this really useful/used?
|
||||
if engine in ("postgresql", "mariadb"):
|
||||
databases["sqlite"] = {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": str((data_dir / "db.sqlite3").resolve()),
|
||||
"OPTIONS": {},
|
||||
}
|
||||
|
||||
return databases
|
||||
192
src/paperless/settings/parsers.py
Normal file
192
src/paperless/settings/parsers.py
Normal file
@@ -0,0 +1,192 @@
|
||||
import copy
|
||||
import os
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import TypeVar
|
||||
from typing import overload
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
def str_to_bool(value: str) -> bool:
|
||||
"""
|
||||
Converts a string representation of truth to a boolean value.
|
||||
|
||||
Recognizes 'true', '1', 't', 'y', 'yes' as True, and
|
||||
'false', '0', 'f', 'n', 'no' as False. Case-insensitive.
|
||||
|
||||
Args:
|
||||
value: The string to convert.
|
||||
|
||||
Returns:
|
||||
The boolean representation of the string.
|
||||
|
||||
Raises:
|
||||
ValueError: If the string is not a recognized boolean value.
|
||||
"""
|
||||
val_lower = value.strip().lower()
|
||||
if val_lower in ("true", "1", "t", "y", "yes"):
|
||||
return True
|
||||
elif val_lower in ("false", "0", "f", "n", "no"):
|
||||
return False
|
||||
raise ValueError(f"Cannot convert '{value}' to a boolean.")
|
||||
|
||||
|
||||
@overload
|
||||
def get_int_from_env(key: str) -> int | None: ...
|
||||
|
||||
|
||||
@overload
|
||||
def get_int_from_env(key: str, default: None) -> int | None: ...
|
||||
|
||||
|
||||
@overload
|
||||
def get_int_from_env(key: str, default: int) -> int: ...
|
||||
|
||||
|
||||
def get_int_from_env(key: str, default: int | None = None) -> int | None:
|
||||
"""
|
||||
Return an integer value based on the environment variable.
|
||||
If default is provided, returns that value when key is missing.
|
||||
If default is None, returns None when key is missing.
|
||||
"""
|
||||
if key not in os.environ:
|
||||
return default
|
||||
|
||||
return int(os.environ[key])
|
||||
|
||||
|
||||
def parse_dict_from_str(
|
||||
env_str: str | None,
|
||||
defaults: dict[str, Any] | None = None,
|
||||
type_map: Mapping[str, Callable[[str], Any]] | None = None,
|
||||
separator: str = ",",
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Parses a key-value string into a dictionary, applying defaults and casting types.
|
||||
|
||||
Supports nested keys via dot-notation, e.g.:
|
||||
"database.host=localhost,database.port=5432"
|
||||
|
||||
Args:
|
||||
env_str: The string from the environment variable (e.g., "port=9090,debug=true").
|
||||
defaults: A dictionary of default values (can contain nested dicts).
|
||||
type_map: A dictionary mapping keys (dot-notation allowed) to a type or a parsing
|
||||
function (e.g., {'port': int, 'debug': bool, 'database.port': int}).
|
||||
The special `bool` type triggers custom boolean parsing.
|
||||
separator: The character used to separate key-value pairs. Defaults to ','.
|
||||
|
||||
Returns:
|
||||
A dictionary with the parsed and correctly-typed settings.
|
||||
|
||||
Raises:
|
||||
ValueError: If a value cannot be cast to its specified type.
|
||||
"""
|
||||
|
||||
def _set_nested(d: dict, keys: list[str], value: Any) -> None:
|
||||
"""Set a nested value, creating intermediate dicts as needed."""
|
||||
cur = d
|
||||
for k in keys[:-1]:
|
||||
if k not in cur or not isinstance(cur[k], dict):
|
||||
cur[k] = {}
|
||||
cur = cur[k]
|
||||
cur[keys[-1]] = value
|
||||
|
||||
def _get_nested(d: dict, keys: list[str]) -> Any:
|
||||
"""Get nested value or raise KeyError if not present."""
|
||||
cur = d
|
||||
for k in keys:
|
||||
if not isinstance(cur, dict) or k not in cur:
|
||||
raise KeyError
|
||||
cur = cur[k]
|
||||
return cur
|
||||
|
||||
def _has_nested(d: dict, keys: list[str]) -> bool:
|
||||
try:
|
||||
_get_nested(d, keys)
|
||||
return True
|
||||
except KeyError:
|
||||
return False
|
||||
|
||||
settings: dict[str, Any] = copy.deepcopy(defaults) if defaults else {}
|
||||
_type_map = type_map if type_map else {}
|
||||
|
||||
if not env_str:
|
||||
return settings
|
||||
|
||||
# Parse the environment string using the specified separator
|
||||
pairs = [p.strip() for p in env_str.split(separator) if p.strip()]
|
||||
for pair in pairs:
|
||||
if "=" not in pair:
|
||||
# ignore malformed pairs
|
||||
continue
|
||||
key, val = pair.split("=", 1)
|
||||
key = key.strip()
|
||||
val = val.strip()
|
||||
if not key:
|
||||
continue
|
||||
parts = key.split(".")
|
||||
_set_nested(settings, parts, val)
|
||||
|
||||
# Apply type casting to the updated settings (supports nested keys in type_map)
|
||||
for key, caster in _type_map.items():
|
||||
key_parts = key.split(".")
|
||||
if _has_nested(settings, key_parts):
|
||||
raw_val = _get_nested(settings, key_parts)
|
||||
# Only cast if it's a string (i.e. from env parsing). If defaults already provided
|
||||
# a different type we leave it as-is.
|
||||
if isinstance(raw_val, str):
|
||||
try:
|
||||
if caster is bool:
|
||||
parsed = str_to_bool(raw_val)
|
||||
elif caster is Path:
|
||||
parsed = Path(raw_val).resolve()
|
||||
else:
|
||||
parsed = caster(raw_val)
|
||||
except (ValueError, TypeError) as e:
|
||||
caster_name = getattr(caster, "__name__", repr(caster))
|
||||
raise ValueError(
|
||||
f"Error casting key '{key}' with value '{raw_val}' "
|
||||
f"to type '{caster_name}'",
|
||||
) from e
|
||||
_set_nested(settings, key_parts, parsed)
|
||||
|
||||
return settings
|
||||
|
||||
|
||||
def get_choice_from_env(
|
||||
env_key: str,
|
||||
choices: set[str],
|
||||
default: str | None = None,
|
||||
) -> str:
|
||||
"""
|
||||
Gets and validates an environment variable against a set of allowed choices.
|
||||
|
||||
Args:
|
||||
env_key: The environment variable key to validate
|
||||
choices: Set of valid choices for the environment variable
|
||||
default: Optional default value if environment variable is not set
|
||||
|
||||
Returns:
|
||||
The validated environment variable value
|
||||
|
||||
Raises:
|
||||
ValueError: If the environment variable value is not in choices
|
||||
or if no default is provided and env var is missing
|
||||
"""
|
||||
value = os.environ.get(env_key, default)
|
||||
|
||||
if value is None:
|
||||
raise ValueError(
|
||||
f"Environment variable '{env_key}' is required but not set.",
|
||||
)
|
||||
|
||||
if value not in choices:
|
||||
raise ValueError(
|
||||
f"Environment variable '{env_key}' has invalid value '{value}'. "
|
||||
f"Valid choices are: {', '.join(sorted(choices))}",
|
||||
)
|
||||
|
||||
return value
|
||||
@@ -2,13 +2,16 @@ import os
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from django.test import TestCase
|
||||
from django.test import override_settings
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from documents.tests.utils import DirectoriesMixin
|
||||
from documents.tests.utils import FileSystemAssertsMixin
|
||||
from paperless.checks import audit_log_check
|
||||
from paperless.checks import binaries_check
|
||||
from paperless.checks import check_deprecated_db_settings
|
||||
from paperless.checks import debug_mode_check
|
||||
from paperless.checks import paths_check
|
||||
from paperless.checks import settings_values_check
|
||||
@@ -237,3 +240,67 @@ class TestAuditLogChecks(TestCase):
|
||||
("auditlog table was found but audit log is disabled."),
|
||||
msg.msg,
|
||||
)
|
||||
|
||||
|
||||
class TestDeprecatedDbSettings:
|
||||
"""Test suite for deprecated database settings system check."""
|
||||
|
||||
def test_no_deprecated_vars_no_warning(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
) -> None:
|
||||
"""Test that no warning is raised when no deprecated vars are set."""
|
||||
mocker.patch.dict(os.environ, {}, clear=True)
|
||||
|
||||
warnings = check_deprecated_db_settings(None)
|
||||
assert warnings == []
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("env_var", "expected_hint_fragment"),
|
||||
[
|
||||
("PAPERLESS_DB_TIMEOUT", "timeout"),
|
||||
("PAPERLESS_DB_POOLSIZE", "pool.min_size,pool.max_size"),
|
||||
("PAPERLESS_DBSSLMODE", "sslmode"),
|
||||
("PAPERLESS_DBSSLROOTCERT", "sslrootcert"),
|
||||
("PAPERLESS_DBSSLCERT", "sslcert"),
|
||||
("PAPERLESS_DBSSLKEY", "sslkey"),
|
||||
],
|
||||
)
|
||||
def test_deprecated_var_triggers_warning(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
env_var: str,
|
||||
expected_hint_fragment: str,
|
||||
) -> None:
|
||||
"""Test that each deprecated var triggers appropriate warning."""
|
||||
mocker.patch.dict(os.environ, {env_var: "some_value"}, clear=True)
|
||||
|
||||
warnings = check_deprecated_db_settings(None)
|
||||
|
||||
assert len(warnings) == 1
|
||||
assert warnings[0].id == "paperless.W001"
|
||||
assert env_var in warnings[0].hint
|
||||
assert expected_hint_fragment in warnings[0].hint
|
||||
assert "v3.2" in warnings[0].hint
|
||||
|
||||
def test_multiple_deprecated_vars(
|
||||
self,
|
||||
mocker: MockerFixture,
|
||||
) -> None:
|
||||
"""Test that multiple deprecated vars are all listed in warning."""
|
||||
mocker.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"PAPERLESS_DB_TIMEOUT": "30",
|
||||
"PAPERLESS_DB_POOLSIZE": "10",
|
||||
"PAPERLESS_DBSSLMODE": "require",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
|
||||
warnings = check_deprecated_db_settings(None)
|
||||
|
||||
assert len(warnings) == 1
|
||||
assert "PAPERLESS_DB_TIMEOUT" in warnings[0].hint
|
||||
assert "PAPERLESS_DB_POOLSIZE" in warnings[0].hint
|
||||
assert "PAPERLESS_DBSSLMODE" in warnings[0].hint
|
||||
|
||||
@@ -1,19 +1,21 @@
|
||||
import datetime
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest import TestCase
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from celery.schedules import crontab
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from paperless.settings import _parse_base_paths
|
||||
from paperless.settings import _parse_beat_schedule
|
||||
from paperless.settings import _parse_dateparser_languages
|
||||
from paperless.settings import _parse_db_settings
|
||||
from paperless.settings import _parse_ignore_dates
|
||||
from paperless.settings import _parse_paperless_url
|
||||
from paperless.settings import _parse_redis_url
|
||||
from paperless.settings import default_threads_per_worker
|
||||
from paperless.settings.custom import parse_db_settings
|
||||
|
||||
|
||||
class TestIgnoreDateParsing(TestCase):
|
||||
@@ -378,62 +380,302 @@ class TestCeleryScheduleParsing(TestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestDBSettings(TestCase):
|
||||
def test_db_timeout_with_sqlite(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- PAPERLESS_DB_TIMEOUT is set
|
||||
WHEN:
|
||||
- Settings are parsed
|
||||
THEN:
|
||||
- PAPERLESS_DB_TIMEOUT set for sqlite
|
||||
"""
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"PAPERLESS_DB_TIMEOUT": "10",
|
||||
},
|
||||
):
|
||||
databases = _parse_db_settings()
|
||||
class TestParseDbSettings:
|
||||
"""Test suite for parse_db_settings function."""
|
||||
|
||||
self.assertDictEqual(
|
||||
@pytest.mark.parametrize(
|
||||
("env_vars", "expected_database_settings"),
|
||||
[
|
||||
pytest.param(
|
||||
{},
|
||||
{
|
||||
"timeout": 10.0,
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
databases["default"]["OPTIONS"],
|
||||
)
|
||||
|
||||
def test_db_timeout_with_not_sqlite(self) -> None:
|
||||
"""
|
||||
GIVEN:
|
||||
- PAPERLESS_DB_TIMEOUT is set but db is not sqlite
|
||||
WHEN:
|
||||
- Settings are parsed
|
||||
THEN:
|
||||
- PAPERLESS_DB_TIMEOUT set correctly in non-sqlite db & for fallback sqlite db
|
||||
"""
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"PAPERLESS_DBHOST": "127.0.0.1",
|
||||
"PAPERLESS_DB_TIMEOUT": "10",
|
||||
},
|
||||
):
|
||||
databases = _parse_db_settings()
|
||||
|
||||
self.assertDictEqual(
|
||||
databases["default"]["OPTIONS"],
|
||||
databases["default"]["OPTIONS"]
|
||||
| {
|
||||
"connect_timeout": 10.0,
|
||||
},
|
||||
)
|
||||
self.assertDictEqual(
|
||||
id="default-sqlite",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"timeout": 10.0,
|
||||
"PAPERLESS_DBENGINE": "sqlite",
|
||||
"PAPERLESS_DB_OPTIONS": "timeout=30",
|
||||
},
|
||||
databases["sqlite"]["OPTIONS"],
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {
|
||||
"timeout": 30,
|
||||
},
|
||||
},
|
||||
},
|
||||
id="sqlite-with-timeout-override",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"PAPERLESS_DBENGINE": "postgresql",
|
||||
"PAPERLESS_DBHOST": "localhost",
|
||||
},
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.postgresql",
|
||||
"HOST": "localhost",
|
||||
"NAME": "paperless",
|
||||
"USER": "paperless",
|
||||
"PASSWORD": "paperless",
|
||||
"OPTIONS": {
|
||||
"sslmode": "prefer",
|
||||
"sslrootcert": None,
|
||||
"sslcert": None,
|
||||
"sslkey": None,
|
||||
},
|
||||
},
|
||||
"sqlite": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
id="postgresql-defaults",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"PAPERLESS_DBENGINE": "postgresql",
|
||||
"PAPERLESS_DBHOST": "paperless-db-host",
|
||||
"PAPERLESS_DBPORT": "1111",
|
||||
"PAPERLESS_DBNAME": "customdb",
|
||||
"PAPERLESS_DBUSER": "customuser",
|
||||
"PAPERLESS_DBPASS": "custompass",
|
||||
"PAPERLESS_DB_OPTIONS": "pool.max_size=50,pool.min_size=2,sslmode=require",
|
||||
},
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.postgresql",
|
||||
"HOST": "paperless-db-host",
|
||||
"PORT": 1111,
|
||||
"NAME": "customdb",
|
||||
"USER": "customuser",
|
||||
"PASSWORD": "custompass",
|
||||
"OPTIONS": {
|
||||
"sslmode": "require",
|
||||
"sslrootcert": None,
|
||||
"sslcert": None,
|
||||
"sslkey": None,
|
||||
"pool": {
|
||||
"min_size": 2,
|
||||
"max_size": 50,
|
||||
},
|
||||
},
|
||||
},
|
||||
"sqlite": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
id="postgresql-overrides",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"PAPERLESS_DBENGINE": "postgresql",
|
||||
"PAPERLESS_DBHOST": "pghost",
|
||||
"PAPERLESS_DB_POOLSIZE": "10",
|
||||
},
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.postgresql",
|
||||
"HOST": "pghost",
|
||||
"NAME": "paperless",
|
||||
"USER": "paperless",
|
||||
"PASSWORD": "paperless",
|
||||
"OPTIONS": {
|
||||
"sslmode": "prefer",
|
||||
"sslrootcert": None,
|
||||
"sslcert": None,
|
||||
"sslkey": None,
|
||||
"pool": {
|
||||
"min_size": 1,
|
||||
"max_size": 10,
|
||||
},
|
||||
},
|
||||
},
|
||||
"sqlite": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
id="postgresql-legacy-poolsize",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"PAPERLESS_DBENGINE": "postgresql",
|
||||
"PAPERLESS_DBHOST": "pghost",
|
||||
"PAPERLESS_DBSSLMODE": "require",
|
||||
"PAPERLESS_DBSSLROOTCERT": "/certs/ca.crt",
|
||||
"PAPERLESS_DB_TIMEOUT": "30",
|
||||
},
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.postgresql",
|
||||
"HOST": "pghost",
|
||||
"NAME": "paperless",
|
||||
"USER": "paperless",
|
||||
"PASSWORD": "paperless",
|
||||
"OPTIONS": {
|
||||
"sslmode": "require",
|
||||
"sslrootcert": "/certs/ca.crt",
|
||||
"sslcert": None,
|
||||
"sslkey": None,
|
||||
"connect_timeout": 30,
|
||||
},
|
||||
},
|
||||
"sqlite": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
id="postgresql-legacy-ssl-and-timeout",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"PAPERLESS_DBENGINE": "mariadb",
|
||||
"PAPERLESS_DBHOST": "localhost",
|
||||
},
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.mysql",
|
||||
"HOST": "localhost",
|
||||
"NAME": "paperless",
|
||||
"USER": "paperless",
|
||||
"PASSWORD": "paperless",
|
||||
"OPTIONS": {
|
||||
"read_default_file": "/etc/mysql/my.cnf",
|
||||
"charset": "utf8mb4",
|
||||
"collation": "utf8mb4_unicode_ci",
|
||||
"ssl_mode": "PREFERRED",
|
||||
"ssl": {
|
||||
"ca": None,
|
||||
"cert": None,
|
||||
"key": None,
|
||||
},
|
||||
},
|
||||
},
|
||||
"sqlite": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
id="mariadb-defaults",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"PAPERLESS_DBENGINE": "mariadb",
|
||||
"PAPERLESS_DBHOST": "paperless-mariadb-host",
|
||||
"PAPERLESS_DBPORT": "5555",
|
||||
"PAPERLESS_DBUSER": "my-cool-user",
|
||||
"PAPERLESS_DBPASS": "my-secure-password",
|
||||
"PAPERLESS_DB_OPTIONS": "ssl.ca=/path/to/ca.pem,ssl_mode=REQUIRED",
|
||||
},
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.mysql",
|
||||
"HOST": "paperless-mariadb-host",
|
||||
"PORT": 5555,
|
||||
"NAME": "paperless",
|
||||
"USER": "my-cool-user",
|
||||
"PASSWORD": "my-secure-password",
|
||||
"OPTIONS": {
|
||||
"read_default_file": "/etc/mysql/my.cnf",
|
||||
"charset": "utf8mb4",
|
||||
"collation": "utf8mb4_unicode_ci",
|
||||
"ssl_mode": "REQUIRED",
|
||||
"ssl": {
|
||||
"ca": "/path/to/ca.pem",
|
||||
"cert": None,
|
||||
"key": None,
|
||||
},
|
||||
},
|
||||
},
|
||||
"sqlite": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
id="mariadb-overrides",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"PAPERLESS_DBENGINE": "mariadb",
|
||||
"PAPERLESS_DBHOST": "mariahost",
|
||||
"PAPERLESS_DBSSLMODE": "REQUIRED",
|
||||
"PAPERLESS_DBSSLROOTCERT": "/certs/ca.pem",
|
||||
"PAPERLESS_DBSSLCERT": "/certs/client.pem",
|
||||
"PAPERLESS_DBSSLKEY": "/certs/client.key",
|
||||
"PAPERLESS_DB_TIMEOUT": "25",
|
||||
},
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.mysql",
|
||||
"HOST": "mariahost",
|
||||
"NAME": "paperless",
|
||||
"USER": "paperless",
|
||||
"PASSWORD": "paperless",
|
||||
"OPTIONS": {
|
||||
"read_default_file": "/etc/mysql/my.cnf",
|
||||
"charset": "utf8mb4",
|
||||
"collation": "utf8mb4_unicode_ci",
|
||||
"ssl_mode": "REQUIRED",
|
||||
"ssl": {
|
||||
"ca": "/certs/ca.pem",
|
||||
"cert": "/certs/client.pem",
|
||||
"key": "/certs/client.key",
|
||||
},
|
||||
"connect_timeout": 25,
|
||||
},
|
||||
},
|
||||
"sqlite": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": None, # Will be replaced with tmp_path
|
||||
"OPTIONS": {},
|
||||
},
|
||||
},
|
||||
id="mariadb-legacy-ssl-and-timeout",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_parse_db_settings(
|
||||
self,
|
||||
tmp_path: Path,
|
||||
mocker: MockerFixture,
|
||||
env_vars: dict[str, str],
|
||||
expected_database_settings: dict[str, dict],
|
||||
) -> None:
|
||||
"""Test various database configurations with defaults and overrides."""
|
||||
# Clear environment and set test vars
|
||||
mocker.patch.dict(os.environ, env_vars, clear=True)
|
||||
|
||||
# Update expected paths with actual tmp_path
|
||||
if (
|
||||
"default" in expected_database_settings
|
||||
and expected_database_settings["default"]["NAME"] is None
|
||||
):
|
||||
expected_database_settings["default"]["NAME"] = str(
|
||||
tmp_path / "db.sqlite3",
|
||||
)
|
||||
if "sqlite" in expected_database_settings:
|
||||
expected_database_settings["sqlite"]["NAME"] = str(
|
||||
tmp_path / "db.sqlite3",
|
||||
)
|
||||
|
||||
settings = parse_db_settings(tmp_path)
|
||||
|
||||
assert settings == expected_database_settings
|
||||
|
||||
|
||||
class TestPaperlessURLSettings(TestCase):
|
||||
|
||||
Reference in New Issue
Block a user