anonymousland-synapse/tests/storage/test_base.py

199 lines
6.1 KiB
Python
Raw Normal View History

2014-08-12 10:10:52 -04:00
# -*- coding: utf-8 -*-
2016-01-06 23:26:29 -05:00
# Copyright 2014-2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
2014-08-12 10:10:52 -04:00
from tests import unittest
2014-08-12 10:10:52 -04:00
from twisted.internet import defer
2016-02-19 10:34:38 -05:00
from mock import Mock
2014-08-12 10:10:52 -04:00
from collections import OrderedDict
from synapse.server import HomeServer
2014-08-12 10:10:52 -04:00
from synapse.storage._base import SQLBaseStore
from synapse.storage.engines import create_engine
2014-08-12 10:10:52 -04:00
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()
2015-05-15 05:54:04 -04:00
self.mock_conn = Mock(spec_set=["cursor", "rollback", "commit"])
2015-04-08 11:57:14 -04:00
self.mock_conn.cursor.return_value = self.mock_txn
2015-05-15 05:54:04 -04:00
self.mock_conn.rollback.return_value = None
2014-08-12 10:10:52 -04:00
# 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
2015-04-08 11:57:14 -04:00
def runWithConnection(func, *args, **kwargs):
return defer.succeed(func(self.mock_conn, *args, **kwargs))
self.db_pool.runWithConnection = runWithConnection
2015-02-11 10:01:15 -05:00
config = Mock()
config.event_cache_size = 1
2016-02-11 09:10:00 -05:00
config.database_config = {"name": "sqlite3"}
hs = HomeServer(
"test",
db_pool=self.db_pool,
config=config,
2016-02-11 09:10:00 -05:00
database_engine=create_engine(config),
)
2014-08-12 10:10:52 -04:00
self.datastore = SQLBaseStore(hs)
@defer.inlineCallbacks
def test_insert_1col(self):
self.mock_txn.rowcount = 1
yield self.datastore._simple_insert(
2016-02-19 10:34:38 -05:00
table="tablename",
values={"columname": "Value"}
2014-08-12 10:10:52 -04:00
)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"INSERT INTO tablename (columname) VALUES(?)", ("Value",)
2014-08-12 10:10:52 -04:00
)
@defer.inlineCallbacks
def test_insert_3cols(self):
self.mock_txn.rowcount = 1
yield self.datastore._simple_insert(
2016-02-19 10:34:38 -05:00
table="tablename",
# Use OrderedDict() so we can assert on the SQL generated
values=OrderedDict([("colA", 1), ("colB", 2), ("colC", 3)])
2014-08-12 10:10:52 -04:00
)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"INSERT INTO tablename (colA, colB, colC) VALUES(?, ?, ?)",
(1, 2, 3,)
2014-08-12 10:10:52 -04:00
)
@defer.inlineCallbacks
def test_select_one_1col(self):
self.mock_txn.rowcount = 1
2014-11-10 10:32:35 -05:00
self.mock_txn.fetchall.return_value = [("Value",)]
2014-08-12 10:10:52 -04:00
value = yield self.datastore._simple_select_one_onecol(
2016-02-19 10:34:38 -05:00
table="tablename",
keyvalues={"keycol": "TheKey"},
retcol="retcol"
2014-08-12 10:10:52 -04:00
)
self.assertEquals("Value", value)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"SELECT retcol FROM tablename WHERE keycol = ?", ["TheKey"]
2014-08-12 10:10:52 -04:00
)
@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(
2016-02-19 10:34:38 -05:00
table="tablename",
keyvalues={"keycol": "TheKey"},
retcols=["colA", "colB", "colC"]
2014-08-12 10:10:52 -04:00
)
self.assertEquals({"colA": 1, "colB": 2, "colC": 3}, ret)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"SELECT colA, colB, colC FROM tablename WHERE keycol = ?",
["TheKey"]
2014-08-12 10:10:52 -04:00
)
@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(
2016-02-19 10:34:38 -05:00
table="tablename",
keyvalues={"keycol": "Not here"},
retcols=["colA"],
allow_none=True
2014-08-12 10:10:52 -04:00
)
self.assertFalse(ret)
@defer.inlineCallbacks
def test_select_list(self):
2016-02-19 10:34:38 -05:00
self.mock_txn.rowcount = 3
2014-08-12 10:10:52 -04:00
self.mock_txn.fetchall.return_value = ((1,), (2,), (3,))
self.mock_txn.description = (
2016-02-19 10:34:38 -05:00
("colA", None, None, None, None, None, None),
2014-08-12 10:10:52 -04:00
)
ret = yield self.datastore._simple_select_list(
2016-02-19 10:34:38 -05:00
table="tablename",
keyvalues={"keycol": "A set"},
retcols=["colA"],
2014-08-12 10:10:52 -04:00
)
self.assertEquals([{"colA": 1}, {"colA": 2}, {"colA": 3}], ret)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"SELECT colA FROM tablename WHERE keycol = ?",
["A set"]
2014-08-12 10:10:52 -04:00
)
@defer.inlineCallbacks
def test_update_one_1col(self):
self.mock_txn.rowcount = 1
yield self.datastore._simple_update_one(
2016-02-19 10:34:38 -05:00
table="tablename",
keyvalues={"keycol": "TheKey"},
updatevalues={"columnname": "New Value"}
2014-08-12 10:10:52 -04:00
)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"UPDATE tablename SET columnname = ? WHERE keycol = ?",
["New Value", "TheKey"]
2014-08-12 10:10:52 -04:00
)
@defer.inlineCallbacks
def test_update_one_4cols(self):
self.mock_txn.rowcount = 1
yield self.datastore._simple_update_one(
2016-02-19 10:34:38 -05:00
table="tablename",
keyvalues=OrderedDict([("colA", 1), ("colB", 2)]),
updatevalues=OrderedDict([("colC", 3), ("colD", 4)])
2014-08-12 10:10:52 -04:00
)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"UPDATE tablename SET colC = ?, colD = ? WHERE"
" colA = ? AND colB = ?",
[3, 4, 1, 2]
2014-08-12 10:10:52 -04:00
)
@defer.inlineCallbacks
def test_delete_one(self):
self.mock_txn.rowcount = 1
yield self.datastore._simple_delete_one(
2016-02-19 10:34:38 -05:00
table="tablename",
keyvalues={"keycol": "Go away"},
2014-08-12 10:10:52 -04:00
)
self.mock_txn.execute.assert_called_with(
2016-02-19 10:34:38 -05:00
"DELETE FROM tablename WHERE keycol = ?", ["Go away"]
2014-08-12 10:10:52 -04:00
)