Files
@ 0d155bc1eace
Branch filter:
Location: rattail-project/rattail/tests/db/test_util.py
0d155bc1eace
4.5 KiB
text/x-python
Tweak docs.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | # -*- coding: utf-8 -*-
import warnings
from unittest import TestCase
from sqlalchemy.pool import SingletonThreadPool, NullPool
from edbob.configuration import AppConfigParser
from rattail.db import util
class TestEngineConfig(TestCase):
def test_standard(self):
config = {
'sqlalchemy.url': 'sqlite://',
}
engine = util.engine_from_config(config)
self.assertEqual(str(engine.url), 'sqlite://')
self.assertTrue(isinstance(engine.pool, SingletonThreadPool))
def test_custom_poolclass(self):
config = {
'sqlalchemy.url': 'sqlite://',
'sqlalchemy.poolclass': 'sqlalchemy.pool:NullPool',
}
engine = util.engine_from_config(config)
self.assertEqual(str(engine.url), 'sqlite://')
self.assertTrue(isinstance(engine.pool, NullPool))
class TestGetEngines(TestCase):
def setUp(self):
self.config = AppConfigParser(u'rattail')
def test_default_section_is_rattail_db(self):
self.config.set(u'rattail.db', u'keys', u'default')
self.config.set(u'rattail.db', u'default.url', u'sqlite://')
engines = util.get_engines(self.config)
self.assertEqual(len(engines), 1)
self.assertEqual(engines.keys()[0], u'default')
self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
def test_custom_section_is_honored(self):
self.config.set(u'mycustomdb', u'keys', u'default')
self.config.set(u'mycustomdb', u'default.url', u'sqlite://')
engines = util.get_engines(self.config, section=u'mycustomdb')
self.assertEqual(len(engines), 1)
self.assertEqual(engines.keys()[0], u'default')
self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
def test_default_section_falls_back_to_edbob_db(self):
self.config.set(u'edbob.db', u'keys', u'default')
self.config.set(u'edbob.db', u'default.url', u'sqlite://')
with warnings.catch_warnings(record=True) as ignore_warnings:
engines = util.get_engines(self.config)
self.assertEqual(len(engines), 1)
self.assertEqual(engines.keys()[0], u'default')
self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
def test_default_prefix_does_not_require_keys_declaration(self):
self.config.set(u'rattail.db', u'default.url', u'sqlite://')
engines = util.get_engines(self.config)
self.assertEqual(len(engines), 1)
self.assertEqual(engines.keys()[0], u'default')
self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
def test_default_prefix_falls_back_to_sqlalchemy(self):
# Still no need to define "keys" option here.
self.config.set(u'rattail.db', u'sqlalchemy.url', u'sqlite://')
engines = util.get_engines(self.config)
self.assertEqual(len(engines), 1)
self.assertEqual(engines.keys()[0], u'default')
self.assertEqual(unicode(engines[u'default'].url), u'sqlite://')
def test_defined_keys_are_included_in_engines_result(self):
# Note there is no "default" key here.
self.config.set(u'rattail.db', u'keys', u'host, store')
self.config.set(u'rattail.db', u'host.url', u'sqlite:///rattail.host.sqlite')
self.config.set(u'rattail.db', u'store.url', u'sqlite:///rattail.store.sqlite')
engines = util.get_engines(self.config)
self.assertEqual(len(engines), 2)
self.assertEqual(sorted(engines.keys()), [u'host', u'store'])
self.assertEqual(unicode(engines[u'host'].url), u'sqlite:///rattail.host.sqlite')
self.assertEqual(unicode(engines[u'store'].url), u'sqlite:///rattail.store.sqlite')
class TestGetDefaultEngine(TestCase):
def setUp(self):
self.config = AppConfigParser(u'rattail')
def test_default_engine_is_loaded_from_rattail_db_section_by_default(self):
self.config.set(u'rattail.db', u'keys', u'default')
self.config.set(u'rattail.db', u'default.url', u'sqlite://')
engine = util.get_default_engine(self.config)
self.assertEqual(unicode(engine.url), u'sqlite://')
def test_default_engine_is_loaded_from_custom_section_if_specified(self):
self.config.set(u'mycustomdb', u'keys', u'default')
self.config.set(u'mycustomdb', u'default.url', u'sqlite://')
engine = util.get_default_engine(self.config, section=u'mycustomdb')
self.assertEqual(unicode(engine.url), u'sqlite://')
def test_no_engine_is_returned_if_none_is_defined(self):
engine = util.get_default_engine(self.config)
self.assertTrue(engine is None)
|