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

feat: add wutta import-csv command

Browse source
This commit is contained in:
Lance Edgar 2024-12-05 21:19:06 -06:00
parent 84a8beaf46
commit f43a066341
19 changed files with 500 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

40
tests/importing/test_csv.py
View file

@ -14,6 +14,12 @@ class TestFromCsv(DataTestCase):
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)
@ -33,19 +39,36 @@ class TestFromCsv(DataTestCase):
model = self.app.model
imp = self.make_importer(model_class=model.Setting)
path = self.write_file('data.txt', '')
imp.input_file_path = path
# 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, path)
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)
path = self.write_file('data.txt', '')
imp.input_file_path = path
imp.input_file_path = self.data_path
imp.open_input_file()
imp.close_input_file()
self.assertFalse(hasattr(imp, 'input_reader'))
@ -55,12 +78,7 @@ class TestFromCsv(DataTestCase):
model = self.app.model
imp = self.make_importer(model_class=model.Setting)
path = self.write_file('data.csv', """\
name,value
foo,bar
foo2,bar2
""")
imp.input_file_path = path
imp.input_file_path = self.data_path
imp.open_input_file()
objects = imp.get_source_objects()
imp.close_input_file()

25
tests/importing/test_handlers.py
View file

@ -173,6 +173,31 @@ class TestImportHandler(DataTestCase):
self.assertRaises(KeyError, handler.get_importer, 'BunchOfNonsense', model_class=model.Setting)
class TestFromFileHandler(DataTestCase):
def make_handler(self, **kwargs):
return mod.FromFileHandler(self.config, **kwargs)
def test_process_data(self):
handler = self.make_handler()
path = self.write_file('data.txt', '')
with patch.object(mod.ImportHandler, 'process_data') as process_data:
# bare
handler.process_data()
process_data.assert_called_once_with()
# with file path
process_data.reset_mock()
handler.process_data(input_file_path=path)
process_data.assert_called_once_with(input_file_path=path)
# with folder
process_data.reset_mock()
handler.process_data(input_file_path=self.tempdir)
process_data.assert_called_once_with(input_file_dir=self.tempdir)
class TestToSqlalchemyHandler(DataTestCase):
def make_handler(self, **kwargs):