test_unsafe_locale.py 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
  1. # Copyright 2022 The Matrix.org Foundation C.I.C.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from unittest.mock import MagicMock, patch
  15. from synapse.storage.database import make_conn
  16. from synapse.storage.engines import PostgresEngine
  17. from synapse.storage.engines._base import IncorrectDatabaseSetup
  18. from tests.unittest import HomeserverTestCase
  19. from tests.utils import USE_POSTGRES_FOR_TESTS
  20. class UnsafeLocaleTest(HomeserverTestCase):
  21. if not USE_POSTGRES_FOR_TESTS:
  22. skip = "Requires Postgres"
  23. @patch("synapse.storage.engines.postgres.PostgresEngine.get_db_locale")
  24. def test_unsafe_locale(self, mock_db_locale: MagicMock) -> None:
  25. mock_db_locale.return_value = ("B", "B")
  26. database = self.hs.get_datastores().databases[0]
  27. db_conn = make_conn(database._database_config, database.engine, "test_unsafe")
  28. with self.assertRaises(IncorrectDatabaseSetup):
  29. database.engine.check_database(db_conn)
  30. with self.assertRaises(IncorrectDatabaseSetup):
  31. database.engine.check_new_database(db_conn)
  32. db_conn.close()
  33. def test_safe_locale(self) -> None:
  34. database = self.hs.get_datastores().databases[0]
  35. assert isinstance(database.engine, PostgresEngine)
  36. db_conn = make_conn(database._database_config, database.engine, "test_unsafe")
  37. with db_conn.cursor() as txn:
  38. res = database.engine.get_db_locale(txn)
  39. self.assertEqual(res, ("C", "C"))
  40. db_conn.close()