mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-05-07 22:03:31 +08:00
c2c9950bb1
Allow PostgreSQL socket paths without forcing a TCP port and reuse a single URL builder for sync, async, and migration flows. Document Redis socket URLs and close the socket connection request. Closes #5720
102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
import sys
|
|
import unittest
|
|
from enum import Enum
|
|
from types import ModuleType
|
|
|
|
|
|
def _stub_module(name: str, **attrs):
|
|
module = sys.modules.get(name)
|
|
if module is None:
|
|
module = ModuleType(name)
|
|
sys.modules[name] = module
|
|
for key, value in attrs.items():
|
|
setattr(module, key, value)
|
|
return module
|
|
|
|
|
|
class _DummyLogger:
|
|
def __getattr__(self, _name):
|
|
return lambda *args, **kwargs: None
|
|
|
|
|
|
_stub_module(
|
|
"app.log",
|
|
logger=_DummyLogger(),
|
|
log_settings=_DummyLogger(),
|
|
LogConfigModel=type("LogConfigModel", (), {}),
|
|
)
|
|
_stub_module("psutil")
|
|
_schemas_module = _stub_module(
|
|
"app.schemas", MediaType=Enum("MediaType", {"Movie": "Movie", "TV": "TV"})
|
|
)
|
|
_schemas_module.__getattr__ = lambda name: type(name, (), {})
|
|
_stub_module("version", APP_VERSION="test")
|
|
|
|
|
|
from app.core.config import Settings
|
|
|
|
|
|
class PostgreSQLSocketConfigTests(unittest.TestCase):
|
|
def test_postgresql_tcp_url_keeps_host_and_port(self):
|
|
settings = Settings(
|
|
DB_POSTGRESQL_HOST="db",
|
|
DB_POSTGRESQL_PORT="5433",
|
|
DB_POSTGRESQL_DATABASE="moviepilot",
|
|
DB_POSTGRESQL_USERNAME="user",
|
|
DB_POSTGRESQL_PASSWORD="pass",
|
|
)
|
|
|
|
self.assertFalse(settings.DB_POSTGRESQL_SOCKET_MODE)
|
|
self.assertEqual(
|
|
settings.DB_POSTGRESQL_URL(),
|
|
"postgresql://user:pass@db:5433/moviepilot",
|
|
)
|
|
self.assertEqual(
|
|
settings.DB_POSTGRESQL_URL("asyncpg"),
|
|
"postgresql+asyncpg://user:pass@db:5433/moviepilot",
|
|
)
|
|
self.assertEqual(settings.DB_POSTGRESQL_TARGET, "db:5433")
|
|
|
|
def test_postgresql_socket_url_uses_host_query_param(self):
|
|
settings = Settings(
|
|
DB_POSTGRESQL_HOST="/var/run/postgresql",
|
|
DB_POSTGRESQL_PORT="",
|
|
DB_POSTGRESQL_DATABASE="moviepilot",
|
|
DB_POSTGRESQL_USERNAME="user",
|
|
DB_POSTGRESQL_PASSWORD="pass",
|
|
)
|
|
|
|
self.assertTrue(settings.DB_POSTGRESQL_SOCKET_MODE)
|
|
self.assertIsNone(settings.DB_POSTGRESQL_PORT_VALUE)
|
|
self.assertEqual(
|
|
settings.DB_POSTGRESQL_URL(),
|
|
"postgresql://user:pass@/moviepilot?host=%2Fvar%2Frun%2Fpostgresql",
|
|
)
|
|
self.assertEqual(
|
|
settings.DB_POSTGRESQL_URL("asyncpg"),
|
|
"postgresql+asyncpg://user:pass@/moviepilot?host=%2Fvar%2Frun%2Fpostgresql",
|
|
)
|
|
self.assertEqual(settings.DB_POSTGRESQL_TARGET, "socket /var/run/postgresql")
|
|
|
|
def test_postgresql_socket_url_can_keep_explicit_port(self):
|
|
settings = Settings(
|
|
DB_POSTGRESQL_HOST="/var/run/postgresql",
|
|
DB_POSTGRESQL_PORT="5432",
|
|
DB_POSTGRESQL_DATABASE="moviepilot",
|
|
DB_POSTGRESQL_USERNAME="user",
|
|
DB_POSTGRESQL_PASSWORD="",
|
|
)
|
|
|
|
self.assertEqual(
|
|
settings.DB_POSTGRESQL_URL(),
|
|
"postgresql://user@/moviepilot?host=%2Fvar%2Frun%2Fpostgresql&port=5432",
|
|
)
|
|
self.assertEqual(
|
|
settings.DB_POSTGRESQL_TARGET,
|
|
"socket /var/run/postgresql (port 5432)",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|