test_base.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2014-2016 OpenMarket Ltd
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License");
  5. # you may not use this file except in compliance with the License.
  6. # You may obtain a copy of the License at
  7. #
  8. # http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. from collections import OrderedDict
  16. from mock import Mock
  17. from twisted.internet import defer
  18. from synapse.server import HomeServer
  19. from synapse.storage._base import SQLBaseStore
  20. from synapse.storage.engines import create_engine
  21. from tests import unittest
  22. class SQLBaseStoreTestCase(unittest.TestCase):
  23. """ Test the "simple" SQL generating methods in SQLBaseStore. """
  24. def setUp(self):
  25. self.db_pool = Mock(spec=["runInteraction"])
  26. self.mock_txn = Mock()
  27. self.mock_conn = Mock(spec_set=["cursor", "rollback", "commit"])
  28. self.mock_conn.cursor.return_value = self.mock_txn
  29. self.mock_conn.rollback.return_value = None
  30. # Our fake runInteraction just runs synchronously inline
  31. def runInteraction(func, *args, **kwargs):
  32. return defer.succeed(func(self.mock_txn, *args, **kwargs))
  33. self.db_pool.runInteraction = runInteraction
  34. def runWithConnection(func, *args, **kwargs):
  35. return defer.succeed(func(self.mock_conn, *args, **kwargs))
  36. self.db_pool.runWithConnection = runWithConnection
  37. config = Mock()
  38. config.event_cache_size = 1
  39. config.database_config = {"name": "sqlite3"}
  40. hs = HomeServer(
  41. "test",
  42. db_pool=self.db_pool,
  43. config=config,
  44. database_engine=create_engine(config.database_config),
  45. )
  46. self.datastore = SQLBaseStore(None, hs)
  47. @defer.inlineCallbacks
  48. def test_insert_1col(self):
  49. self.mock_txn.rowcount = 1
  50. yield self.datastore._simple_insert(
  51. table="tablename", values={"columname": "Value"}
  52. )
  53. self.mock_txn.execute.assert_called_with(
  54. "INSERT INTO tablename (columname) VALUES(?)", ("Value",)
  55. )
  56. @defer.inlineCallbacks
  57. def test_insert_3cols(self):
  58. self.mock_txn.rowcount = 1
  59. yield self.datastore._simple_insert(
  60. table="tablename",
  61. # Use OrderedDict() so we can assert on the SQL generated
  62. values=OrderedDict([("colA", 1), ("colB", 2), ("colC", 3)]),
  63. )
  64. self.mock_txn.execute.assert_called_with(
  65. "INSERT INTO tablename (colA, colB, colC) VALUES(?, ?, ?)", (1, 2, 3)
  66. )
  67. @defer.inlineCallbacks
  68. def test_select_one_1col(self):
  69. self.mock_txn.rowcount = 1
  70. self.mock_txn.__iter__ = Mock(return_value=iter([("Value",)]))
  71. value = yield self.datastore._simple_select_one_onecol(
  72. table="tablename", keyvalues={"keycol": "TheKey"}, retcol="retcol"
  73. )
  74. self.assertEquals("Value", value)
  75. self.mock_txn.execute.assert_called_with(
  76. "SELECT retcol FROM tablename WHERE keycol = ?", ["TheKey"]
  77. )
  78. @defer.inlineCallbacks
  79. def test_select_one_3col(self):
  80. self.mock_txn.rowcount = 1
  81. self.mock_txn.fetchone.return_value = (1, 2, 3)
  82. ret = yield self.datastore._simple_select_one(
  83. table="tablename",
  84. keyvalues={"keycol": "TheKey"},
  85. retcols=["colA", "colB", "colC"],
  86. )
  87. self.assertEquals({"colA": 1, "colB": 2, "colC": 3}, ret)
  88. self.mock_txn.execute.assert_called_with(
  89. "SELECT colA, colB, colC FROM tablename WHERE keycol = ?", ["TheKey"]
  90. )
  91. @defer.inlineCallbacks
  92. def test_select_one_missing(self):
  93. self.mock_txn.rowcount = 0
  94. self.mock_txn.fetchone.return_value = None
  95. ret = yield self.datastore._simple_select_one(
  96. table="tablename",
  97. keyvalues={"keycol": "Not here"},
  98. retcols=["colA"],
  99. allow_none=True,
  100. )
  101. self.assertFalse(ret)
  102. @defer.inlineCallbacks
  103. def test_select_list(self):
  104. self.mock_txn.rowcount = 3
  105. self.mock_txn.__iter__ = Mock(return_value=iter([(1,), (2,), (3,)]))
  106. self.mock_txn.description = (("colA", None, None, None, None, None, None),)
  107. ret = yield self.datastore._simple_select_list(
  108. table="tablename", keyvalues={"keycol": "A set"}, retcols=["colA"]
  109. )
  110. self.assertEquals([{"colA": 1}, {"colA": 2}, {"colA": 3}], ret)
  111. self.mock_txn.execute.assert_called_with(
  112. "SELECT colA FROM tablename WHERE keycol = ?", ["A set"]
  113. )
  114. @defer.inlineCallbacks
  115. def test_update_one_1col(self):
  116. self.mock_txn.rowcount = 1
  117. yield self.datastore._simple_update_one(
  118. table="tablename",
  119. keyvalues={"keycol": "TheKey"},
  120. updatevalues={"columnname": "New Value"},
  121. )
  122. self.mock_txn.execute.assert_called_with(
  123. "UPDATE tablename SET columnname = ? WHERE keycol = ?",
  124. ["New Value", "TheKey"],
  125. )
  126. @defer.inlineCallbacks
  127. def test_update_one_4cols(self):
  128. self.mock_txn.rowcount = 1
  129. yield self.datastore._simple_update_one(
  130. table="tablename",
  131. keyvalues=OrderedDict([("colA", 1), ("colB", 2)]),
  132. updatevalues=OrderedDict([("colC", 3), ("colD", 4)]),
  133. )
  134. self.mock_txn.execute.assert_called_with(
  135. "UPDATE tablename SET colC = ?, colD = ? WHERE" " colA = ? AND colB = ?",
  136. [3, 4, 1, 2],
  137. )
  138. @defer.inlineCallbacks
  139. def test_delete_one(self):
  140. self.mock_txn.rowcount = 1
  141. yield self.datastore._simple_delete_one(
  142. table="tablename", keyvalues={"keycol": "Go away"}
  143. )
  144. self.mock_txn.execute.assert_called_with(
  145. "DELETE FROM tablename WHERE keycol = ?", ["Go away"]
  146. )