feat: add support for wutta export-csv command
This commit is contained in:
parent
c873cc462e
commit
61deaad251
21 changed files with 1186 additions and 16 deletions
|
|
@ -1,6 +1,7 @@
|
|||
# -*- coding: utf-8; -*-
|
||||
|
||||
import inspect
|
||||
import sys
|
||||
from unittest import TestCase
|
||||
from unittest.mock import patch, Mock
|
||||
|
||||
|
|
@ -67,6 +68,75 @@ class TestImportCommandHandler(DataTestCase):
|
|||
transaction_comment="hello world",
|
||||
)
|
||||
|
||||
def test_run_missing_input(self):
|
||||
handler = self.make_handler(
|
||||
import_handler="wuttasync.importing.csv:FromCsvToWutta"
|
||||
)
|
||||
|
||||
class Object:
|
||||
def __init__(self, **kw):
|
||||
self.__dict__.update(kw)
|
||||
|
||||
# fails without input_file_path
|
||||
with patch.object(sys, "exit") as exit_:
|
||||
exit_.side_effect = RuntimeError
|
||||
ctx = Object(
|
||||
params={},
|
||||
parent=Object(params={}),
|
||||
)
|
||||
try:
|
||||
handler.run(ctx)
|
||||
except RuntimeError:
|
||||
pass
|
||||
exit_.assert_called_once_with(1)
|
||||
|
||||
# runs with input_file_path
|
||||
with patch.object(sys, "exit") as exit_:
|
||||
exit_.side_effect = RuntimeError
|
||||
ctx = Object(
|
||||
params={"input_file_path": self.tempdir},
|
||||
parent=Object(
|
||||
params={},
|
||||
),
|
||||
)
|
||||
self.assertRaises(FileNotFoundError, handler.run, ctx)
|
||||
exit_.assert_not_called()
|
||||
|
||||
def test_run_missing_output(self):
|
||||
handler = self.make_handler(
|
||||
import_handler="wuttasync.exporting.csv:FromWuttaToCsv"
|
||||
)
|
||||
|
||||
class Object:
|
||||
def __init__(self, **kw):
|
||||
self.__dict__.update(kw)
|
||||
|
||||
# fails without output_file_path
|
||||
with patch.object(sys, "exit") as exit_:
|
||||
exit_.side_effect = RuntimeError
|
||||
ctx = Object(
|
||||
params={},
|
||||
parent=Object(params={}),
|
||||
)
|
||||
try:
|
||||
handler.run(ctx)
|
||||
except RuntimeError:
|
||||
pass
|
||||
exit_.assert_called_once_with(1)
|
||||
|
||||
# runs with output_file_path
|
||||
with patch.object(sys, "exit") as exit_:
|
||||
exit_.side_effect = RuntimeError
|
||||
ctx = Object(
|
||||
params={"output_file_path": self.tempdir},
|
||||
parent=Object(
|
||||
params={},
|
||||
),
|
||||
)
|
||||
# self.assertRaises(FileNotFoundError, handler.run, ctx)
|
||||
handler.run(ctx)
|
||||
exit_.assert_not_called()
|
||||
|
||||
def test_list_models(self):
|
||||
handler = self.make_handler(
|
||||
import_handler="wuttasync.importing.csv:FromCsvToWutta"
|
||||
|
|
@ -96,6 +166,23 @@ class TestImporterCommand(TestCase):
|
|||
self.assertIn("dry_run", sig2.parameters)
|
||||
|
||||
|
||||
class TestFileExporterCommand(TestCase):
|
||||
|
||||
def test_basic(self):
|
||||
def myfunc(ctx, **kwargs):
|
||||
pass
|
||||
|
||||
sig1 = inspect.signature(myfunc)
|
||||
self.assertIn("kwargs", sig1.parameters)
|
||||
self.assertNotIn("dry_run", sig1.parameters)
|
||||
self.assertNotIn("output_file_path", sig1.parameters)
|
||||
wrapt = mod.file_export_command(myfunc)
|
||||
sig2 = inspect.signature(wrapt)
|
||||
self.assertNotIn("kwargs", sig2.parameters)
|
||||
self.assertIn("dry_run", sig2.parameters)
|
||||
self.assertIn("output_file_path", sig2.parameters)
|
||||
|
||||
|
||||
class TestFileImporterCommand(TestCase):
|
||||
|
||||
def test_basic(self):
|
||||
|
|
|
|||
22
tests/cli/test_export_csv.py
Normal file
22
tests/cli/test_export_csv.py
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
# -*- coding: utf-8; -*-
|
||||
|
||||
from unittest import TestCase
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from wuttasync.cli import export_csv as mod, ImportCommandHandler
|
||||
|
||||
|
||||
class TestExportCsv(TestCase):
|
||||
|
||||
def test_basic(self):
|
||||
params = {
|
||||
"models": [],
|
||||
"create": True,
|
||||
"update": True,
|
||||
"delete": False,
|
||||
"dry_run": True,
|
||||
}
|
||||
ctx = MagicMock(params=params)
|
||||
with patch.object(ImportCommandHandler, "run") as run:
|
||||
mod.export_csv(ctx)
|
||||
run.assert_called_once_with(ctx)
|
||||
99
tests/exporting/test_base.py
Normal file
99
tests/exporting/test_base.py
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
# -*- coding: utf-8; -*-
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from wuttjamaican.testing import DataTestCase
|
||||
|
||||
from wuttasync.exporting import base as mod, ExportHandler
|
||||
|
||||
|
||||
class TestToFile(DataTestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.setup_db()
|
||||
self.handler = ExportHandler(self.config)
|
||||
|
||||
def make_exporter(self, **kwargs):
|
||||
kwargs.setdefault("handler", self.handler)
|
||||
return mod.ToFile(self.config, **kwargs)
|
||||
|
||||
def test_setup(self):
|
||||
model = self.app.model
|
||||
|
||||
# output file is opened
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
self.assertFalse(exp.dry_run)
|
||||
with patch.object(exp, "open_output_file") as open_output_file:
|
||||
exp.setup()
|
||||
open_output_file.assert_called_once_with()
|
||||
|
||||
# but not if in dry run mode
|
||||
with patch.object(self.handler, "dry_run", new=True):
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
self.assertTrue(exp.dry_run)
|
||||
with patch.object(exp, "open_output_file") as open_output_file:
|
||||
exp.setup()
|
||||
open_output_file.assert_not_called()
|
||||
|
||||
def test_teardown(self):
|
||||
model = self.app.model
|
||||
|
||||
# output file is closed
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
self.assertFalse(exp.dry_run)
|
||||
with patch.object(exp, "close_output_file") as close_output_file:
|
||||
exp.teardown()
|
||||
close_output_file.assert_called_once_with()
|
||||
|
||||
# but not if in dry run mode
|
||||
with patch.object(self.handler, "dry_run", new=True):
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
self.assertTrue(exp.dry_run)
|
||||
with patch.object(exp, "close_output_file") as close_output_file:
|
||||
exp.teardown()
|
||||
close_output_file.assert_not_called()
|
||||
|
||||
def test_get_output_file_path(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
|
||||
# output path must be set
|
||||
self.assertRaises(ValueError, exp.get_output_file_path)
|
||||
|
||||
# path is guessed from dir+filename
|
||||
path1 = self.write_file("data1.txt", "")
|
||||
exp.output_file_path = self.tempdir
|
||||
exp.output_file_name = "data1.txt"
|
||||
self.assertEqual(exp.get_output_file_path(), path1)
|
||||
|
||||
# path can be explicitly set
|
||||
path2 = self.write_file("data2.txt", "")
|
||||
exp.output_file_path = path2
|
||||
self.assertEqual(exp.get_output_file_path(), path2)
|
||||
|
||||
def test_get_output_file_name(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
|
||||
# name cannot be guessed
|
||||
self.assertRaises(NotImplementedError, exp.get_output_file_name)
|
||||
|
||||
# name can be explicitly set
|
||||
exp.output_file_name = "data.txt"
|
||||
self.assertEqual(exp.get_output_file_name(), "data.txt")
|
||||
|
||||
def test_open_output_file(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
self.assertRaises(NotImplementedError, exp.open_output_file)
|
||||
|
||||
def test_close_output_file(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
|
||||
path = self.write_file("data.txt", "")
|
||||
with open(path, "wt") as f:
|
||||
exp.output_file = f
|
||||
with patch.object(f, "close") as close:
|
||||
exp.close_output_file()
|
||||
close.assert_called_once_with()
|
||||
209
tests/exporting/test_csv.py
Normal file
209
tests/exporting/test_csv.py
Normal file
|
|
@ -0,0 +1,209 @@
|
|||
# -*- coding: utf-8; -*-
|
||||
|
||||
import csv
|
||||
import io
|
||||
from unittest.mock import patch
|
||||
|
||||
from wuttjamaican.testing import DataTestCase
|
||||
|
||||
from wuttasync.exporting import csv as mod, ExportHandler
|
||||
from wuttasync.importing import FromWuttaHandler, FromWutta
|
||||
|
||||
|
||||
class TestToCsv(DataTestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.setup_db()
|
||||
self.handler = ExportHandler(self.config)
|
||||
|
||||
def make_exporter(self, **kwargs):
|
||||
kwargs.setdefault("handler", self.handler)
|
||||
kwargs.setdefault("output_file_path", self.tempdir)
|
||||
return mod.ToCsv(self.config, **kwargs)
|
||||
|
||||
def test_get_output_file_name(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
|
||||
# name can be guessed
|
||||
self.assertEqual(exp.get_output_file_name(), "Setting.csv")
|
||||
|
||||
# name can be explicitly set
|
||||
exp.output_file_name = "data.txt"
|
||||
self.assertEqual(exp.get_output_file_name(), "data.txt")
|
||||
|
||||
def test_open_output_file(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
self.assertIsNone(exp.output_file)
|
||||
self.assertIsNone(exp.output_writer)
|
||||
exp.open_output_file()
|
||||
try:
|
||||
self.assertIsInstance(exp.output_file, io.TextIOBase)
|
||||
self.assertIsInstance(exp.output_writer, csv.DictWriter)
|
||||
finally:
|
||||
exp.output_file.close()
|
||||
|
||||
def test_close_output_file(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
|
||||
self.assertIsNone(exp.output_file)
|
||||
self.assertIsNone(exp.output_writer)
|
||||
exp.open_output_file()
|
||||
self.assertIsNotNone(exp.output_file)
|
||||
self.assertIsNotNone(exp.output_writer)
|
||||
exp.close_output_file()
|
||||
self.assertIsNone(exp.output_file)
|
||||
self.assertIsNone(exp.output_writer)
|
||||
|
||||
def test_coerce_csv(self):
|
||||
model = self.app.model
|
||||
|
||||
# string value
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
result = exp.coerce_csv({"name": "foo", "value": "bar"})
|
||||
self.assertEqual(result, {"name": "foo", "value": "bar"})
|
||||
|
||||
# null value converts to empty string
|
||||
result = exp.coerce_csv({"name": "foo", "value": None})
|
||||
self.assertEqual(result, {"name": "foo", "value": ""})
|
||||
|
||||
# float value passed thru as-is
|
||||
result = exp.coerce_csv({"name": "foo", "value": 12.34})
|
||||
self.assertEqual(result, {"name": "foo", "value": 12.34})
|
||||
self.assertIsInstance(result["value"], float)
|
||||
|
||||
def test_update_target_object(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(model_class=model.Setting)
|
||||
|
||||
exp.setup()
|
||||
|
||||
with patch.object(exp, "output_writer") as output_writer:
|
||||
|
||||
# writer is called for normal run
|
||||
data = {"name": "foo", "value": "bar"}
|
||||
exp.update_target_object(None, data)
|
||||
output_writer.writerow.assert_called_once_with(data)
|
||||
|
||||
# but not called for dry run
|
||||
output_writer.writerow.reset_mock()
|
||||
with patch.object(self.handler, "dry_run", new=True):
|
||||
exp.update_target_object(None, data)
|
||||
output_writer.writerow.assert_not_called()
|
||||
|
||||
exp.teardown()
|
||||
|
||||
|
||||
class MockMixinExporter(mod.FromSqlalchemyToCsvMixin, FromWutta, mod.ToCsv):
|
||||
pass
|
||||
|
||||
|
||||
class TestFromSqlalchemyToCsvMixin(DataTestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.setup_db()
|
||||
self.handler = ExportHandler(self.config)
|
||||
|
||||
def make_exporter(self, **kwargs):
|
||||
kwargs.setdefault("handler", self.handler)
|
||||
return MockMixinExporter(self.config, **kwargs)
|
||||
|
||||
def test_model_title(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(source_model_class=model.Setting)
|
||||
|
||||
# default comes from model class
|
||||
self.assertEqual(exp.get_model_title(), "Setting")
|
||||
|
||||
# but can override
|
||||
exp.model_title = "Widget"
|
||||
self.assertEqual(exp.get_model_title(), "Widget")
|
||||
|
||||
def test_get_simple_fields(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(source_model_class=model.Setting)
|
||||
|
||||
# default comes from model class
|
||||
self.assertEqual(exp.get_simple_fields(), ["name", "value"])
|
||||
|
||||
# but can override
|
||||
exp.simple_fields = ["name"]
|
||||
self.assertEqual(exp.get_simple_fields(), ["name"])
|
||||
|
||||
# no default if no model class
|
||||
exp = self.make_exporter()
|
||||
self.assertEqual(exp.get_simple_fields(), [])
|
||||
|
||||
def test_normalize_source_object(self):
|
||||
model = self.app.model
|
||||
exp = self.make_exporter(source_model_class=model.Setting)
|
||||
setting = model.Setting(name="foo", value="bar")
|
||||
data = exp.normalize_source_object(setting)
|
||||
self.assertEqual(data, {"name": "foo", "value": "bar"})
|
||||
|
||||
def test_make_object(self):
|
||||
model = self.app.model
|
||||
|
||||
# normal
|
||||
exp = self.make_exporter(source_model_class=model.Setting)
|
||||
obj = exp.make_object()
|
||||
self.assertIsInstance(obj, model.Setting)
|
||||
|
||||
# no model_class
|
||||
exp = self.make_exporter()
|
||||
self.assertRaises(TypeError, exp.make_object)
|
||||
|
||||
|
||||
class MockMixinHandler(
|
||||
mod.FromSqlalchemyToCsvHandlerMixin, FromWuttaHandler, mod.ToCsvHandler
|
||||
):
|
||||
FromImporterBase = FromWutta
|
||||
|
||||
|
||||
class TestFromSqlalchemyToCsvHandlerMixin(DataTestCase):
|
||||
|
||||
def make_handler(self, **kwargs):
|
||||
return MockMixinHandler(self.config, **kwargs)
|
||||
|
||||
def test_get_source_model(self):
|
||||
with patch.object(
|
||||
mod.FromSqlalchemyToCsvHandlerMixin, "define_importers", return_value={}
|
||||
):
|
||||
handler = self.make_handler()
|
||||
self.assertRaises(NotImplementedError, handler.get_source_model)
|
||||
|
||||
def test_define_importers(self):
|
||||
model = self.app.model
|
||||
with patch.object(
|
||||
mod.FromSqlalchemyToCsvHandlerMixin, "get_source_model", return_value=model
|
||||
):
|
||||
handler = self.make_handler()
|
||||
importers = handler.define_importers()
|
||||
self.assertIn("Setting", importers)
|
||||
self.assertTrue(issubclass(importers["Setting"], FromWutta))
|
||||
self.assertTrue(issubclass(importers["Setting"], mod.ToCsv))
|
||||
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.FromSqlalchemyToCsvHandlerMixin, "define_importers", return_value={}
|
||||
):
|
||||
handler = self.make_handler()
|
||||
factory = handler.make_importer_factory(model.Setting, "Setting")
|
||||
self.assertTrue(issubclass(factory, FromWutta))
|
||||
self.assertTrue(issubclass(factory, mod.ToCsv))
|
||||
|
||||
|
||||
class TestFromWuttaToCsv(DataTestCase):
|
||||
|
||||
def make_handler(self, **kwargs):
|
||||
return mod.FromWuttaToCsv(self.config, **kwargs)
|
||||
|
||||
def test_get_source_model(self):
|
||||
handler = self.make_handler()
|
||||
self.assertIs(handler.get_source_model(), self.app.model)
|
||||
4
tests/exporting/test_handlers.py
Normal file
4
tests/exporting/test_handlers.py
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
# -*- coding: utf-8; -*-
|
||||
|
||||
# nothing to test yet really, just ensuring coverage
|
||||
from wuttasync.exporting import handlers as mod
|
||||
Loading…
Add table
Add a link
Reference in a new issue