Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Added simple sqla tests for object construction.

  • Loading branch information...
commit c9eba8b981a0c0f61c85906d42d339a549021080 1 parent d7a5d67
@mmerickel mmerickel authored
View
17 anykeystore/backends/sqla.py
@@ -38,12 +38,7 @@ def __init__(self, url, **kw):
if not self.table:
table_name = kw.pop('table_name', 'key_storage')
meta = kw.pop('metadata', api.MetaData())
- self.table = api.Table(table_name, meta,
- api.Column(
- 'key', api.String(256), primary_key=True, nullable=False),
- api.Column('value', api.PickleType(), nullable=False),
- api.Column('expires', api.DateTime()),
- )
+ self.table = self._make_table(table_name, meta)
kw['url'] = url
self.engine = api.engine_from_config(kw, prefix='')
@@ -51,6 +46,16 @@ def __init__(self, url, **kw):
def backend_api(cls):
return __import__('sqlalchemy')
+ def _make_table(self, name, meta):
+ api = self.backend_api
+ table = api.Table(name, meta,
+ api.Column(
+ 'key', api.String(256), primary_key=True, nullable=False),
+ api.Column('value', api.PickleType(), nullable=False),
+ api.Column('expires', api.DateTime()),
+ )
+ return table
+
def _create_table(self):
self.table.create(checkfirst=True, bind=self.engine)
self._created = True
View
49 anykeystore/tests/test_backends/test_sqla.py
@@ -1,31 +1,34 @@
import unittest2 as unittest
+from mock import MagicMock
class TestSQLStore(unittest.TestCase):
- def _makeOne(self, url='sqlite://', **kw):
+ def _getTargetClass(self):
from anykeystore.backends.sqla import SQLStore
- store = SQLStore(url, **kw)
- return store
+ return SQLStore
- def test_it(self):
- store = self._makeOne()
- store.store('foo', 'bar')
- value = store.retrieve('foo')
- self.assertEqual(value, 'bar')
+ def test_init_with_table(self):
+ cls = self._getTargetClass()
+ api = MagicMock(spec=['engine_from_config'])
+ result = cls('myurl', table='foo', backend_api=api)
+ self.assertEqual(result.table, 'foo')
- def test_it_delete(self):
- store = self._makeOne()
- store.store('foo', 'bar')
- store.delete('foo')
- self.assertRaises(KeyError, store.retrieve, 'foo')
+ def test_init_with_table_name(self):
+ cls = self._getTargetClass()
+ cls._make_table = MagicMock(return_value='some_table')
+ api = MagicMock(spec=['engine_from_config', 'MetaData'])
+ api.MetaData.return_value = 'a_meta'
+ result = cls(
+ 'myurl', table_name='foo', backend_api=api)
+ self.assertEqual(result.table, 'some_table')
+ cls._make_table.assert_called_with('foo', 'a_meta')
- def test_it_old(self):
- store = self._makeOne()
- store.store('foo', 'bar', expires=-1)
- self.assertRaises(KeyError, store.retrieve, 'foo')
-
- def test_it_purge(self):
- store = self._makeOne()
- store.store('foo', 'bar', expires=0.01)
- store.purge_expired()
- self.assertRaises(KeyError, store.retrieve, 'foo')
+ def test_init_with_table_name_and_metadata(self):
+ cls = self._getTargetClass()
+ cls._make_table = MagicMock(return_value='some_table')
+ api = MagicMock(spec=['engine_from_config', 'MetaData'])
+ api.MetaData.return_value = 'a_meta'
+ result = cls(
+ 'myurl', table_name='foo', metadata='b_meta', backend_api=api)
+ self.assertEqual(result.table, 'some_table')
+ cls._make_table.assert_called_with('foo', 'b_meta')
Please sign in to comment.
Something went wrong with that request. Please try again.