diff options
Diffstat (limited to 'tests/storage')
| -rw-r--r-- | tests/storage/__init__.py | 0 | ||||
| -rw-r--r-- | tests/storage/test_base.py | 191 |
2 files changed, 191 insertions, 0 deletions
diff --git a/tests/storage/__init__.py b/tests/storage/__init__.py new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/tests/storage/__init__.py diff --git a/tests/storage/test_base.py b/tests/storage/test_base.py new file mode 100644 index 0000000000..72869ef910 --- /dev/null +++ b/tests/storage/test_base.py @@ -0,0 +1,191 @@ +# -*- coding: utf-8 -*- + +from twisted.trial import unittest +from twisted.internet import defer + +from mock import Mock, call + +from collections import OrderedDict + +from synapse.server import HomeServer +from synapse.storage._base import SQLBaseStore + + +class SQLBaseStoreTestCase(unittest.TestCase): + """ Test the "simple" SQL generating methods in SQLBaseStore. """ + + def setUp(self): + self.db_pool = Mock(spec=["runInteraction"]) + self.mock_txn = Mock() + # Our fake runInteraction just runs synchronously inline + + def runInteraction(func, *args, **kwargs): + return defer.succeed(func(self.mock_txn, *args, **kwargs)) + self.db_pool.runInteraction = runInteraction + + hs = HomeServer("test", + db_pool=self.db_pool) + + self.datastore = SQLBaseStore(hs) + + @defer.inlineCallbacks + def test_insert_1col(self): + self.mock_txn.rowcount = 1 + + yield self.datastore._simple_insert( + table="tablename", + values={"columname": "Value"} + ) + + self.mock_txn.execute.assert_called_with( + "INSERT INTO tablename (columname) VALUES(?)", + ["Value"] + ) + + @defer.inlineCallbacks + def test_insert_3cols(self): + self.mock_txn.rowcount = 1 + + yield self.datastore._simple_insert( + table="tablename", + # Use OrderedDict() so we can assert on the SQL generated + values=OrderedDict([("colA", 1), ("colB", 2), ("colC", 3)]) + ) + + self.mock_txn.execute.assert_called_with( + "INSERT INTO tablename (colA, colB, colC) VALUES(?, ?, ?)", + [1, 2, 3] + ) + + @defer.inlineCallbacks + def test_select_one_1col(self): + self.mock_txn.rowcount = 1 + self.mock_txn.fetchone.return_value = ("Value",) + + value = yield self.datastore._simple_select_one_onecol( + table="tablename", + keyvalues={"keycol": "TheKey"}, + retcol="retcol" + ) + + self.assertEquals("Value", value) + self.mock_txn.execute.assert_called_with( + "SELECT retcol FROM tablename WHERE keycol = ?", + ["TheKey"] + ) + + @defer.inlineCallbacks + def test_select_one_3col(self): + self.mock_txn.rowcount = 1 + self.mock_txn.fetchone.return_value = (1, 2, 3) + + ret = yield self.datastore._simple_select_one( + table="tablename", + keyvalues={"keycol": "TheKey"}, + retcols=["colA", "colB", "colC"] + ) + + self.assertEquals({"colA": 1, "colB": 2, "colC": 3}, ret) + self.mock_txn.execute.assert_called_with( + "SELECT colA, colB, colC FROM tablename WHERE keycol = ?", + ["TheKey"] + ) + + @defer.inlineCallbacks + def test_select_one_missing(self): + self.mock_txn.rowcount = 0 + self.mock_txn.fetchone.return_value = None + + ret = yield self.datastore._simple_select_one( + table="tablename", + keyvalues={"keycol": "Not here"}, + retcols=["colA"], + allow_none=True + ) + + self.assertFalse(ret) + + @defer.inlineCallbacks + def test_select_list(self): + self.mock_txn.rowcount = 3; + self.mock_txn.fetchall.return_value = ((1,), (2,), (3,)) + self.mock_txn.description = ( + ("colA", None, None, None, None, None, None), + ) + + ret = yield self.datastore._simple_select_list( + table="tablename", + keyvalues={"keycol": "A set"}, + retcols=["colA"], + ) + + self.assertEquals([{"colA": 1}, {"colA": 2}, {"colA": 3}], ret) + self.mock_txn.execute.assert_called_with( + "SELECT colA FROM tablename WHERE keycol = ?", + ["A set"] + ) + + @defer.inlineCallbacks + def test_update_one_1col(self): + self.mock_txn.rowcount = 1 + + yield self.datastore._simple_update_one( + table="tablename", + keyvalues={"keycol": "TheKey"}, + updatevalues={"columnname": "New Value"} + ) + + self.mock_txn.execute.assert_called_with( + "UPDATE tablename SET columnname = ? WHERE keycol = ?", + ["New Value", "TheKey"] + ) + + @defer.inlineCallbacks + def test_update_one_4cols(self): + self.mock_txn.rowcount = 1 + + yield self.datastore._simple_update_one( + table="tablename", + keyvalues=OrderedDict([("colA", 1), ("colB", 2)]), + updatevalues=OrderedDict([("colC", 3), ("colD", 4)]) + ) + + self.mock_txn.execute.assert_called_with( + "UPDATE tablename SET colC = ?, colD = ? WHERE " + + "colA = ? AND colB = ?", + [3, 4, 1, 2] + ) + + @defer.inlineCallbacks + def test_update_one_with_return(self): + self.mock_txn.rowcount = 1 + self.mock_txn.fetchone.return_value = ("Old Value",) + + ret = yield self.datastore._simple_update_one( + table="tablename", + keyvalues={"keycol": "TheKey"}, + updatevalues={"columname": "New Value"}, + retcols=["columname"] + ) + + self.assertEquals({"columname": "Old Value"}, ret) + self.mock_txn.execute.assert_has_calls([ + call('SELECT columname FROM tablename WHERE keycol = ?', + ['TheKey']), + call("UPDATE tablename SET columname = ? WHERE keycol = ?", + ["New Value", "TheKey"]) + ]) + + @defer.inlineCallbacks + def test_delete_one(self): + self.mock_txn.rowcount = 1 + + yield self.datastore._simple_delete_one( + table="tablename", + keyvalues={"keycol": "Go away"}, + ) + + self.mock_txn.execute.assert_called_with( + "DELETE FROM tablename WHERE keycol = ?", + ["Go away"] + ) |