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

feat: add warnings mode for import/export handlers, commands

Browse source 
can now specify `--warn` for import/export CLI, to get diff email when
changes occur.

this also adds `get_import_handler()` and friends, via app provider.

also declare email settings for the 2 existing importers
This commit is contained in:
Lance Edgar 2025-12-20 15:32:15 -06:00
parent 1e7722de91
commit 19574ea4a0
18 changed files with 1150 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

126
tests/importing/test_handlers.py
View file

@ -2,12 +2,18 @@
from collections import OrderedDict
from unittest.mock import patch
from uuid import UUID
from wuttjamaican.testing import DataTestCase
from wuttasync.importing import handlers as mod, Importer, ToSqlalchemy
class FromFooToBar(mod.ImportHandler):
source_key = "foo"
target_key = "bar"
class TestImportHandler(DataTestCase):
def make_handler(self, **kwargs):
@ -30,10 +36,10 @@ class TestImportHandler(DataTestCase):
def test_get_key(self):
handler = self.make_handler()
self.assertEqual(handler.get_key(), "to_None.from_None.import")
self.assertEqual(handler.get_key(), "import.to_None.from_None")
with patch.multiple(mod.ImportHandler, source_key="csv", target_key="wutta"):
self.assertEqual(handler.get_key(), "to_wutta.from_csv.import")
self.assertEqual(handler.get_key(), "import.to_wutta.from_csv")
def test_get_spec(self):
handler = self.make_handler()
@ -149,15 +155,41 @@ class TestImportHandler(DataTestCase):
kw = {}
result = handler.consume_kwargs(kw)
self.assertIs(result, kw)
self.assertEqual(result, {})
# captures dry-run flag
# dry_run (not consumed)
self.assertFalse(handler.dry_run)
kw["dry_run"] = True
result = handler.consume_kwargs(kw)
self.assertIs(result, kw)
self.assertIn("dry_run", kw)
self.assertTrue(kw["dry_run"])
self.assertTrue(handler.dry_run)
# warnings (consumed)
self.assertFalse(handler.warnings)
kw["warnings"] = True
result = handler.consume_kwargs(kw)
self.assertIs(result, kw)
self.assertNotIn("warnings", kw)
self.assertTrue(handler.warnings)
# warnings_recipients (consumed)
self.assertIsNone(handler.warnings_recipients)
kw["warnings_recipients"] = "bob@example.com"
result = handler.consume_kwargs(kw)
self.assertIs(result, kw)
self.assertNotIn("warnings_recipients", kw)
self.assertEqual(handler.warnings_recipients, ["bob@example.com"])
# warnings_max_diffs (consumed)
self.assertEqual(handler.warnings_max_diffs, 15)
kw["warnings_max_diffs"] = 30
result = handler.consume_kwargs(kw)
self.assertIs(result, kw)
self.assertNotIn("warnings_max_diffs", kw)
self.assertEqual(handler.warnings_max_diffs, 30)
def test_define_importers(self):
handler = self.make_handler()
importers = handler.define_importers()
@ -187,6 +219,94 @@ class TestImportHandler(DataTestCase):
KeyError, handler.get_importer, "BunchOfNonsense", model_class=model.Setting
)
def test_get_warnings_email_key(self):
handler = FromFooToBar(self.config)
# default
key = handler.get_warnings_email_key()
self.assertEqual(key, "import_to_bar_from_foo_warning")
# override
handler.warnings_email_key = "from_foo_to_bar"
key = handler.get_warnings_email_key()
self.assertEqual(key, "from_foo_to_bar")
def test_process_changes(self):
model = self.app.model
handler = self.make_handler()
email_handler = self.app.get_email_handler()
handler.process_started = self.app.localtime()
alice = model.User(username="alice")
bob = model.User(username="bob")
charlie = model.User(username="charlie")
changes = {
"User": (
[
(
alice,
{
"uuid": UUID("06946d64-1ebf-79db-8000-ce40345044fe"),
"username": "alice",
},
),
],
[
(
bob,
{
"uuid": UUID("06946d64-1ebf-7a8c-8000-05d78792b084"),
"username": "bob",
},
{
"uuid": UUID("06946d64-1ebf-7a8c-8000-05d78792b084"),
"username": "bobbie",
},
),
],
[
(
charlie,
{
"uuid": UUID("06946d64-1ebf-7ad4-8000-1ba52f720c48"),
"username": "charlie",
},
),
],
),
}
# no email if not in warnings mode
self.assertFalse(handler.warnings)
with patch.object(self.app, "send_email") as send_email:
handler.process_changes(changes)
send_email.assert_not_called()
# email sent (to default recip) if in warnings mode
handler.warnings = True
self.config.setdefault("wutta.email.default.to", "admin@example.com")
with patch.object(email_handler, "deliver_message") as deliver_message:
handler.process_changes(changes)
deliver_message.assert_called_once()
args, kwargs = deliver_message.call_args
self.assertEqual(kwargs, {"recips": None})
self.assertEqual(len(args), 1)
msg = args[0]
self.assertEqual(msg.to, ["admin@example.com"])
# can override email recip
handler.warnings_recipients = ["bob@example.com"]
with patch.object(email_handler, "deliver_message") as deliver_message:
handler.process_changes(changes)
deliver_message.assert_called_once()
args, kwargs = deliver_message.call_args
self.assertEqual(kwargs, {"recips": None})
self.assertEqual(len(args), 1)
msg = args[0]
self.assertEqual(msg.to, ["bob@example.com"])
class TestFromFileHandler(DataTestCase):