wuttasync/tests/importing/test_csv.py

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

import csv
import uuid as _uuid
from unittest.mock import patch

from wuttjamaican.testing import DataTestCase

from wuttasync.importing import (
    csv as mod,
    ImportHandler,
    ToSqlalchemyHandler,
    ToSqlalchemy,
)


class TestFromCsv(DataTestCase):

    def setUp(self):
        self.setup_db()
        self.handler = ImportHandler(self.config)

        self.data_path = self.write_file(
            "data.txt",
            """\
name,value
foo,bar
foo2,bar2
""",
        )

    def make_importer(self, **kwargs):
        kwargs.setdefault("handler", self.handler)
        return mod.FromCsv(self.config, **kwargs)

    def test_get_input_file_name(self):
        model = self.app.model
        imp = self.make_importer(model_class=model.Setting)

        # name can be guessed
        self.assertEqual(imp.get_input_file_name(), "Setting.csv")

        # name can be explicitly set
        imp.input_file_name = "data.txt"
        self.assertEqual(imp.get_input_file_name(), "data.txt")

    def test_open_input_file(self):
        model = self.app.model
        imp = self.make_importer(model_class=model.Setting)

        # normal operation, input file includes all fields
        imp = self.make_importer(
            model_class=model.Setting, input_file_path=self.data_path
        )
        self.assertEqual(imp.fields, ["name", "value"])
        imp.open_input_file()
        self.assertEqual(imp.input_file.name, self.data_path)
        self.assertIsInstance(imp.input_reader, csv.DictReader)
        self.assertEqual(imp.fields, ["name", "value"])
        imp.input_file.close()

        # this file is missing a field, plus we'll pretend more are
        # supported - but should wind up with just the one field
        missing = self.write_file("missing.txt", "name")
        imp = self.make_importer(model_class=model.Setting, input_file_path=missing)
        imp.fields.extend(["lots", "more"])
        self.assertEqual(imp.fields, ["name", "value", "lots", "more"])
        imp.open_input_file()
        self.assertEqual(imp.fields, ["name"])
        imp.input_file.close()

        # and what happens when no known fields are found
        bogus = self.write_file("bogus.txt", "blarg")
        imp = self.make_importer(model_class=model.Setting, input_file_path=bogus)
        self.assertEqual(imp.fields, ["name", "value"])
        self.assertRaises(ValueError, imp.open_input_file)

    def test_close_input_file(self):
        model = self.app.model
        imp = self.make_importer(model_class=model.Setting)

        imp.input_file_path = self.data_path
        imp.open_input_file()
        imp.close_input_file()
        self.assertFalse(hasattr(imp, "input_reader"))
        self.assertFalse(hasattr(imp, "input_file"))

    def test_get_source_objects(self):
        model = self.app.model
        imp = self.make_importer(model_class=model.Setting)

        imp.input_file_path = self.data_path
        imp.open_input_file()
        objects = imp.get_source_objects()
        imp.close_input_file()
        self.assertEqual(len(objects), 2)
        self.assertEqual(objects[0], {"name": "foo", "value": "bar"})
        self.assertEqual(objects[1], {"name": "foo2", "value": "bar2"})


class MockMixinImporter(mod.FromCsvToSqlalchemyMixin, mod.FromCsv, ToSqlalchemy):
    pass


class TestFromCsvToSqlalchemyMixin(DataTestCase):

    def setUp(self):
        self.setup_db()
        self.handler = ImportHandler(self.config)

    def make_importer(self, **kwargs):
        kwargs.setdefault("handler", self.handler)
        return MockMixinImporter(self.config, **kwargs)

    def test_constructor(self):
        model = self.app.model

        # no uuid keys
        imp = self.make_importer(model_class=model.Setting)
        self.assertEqual(imp.uuid_keys, [])

        # typical
        # nb. as of now Upgrade is the only table using proper UUID
        imp = self.make_importer(model_class=model.Upgrade)
        self.assertEqual(imp.uuid_keys, ["uuid"])

    def test_normalize_source_object(self):
        model = self.app.model

        # no uuid keys
        imp = self.make_importer(model_class=model.Setting)
        result = imp.normalize_source_object({"name": "foo", "value": "bar"})
        self.assertEqual(result, {"name": "foo", "value": "bar"})

        # source has proper UUID
        # nb. as of now Upgrade is the only table using proper UUID
        imp = self.make_importer(
            model_class=model.Upgrade, fields=["uuid", "description"]
        )
        result = imp.normalize_source_object(
            {
                "uuid": _uuid.UUID("06753693-d892-77f0-8000-ce71bf7ebbba"),
                "description": "testing",
            }
        )
        self.assertEqual(
            result,
            {
                "uuid": _uuid.UUID("06753693-d892-77f0-8000-ce71bf7ebbba"),
                "description": "testing",
            },
        )

        # source has string uuid
        # nb. as of now Upgrade is the only table using proper UUID
        imp = self.make_importer(
            model_class=model.Upgrade, fields=["uuid", "description"]
        )
        result = imp.normalize_source_object(
            {"uuid": "06753693d89277f08000ce71bf7ebbba", "description": "testing"}
        )
        self.assertEqual(
            result,
            {
                "uuid": _uuid.UUID("06753693-d892-77f0-8000-ce71bf7ebbba"),
                "description": "testing",
            },
        )


class MockMixinHandler(mod.FromCsvToSqlalchemyHandlerMixin, ToSqlalchemyHandler):
    ToImporterBase = ToSqlalchemy


class TestFromCsvToSqlalchemyHandlerMixin(DataTestCase):

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

    def test_get_target_model(self):
        with patch.object(
            mod.FromCsvToSqlalchemyHandlerMixin, "define_importers", return_value={}
        ):
            handler = self.make_handler()
            self.assertRaises(NotImplementedError, handler.get_target_model)

    def test_define_importers(self):
        model = self.app.model
        with patch.object(
            mod.FromCsvToSqlalchemyHandlerMixin, "get_target_model", return_value=model
        ):
            handler = self.make_handler()
            importers = handler.define_importers()
            self.assertIn("Setting", importers)
            self.assertTrue(issubclass(importers["Setting"], mod.FromCsv))
            self.assertTrue(issubclass(importers["Setting"], ToSqlalchemy))
            self.assertIn("User", importers)
            self.assertIn("Person", importers)
            self.assertIn("Role", importers)

    def test_make_importer_factory(self):
        model = self.app.model
        with patch.object(
            mod.FromCsvToSqlalchemyHandlerMixin, "define_importers", return_value={}
        ):
            handler = self.make_handler()
            factory = handler.make_importer_factory(model.Setting, "Setting")
            self.assertTrue(issubclass(factory, mod.FromCsv))
            self.assertTrue(issubclass(factory, ToSqlalchemy))


class TestFromCsvToWutta(DataTestCase):

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

    def test_get_target_model(self):
        handler = self.make_handler()
        self.assertIs(handler.get_target_model(), self.app.model)