fix(exchange): isolate concurrent connection checks
This commit is contained in:
@@ -505,6 +505,27 @@ class ExchangeConnectionServiceUnitTest(TestCase):
|
||||
self.assertEqual(connections_mock.databases[alias]["PASSWORD"], "secret")
|
||||
self.assertNotIn(alias, storage.__dict__)
|
||||
|
||||
def test_configure_alias_uses_unique_alias_for_unsaved_connection(self):
|
||||
connection = ExchangeConnection(
|
||||
server="127.0.0.1",
|
||||
port=5432,
|
||||
username="postgres",
|
||||
password="secret", # noqa: S106
|
||||
database_name="target_db",
|
||||
schema_name="public",
|
||||
)
|
||||
|
||||
first_alias = ExchangeConnectionService._configure_alias(connection)
|
||||
second_alias = ExchangeConnectionService._configure_alias(connection)
|
||||
|
||||
try:
|
||||
self.assertNotEqual(first_alias, second_alias)
|
||||
self.assertTrue(first_alias.startswith("exchange_target_"))
|
||||
self.assertTrue(second_alias.startswith("exchange_target_"))
|
||||
finally:
|
||||
ExchangeConnectionService._cleanup_alias(first_alias)
|
||||
ExchangeConnectionService._cleanup_alias(second_alias)
|
||||
|
||||
def test_validate_schema_exists_raises_when_schema_missing(self):
|
||||
cursor = MagicMock()
|
||||
cursor.fetchone.return_value = None
|
||||
|
||||
Reference in New Issue
Block a user