1
0

test_txn_limit.py 1.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142
  1. # Copyright 2014-2021 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 twisted.test.proto_helpers import MemoryReactor
  15. from synapse.server import HomeServer
  16. from synapse.storage.types import Cursor
  17. from synapse.util import Clock
  18. from tests import unittest
  19. class SQLTransactionLimitTestCase(unittest.HomeserverTestCase):
  20. """Test SQL transaction limit doesn't break transactions."""
  21. def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
  22. return self.setup_test_homeserver(db_txn_limit=1000)
  23. def test_config(self) -> None:
  24. db_config = self.hs.config.database.get_single_database()
  25. self.assertEqual(db_config.config["txn_limit"], 1000)
  26. def test_select(self) -> None:
  27. def do_select(txn: Cursor) -> None:
  28. txn.execute("SELECT 1")
  29. db_pool = self.hs.get_datastores().databases[0]
  30. # force txn limit to roll over at least once
  31. for _ in range(0, 1001):
  32. self.get_success_or_raise(db_pool.runInteraction("test_select", do_select))