wuttasync/tests/importing/test_wutta.py

# -*- coding: utf-8; -*-

from unittest.mock import patch

import sqlalchemy as sa

from wuttjamaican.testing import DataTestCase

from wuttasync.importing import wutta as mod
from wuttasync.importing import ToWutta


class TestFromWuttaMirror(DataTestCase):

    def make_importer(self, **kwargs):
        return mod.FromWuttaMirror(self.config, **kwargs)

    def test_basic(self):
        importer = self.make_importer()
        self.assertIsInstance(importer, mod.FromWuttaMirror)


class TestFromWuttaToWuttaBase(DataTestCase):

    def make_handler(self, **kwargs):
        return mod.FromWuttaToWuttaBase(self.config, **kwargs)

    def test_dbkey(self):

        # null by default
        handler = self.make_handler()
        self.assertIsNone(handler.dbkey)

        # but caller can specify
        handler = self.make_handler(dbkey="another")
        self.assertEqual(handler.dbkey, "another")

    def test_make_importer_factory(self):
        model = self.app.model
        handler = self.make_handler()

        # returns a typical importer
        factory = handler.make_importer_factory(model.User, "User")
        self.assertTrue(issubclass(factory, mod.FromWuttaMirror))
        self.assertTrue(issubclass(factory, ToWutta))
        self.assertIs(factory.model_class, model.User)
        self.assertEqual(factory.__name__, "UserImporter")

    def test_define_importers(self):
        handler = self.make_handler()

        # all models are included
        importers = handler.define_importers()
        self.assertIn("Setting", importers)
        self.assertIn("Person", importers)
        self.assertIn("Role", importers)
        self.assertIn("Permission", importers)
        self.assertIn("User", importers)
        self.assertIn("UserRole", importers)
        self.assertIn("UserAPIToken", importers)
        self.assertIn("Upgrade", importers)
        self.assertNotIn("BatchMixin", importers)
        self.assertNotIn("BatchRowMixin", importers)
        self.assertNotIn("Base", importers)

        # also, dependencies are implied by sort order
        models = list(importers)
        self.assertLess(models.index("Person"), models.index("User"))
        self.assertLess(models.index("User"), models.index("UserRole"))
        self.assertLess(models.index("User"), models.index("Upgrade"))


class TestFromWuttaToWuttaImport(DataTestCase):

    def make_handler(self, **kwargs):
        return mod.FromWuttaToWuttaImport(self.config, **kwargs)

    def test_make_source_session(self):

        # error if null dbkey
        handler = self.make_handler()
        self.assertIsNone(handler.dbkey)
        self.assertRaises(ValueError, handler.make_source_session)

        # error if dbkey not found
        handler = self.make_handler(dbkey="another")
        self.assertEqual(handler.dbkey, "another")
        self.assertNotIn("another", self.config.appdb_engines)
        self.assertRaises(ValueError, handler.make_source_session)

        # error if dbkey is 'default'
        handler = self.make_handler(dbkey="default")
        self.assertEqual(handler.dbkey, "default")
        self.assertIn("default", self.config.appdb_engines)
        self.assertRaises(ValueError, handler.make_source_session)

        # expected behavior
        another_engine = sa.create_engine("sqlite://")
        handler = self.make_handler(dbkey="another")
        with patch.dict(self.config.appdb_engines, {"another": another_engine}):
            session = handler.make_source_session()
            self.assertIs(session.bind, another_engine)


class TestFromWuttaToWuttaExport(DataTestCase):

    def make_handler(self, **kwargs):
        return mod.FromWuttaToWuttaExport(self.config, **kwargs)

    def test_make_target_session(self):

        # error if null dbkey
        handler = self.make_handler()
        self.assertIsNone(handler.dbkey)
        self.assertRaises(ValueError, handler.make_target_session)

        # error if dbkey not found
        handler = self.make_handler(dbkey="another")
        self.assertEqual(handler.dbkey, "another")
        self.assertNotIn("another", self.config.appdb_engines)
        self.assertRaises(ValueError, handler.make_target_session)

        # error if dbkey is 'default'
        handler = self.make_handler(dbkey="default")
        self.assertEqual(handler.dbkey, "default")
        self.assertIn("default", self.config.appdb_engines)
        self.assertRaises(ValueError, handler.make_target_session)

        # expected behavior
        another_engine = sa.create_engine("sqlite://")
        handler = self.make_handler(dbkey="another")
        with patch.dict(self.config.appdb_engines, {"another": another_engine}):
            session = handler.make_target_session()
            self.assertIs(session.bind, another_engine)