wutta/wutta-corepos
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add support for lane_op and lane_trans DB sessions, models

Browse source
This commit is contained in:
Lance Edgar 2025-01-24 19:33:34 -06:00
parent bf6bf63e68
commit e6921c8533
6 changed files with 158 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

7
tests/test_conf.py
View file

@ -16,14 +16,21 @@ class TestWuttaCoreposConfigExtension(TestCase):
self.assertFalse(hasattr(config, 'core_office_op_engine'))
self.assertFalse(hasattr(config, 'core_office_trans_engine'))
self.assertFalse(hasattr(config, 'core_office_arch_engine'))
self.assertFalse(hasattr(config, 'core_lane_op_engine'))
self.assertFalse(hasattr(config, 'core_lane_trans_engine'))
ext = mod.WuttaCoreposConfigExtension()
ext.configure(config)
self.assertIsNone(config.core_office_op_engine)
self.assertIsNone(config.core_office_trans_engine)
self.assertIsNone(config.core_office_arch_engine)
self.assertIsNone(config.core_lane_op_engine)
self.assertIsNone(config.core_lane_trans_engine)
# but config can change that
config.setdefault('corepos.db.office_op.default.url', 'sqlite://')
config.setdefault('corepos.db.lane_trans.default.url', 'sqlite://')
ext.configure(config)
self.assertIsNotNone(config.core_office_op_engine)
self.assertEqual(str(config.core_office_op_engine.url), 'sqlite://')
self.assertIsNotNone(config.core_lane_trans_engine)
self.assertEqual(str(config.core_lane_trans_engine.url), 'sqlite://')

30
tests/test_handler.py
View file

@ -34,6 +34,18 @@ class TestCoreposHandler(ConfigTestCase):
arch_model = handler.get_model_office_arch()
self.assertIs(arch_model, model)
def test_get_model_lane_op(self):
from corepos.db.lane_op import model
handler = self.make_handler()
op_model = handler.get_model_lane_op()
self.assertIs(op_model, model)
def test_get_model_lane_trans(self):
from corepos.db.lane_trans import model
handler = self.make_handler()
trans_model = handler.get_model_lane_trans()
self.assertIs(trans_model, model)
def test_make_session_office_op(self):
handler = self.make_handler()
engine = sa.create_engine('sqlite://')
@ -61,6 +73,24 @@ class TestCoreposHandler(ConfigTestCase):
self.assertIsInstance(arch_session, orm.Session)
self.assertIs(arch_session.bind, engine)
def test_make_session_lane_op(self):
handler = self.make_handler()
engine = sa.create_engine('sqlite://')
with patch.object(self.config, 'core_lane_op_engines', create=True,
new={'default': engine}):
op_session = handler.make_session_lane_op()
self.assertIsInstance(op_session, orm.Session)
self.assertIs(op_session.bind, engine)
def test_make_session_lane_trans(self):
handler = self.make_handler()
engine = sa.create_engine('sqlite://')
with patch.object(self.config, 'core_lane_trans_engines', create=True,
new={'default': engine}):
trans_session = handler.make_session_lane_trans()
self.assertIsInstance(trans_session, orm.Session)
self.assertIs(trans_session.bind, engine)
def test_get_office_url(self):
handler = self.make_handler()