feat: add the import-versions command, handler logic

only works if wutta-continuum is already installed and enabled.

this also rearranges some existing classes, for better consistency
This commit is contained in:
Lance Edgar 2025-12-18 20:03:47 -06:00
parent c38cd2c179
commit fc250a433c
19 changed files with 1345 additions and 76 deletions

View file

@ -213,6 +213,97 @@ class TestFromFileHandler(DataTestCase):
process_data.assert_called_once_with(input_file_dir=self.tempdir)
class TestFromSqlalchemyHandler(DataTestCase):
def make_handler(self, **kwargs):
return mod.FromSqlalchemyHandler(self.config, **kwargs)
def test_make_source_session(self):
handler = self.make_handler()
self.assertRaises(NotImplementedError, handler.make_source_session)
def test_begin_source_transaction(self):
handler = self.make_handler()
self.assertIsNone(handler.source_session)
with patch.object(handler, "make_source_session", return_value=self.session):
handler.begin_source_transaction()
self.assertIs(handler.source_session, self.session)
def test_commit_source_transaction(self):
model = self.app.model
handler = self.make_handler()
handler.source_session = self.session
self.assertEqual(self.session.query(model.User).count(), 0)
# nb. do not commit this yet
user = model.User(username="fred")
self.session.add(user)
self.assertTrue(self.session.in_transaction())
self.assertIn(user, self.session)
handler.commit_source_transaction()
self.assertIsNone(handler.source_session)
self.assertFalse(self.session.in_transaction())
self.assertNotIn(user, self.session) # hm, surprising?
self.assertEqual(self.session.query(model.User).count(), 1)
def test_rollback_source_transaction(self):
model = self.app.model
handler = self.make_handler()
handler.source_session = self.session
self.assertEqual(self.session.query(model.User).count(), 0)
# nb. do not commit this yet
user = model.User(username="fred")
self.session.add(user)
self.assertTrue(self.session.in_transaction())
self.assertIn(user, self.session)
handler.rollback_source_transaction()
self.assertIsNone(handler.source_session)
self.assertFalse(self.session.in_transaction())
self.assertNotIn(user, self.session)
self.assertEqual(self.session.query(model.User).count(), 0)
def test_get_importer_kwargs(self):
handler = self.make_handler()
handler.source_session = self.session
kw = handler.get_importer_kwargs("User")
self.assertIn("source_session", kw)
self.assertIs(kw["source_session"], self.session)
class TestFromWuttaHandler(DataTestCase):
def make_handler(self, **kwargs):
return mod.FromWuttaHandler(self.config, **kwargs)
def test_get_source_title(self):
handler = self.make_handler()
# uses app title by default
self.config.setdefault("wutta.app_title", "What About This")
self.assertEqual(handler.get_source_title(), "What About This")
# or generic default if present
handler.generic_source_title = "WHATABOUTTHIS"
self.assertEqual(handler.get_source_title(), "WHATABOUTTHIS")
# but prefer specific title if present
handler.source_title = "what_about_this"
self.assertEqual(handler.get_source_title(), "what_about_this")
def test_make_source_session(self):
handler = self.make_handler()
# makes "new" (mocked in our case) app session
with patch.object(self.app, "make_session") as make_session:
make_session.return_value = self.session
session = handler.make_source_session()
make_session.assert_called_once_with()
self.assertIs(session, self.session)
class TestToSqlalchemyHandler(DataTestCase):
def make_handler(self, **kwargs):
@ -256,3 +347,34 @@ class TestToSqlalchemyHandler(DataTestCase):
kw = handler.get_importer_kwargs("Setting")
self.assertIn("target_session", kw)
self.assertIs(kw["target_session"], self.session)
class TestToWuttaHandler(DataTestCase):
def make_handler(self, **kwargs):
return mod.ToWuttaHandler(self.config, **kwargs)
def test_get_target_title(self):
handler = self.make_handler()
# uses app title by default
self.config.setdefault("wutta.app_title", "What About This")
self.assertEqual(handler.get_target_title(), "What About This")
# or generic default if present
handler.generic_target_title = "WHATABOUTTHIS"
self.assertEqual(handler.get_target_title(), "WHATABOUTTHIS")
# but prefer specific title if present
handler.target_title = "what_about_this"
self.assertEqual(handler.get_target_title(), "what_about_this")
def test_make_target_session(self):
handler = self.make_handler()
# makes "new" (mocked in our case) app session
with patch.object(self.app, "make_session") as make_session:
make_session.return_value = self.session
session = handler.make_target_session()
make_session.assert_called_once_with()
self.assertIs(session, self.session)