test_base.py 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200
  1. # -*- coding: utf-8 -*-
  2. # Copyright 2014 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 tests import unittest
  16. from twisted.internet import defer
  17. from mock import Mock, call
  18. from collections import OrderedDict
  19. from synapse.server import HomeServer
  20. from synapse.storage._base import SQLBaseStore
  21. from synapse.storage.engines import create_engine
  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. hs = HomeServer(
  40. "test",
  41. db_pool=self.db_pool,
  42. config=config,
  43. database_engine=create_engine("sqlite3"),
  44. )
  45. self.datastore = SQLBaseStore(hs)
  46. @defer.inlineCallbacks
  47. def test_insert_1col(self):
  48. self.mock_txn.rowcount = 1
  49. yield self.datastore._simple_insert(
  50. table="tablename",
  51. values={"columname": "Value"}
  52. )
  53. self.mock_txn.execute.assert_called_with(
  54. "INSERT INTO tablename (columname) VALUES(?)",
  55. ("Value",)
  56. )
  57. @defer.inlineCallbacks
  58. def test_insert_3cols(self):
  59. self.mock_txn.rowcount = 1
  60. yield self.datastore._simple_insert(
  61. table="tablename",
  62. # Use OrderedDict() so we can assert on the SQL generated
  63. values=OrderedDict([("colA", 1), ("colB", 2), ("colC", 3)])
  64. )
  65. self.mock_txn.execute.assert_called_with(
  66. "INSERT INTO tablename (colA, colB, colC) VALUES(?, ?, ?)",
  67. (1, 2, 3,)
  68. )
  69. @defer.inlineCallbacks
  70. def test_select_one_1col(self):
  71. self.mock_txn.rowcount = 1
  72. self.mock_txn.fetchall.return_value = [("Value",)]
  73. value = yield self.datastore._simple_select_one_onecol(
  74. table="tablename",
  75. keyvalues={"keycol": "TheKey"},
  76. retcol="retcol"
  77. )
  78. self.assertEquals("Value", value)
  79. self.mock_txn.execute.assert_called_with(
  80. "SELECT retcol FROM tablename WHERE keycol = ?",
  81. ["TheKey"]
  82. )
  83. @defer.inlineCallbacks
  84. def test_select_one_3col(self):
  85. self.mock_txn.rowcount = 1
  86. self.mock_txn.fetchone.return_value = (1, 2, 3)
  87. ret = yield self.datastore._simple_select_one(
  88. table="tablename",
  89. keyvalues={"keycol": "TheKey"},
  90. retcols=["colA", "colB", "colC"]
  91. )
  92. self.assertEquals({"colA": 1, "colB": 2, "colC": 3}, ret)
  93. self.mock_txn.execute.assert_called_with(
  94. "SELECT colA, colB, colC FROM tablename WHERE keycol = ?",
  95. ["TheKey"]
  96. )
  97. @defer.inlineCallbacks
  98. def test_select_one_missing(self):
  99. self.mock_txn.rowcount = 0
  100. self.mock_txn.fetchone.return_value = None
  101. ret = yield self.datastore._simple_select_one(
  102. table="tablename",
  103. keyvalues={"keycol": "Not here"},
  104. retcols=["colA"],
  105. allow_none=True
  106. )
  107. self.assertFalse(ret)
  108. @defer.inlineCallbacks
  109. def test_select_list(self):
  110. self.mock_txn.rowcount = 3;
  111. self.mock_txn.fetchall.return_value = ((1,), (2,), (3,))
  112. self.mock_txn.description = (
  113. ("colA", None, None, None, None, None, None),
  114. )
  115. ret = yield self.datastore._simple_select_list(
  116. table="tablename",
  117. keyvalues={"keycol": "A set"},
  118. retcols=["colA"],
  119. )
  120. self.assertEquals([{"colA": 1}, {"colA": 2}, {"colA": 3}], ret)
  121. self.mock_txn.execute.assert_called_with(
  122. "SELECT colA FROM tablename WHERE keycol = ?",
  123. ["A set"]
  124. )
  125. @defer.inlineCallbacks
  126. def test_update_one_1col(self):
  127. self.mock_txn.rowcount = 1
  128. yield self.datastore._simple_update_one(
  129. table="tablename",
  130. keyvalues={"keycol": "TheKey"},
  131. updatevalues={"columnname": "New Value"}
  132. )
  133. self.mock_txn.execute.assert_called_with(
  134. "UPDATE tablename SET columnname = ? WHERE keycol = ?",
  135. ["New Value", "TheKey"]
  136. )
  137. @defer.inlineCallbacks
  138. def test_update_one_4cols(self):
  139. self.mock_txn.rowcount = 1
  140. yield self.datastore._simple_update_one(
  141. table="tablename",
  142. keyvalues=OrderedDict([("colA", 1), ("colB", 2)]),
  143. updatevalues=OrderedDict([("colC", 3), ("colD", 4)])
  144. )
  145. self.mock_txn.execute.assert_called_with(
  146. "UPDATE tablename SET colC = ?, colD = ? WHERE " +
  147. "colA = ? AND colB = ?",
  148. [3, 4, 1, 2]
  149. )
  150. @defer.inlineCallbacks
  151. def test_delete_one(self):
  152. self.mock_txn.rowcount = 1
  153. yield self.datastore._simple_delete_one(
  154. table="tablename",
  155. keyvalues={"keycol": "Go away"},
  156. )
  157. self.mock_txn.execute.assert_called_with(
  158. "DELETE FROM tablename WHERE keycol = ?",
  159. ["Go away"]
  160. )