fix(exchange): limit connection test to postgres access
All checks were successful
CI/CD Pipeline / Quality Gate (push) Successful in 34s
CI/CD Pipeline / Build and Push Images (push) Successful in 19s
CI/CD Pipeline / Internal Notify (push) Successful in 0s
CI/CD Pipeline / Deploy Dev via Compose (push) Successful in 32s

This commit is contained in:
2026-08-10 13:23:23 +02:00
parent 27768edcea
commit 851549587a
2 changed files with 17 additions and 7 deletions

View File

@@ -35,13 +35,14 @@ class ExchangeConnectionService:
@classmethod
def test_connection_payload(cls, **payload) -> dict[str, str]:
"""Проверить подключение и структуру без сохранения в БД."""
"""Проверить доступность PostgreSQL без сохранения в БД."""
connection = ExchangeConnection(is_active=False, **payload)
cls.validate_saved_connection(connection)
alias = cls.test_connection(connection)
cls._cleanup_alias(alias)
return {
"status": "success",
"message": "Подключение проверено. Соединение и структура БД валидны.",
"message": "Подключение проверено. PostgreSQL доступен, логин и пароль верны.",
}
@classmethod

View File

@@ -173,8 +173,16 @@ class ExchangeConnectionServiceUnitTest(TestCase):
def test_test_connection_payload_does_not_persist_connection(self):
with patch.object(
ExchangeConnectionService,
"validate_saved_connection",
) as validate_mock:
"test_connection",
return_value="target_alias",
) as test_connection_mock, patch.object(
ExchangeConnectionService,
"_cleanup_alias",
) as cleanup_mock, patch.object(
ExchangeConnectionService,
"_validate_schema_exists",
side_effect=AssertionError("test-only connection check must not validate schema"),
):
result = ExchangeConnectionService.test_connection_payload(
server="127.0.0.1",
port=5432,
@@ -187,10 +195,11 @@ class ExchangeConnectionServiceUnitTest(TestCase):
self.assertEqual(result["status"], "success")
self.assertEqual(
result["message"],
"Подключение проверено. Соединение и структура БД валидны.",
"Подключение проверено. PostgreSQL доступен, логин и пароль верны.",
)
self.assertEqual(ExchangeConnection.objects.count(), 0)
validate_mock.assert_called_once()
test_connection_mock.assert_called_once()
cleanup_mock.assert_called_once_with("target_alias")
def test_get_active_connection_raises_when_missing(self):
with self.assertRaisesMessage(