# (c) cavaliba.com - test_loader_logging.py from django.test import TestCase import app_home.cache as cache from app_data.loader import load_broker from app_data.models import DataRevision from app_home.models import CavalibaLog from tests import helper AAA = {"username": "loaderuser", "perms": ["p_data_admin"], "sync_mode": False} AAA_SYNC = {"username": "syncuser", "perms": ["p_data_admin"], "sync_mode": True} class TestLoaderLogging(TestCase): def setUp(self): cache.clear() helper.add_schema(classname="lltest") def test_loader_create_logs_info(self): load_broker([{"classname": "lltest", "_action": "create", "keyname": "obj01"}], aaa=AAA) self.assertTrue( CavalibaLog.objects.filter(level="INFO", action="create", status="OK").exists() ) def test_loader_create_adds_revision(self): load_broker([{"classname": "lltest", "_action": "create", "keyname": "obj02"}], aaa=AAA) self.assertEqual( DataRevision.objects.filter( classname="lltest", keyname="obj02", action="create" ).count(), 1, ) def test_loader_update_logs_when_changed(self): helper.add_instance(classname="lltest", keyname="obj03") load_broker( [ { "classname": "lltest", "_action": "update", "keyname": "obj03", "displayname": "Changed", } ], aaa=AAA, ) self.assertTrue( CavalibaLog.objects.filter(level="INFO", action="update", status="OK").exists() ) def test_loader_update_logs_even_when_unchanged(self): # loader logs on every successful save() regardless of diff; only the # revision (see test_loader_update_no_revision_when_unchanged) is # gated on an actual change helper.add_instance(classname="lltest", keyname="obj04") CavalibaLog.objects.all().delete() load_broker( [{"classname": "lltest", "_action": "update", "keyname": "obj04"}], aaa=AAA, ) self.assertTrue( CavalibaLog.objects.filter(level="INFO", action="update", status="OK").exists() ) def test_loader_update_adds_revision_when_changed(self): helper.add_instance(classname="lltest", keyname="obj05") load_broker( [ { "classname": "lltest", "_action": "update", "keyname": "obj05", "displayname": "Changed", } ], aaa=AAA, ) self.assertEqual( DataRevision.objects.filter( classname="lltest", keyname="obj05", action="update" ).count(), 1, ) def test_loader_update_no_revision_when_unchanged(self): helper.add_instance(classname="lltest", keyname="obj06") load_broker( [{"classname": "lltest", "_action": "update", "keyname": "obj06"}], aaa=AAA, ) self.assertEqual( DataRevision.objects.filter( classname="lltest", keyname="obj06", action="update" ).count(), 0, ) def test_loader_delete_logs_info(self): helper.add_instance(classname="lltest", keyname="obj07") load_broker([{"classname": "lltest", "_action": "delete", "keyname": "obj07"}], aaa=AAA) self.assertTrue( CavalibaLog.objects.filter(level="INFO", action="delete", status="OK").exists() ) def test_loader_enable_logs_info(self): helper.add_instance(classname="lltest", keyname="obj08", fields={"is_enabled": "false"}) load_broker([{"classname": "lltest", "_action": "enable", "keyname": "obj08"}], aaa=AAA) self.assertTrue( CavalibaLog.objects.filter(level="INFO", action="enable", status="OK").exists() ) def test_loader_disable_logs_info(self): helper.add_instance(classname="lltest", keyname="obj09") load_broker([{"classname": "lltest", "_action": "disable", "keyname": "obj09"}], aaa=AAA) self.assertTrue( CavalibaLog.objects.filter(level="INFO", action="disable", status="OK").exists() ) def test_loader_username_in_revision(self): load_broker([{"classname": "lltest", "_action": "create", "keyname": "obj10"}], aaa=AAA) rev = DataRevision.objects.filter( classname="lltest", keyname="obj10", action="create" ).first() self.assertIsNotNone(rev) self.assertEqual(rev.username, "loaderuser") def test_loader_init_on_absent_adds_create_revision(self): load_broker([{"classname": "lltest", "_action": "init", "keyname": "obj12"}], aaa=AAA) self.assertEqual( DataRevision.objects.filter( classname="lltest", keyname="obj12", action="create" ).count(), 1, ) def test_loader_init_on_existing_is_silent_noop(self): helper.add_instance(classname="lltest", keyname="obj13") count_before = DataRevision.objects.filter(classname="lltest", keyname="obj13").count() load_broker([{"classname": "lltest", "_action": "init", "keyname": "obj13"}], aaa=AAA) self.assertEqual( DataRevision.objects.filter(classname="lltest", keyname="obj13").count(), count_before, ) self.assertFalse( CavalibaLog.objects.filter( action="init", data__icontains="obj13", status="DENY" ).exists() ) def test_loader_sync_mode_sets_last_sync_on_create(self): from app_data.models import DataInstance load_broker( [{"classname": "lltest", "_action": "create", "keyname": "obj11"}], aaa=AAA_SYNC ) iobj = DataInstance.objects.get(classname="lltest", keyname="obj11") self.assertIsNotNone(iobj.last_sync) def test_loader_sync_mode_sets_last_sync_on_changed_update(self): from app_data.models import DataInstance helper.add_instance(classname="lltest", keyname="obj12") load_broker( [ { "classname": "lltest", "_action": "update", "keyname": "obj12", "displayname": "Updated", } ], aaa=AAA_SYNC, ) iobj = DataInstance.objects.get(classname="lltest", keyname="obj12") self.assertIsNotNone(iobj.last_sync) def test_loader_sync_mode_sets_last_sync_on_unchanged_update(self): # last_sync is a touch tracker ("this object was seen in a sync run"), # not a "was modified with sync_mode" flag — it must update on every # sync_mode save, even when nothing else about the object changed from app_data.models import DataInstance helper.add_instance(classname="lltest", keyname="obj13") load_broker( [{"classname": "lltest", "_action": "update", "keyname": "obj13"}], aaa=AAA_SYNC, ) iobj = DataInstance.objects.get(classname="lltest", keyname="obj13") self.assertIsNotNone(iobj.last_sync) def test_loader_sync_mode_does_not_bump_last_update_on_unchanged_touch(self): # the touch itself is not a "modification" — last_update (unlike # last_sync) must stay untouched when there's no real diff from app_data.models import DataInstance helper.add_instance(classname="lltest", keyname="obj15") last_update_before = DataInstance.objects.get( classname="lltest", keyname="obj15" ).last_update load_broker( [{"classname": "lltest", "_action": "update", "keyname": "obj15"}], aaa=AAA_SYNC, ) iobj = DataInstance.objects.get(classname="lltest", keyname="obj15") self.assertEqual(iobj.last_update, last_update_before) def test_loader_sync_mode_no_revision_on_unchanged_update(self): # sync_mode alone triggers a write (to touch last_sync) but that write # carries no real diff, so it must not create a revision entry either helper.add_instance(classname="lltest", keyname="obj14") count_before = DataRevision.objects.filter(classname="lltest", keyname="obj14").count() load_broker( [{"classname": "lltest", "_action": "update", "keyname": "obj14"}], aaa=AAA_SYNC, ) self.assertEqual( DataRevision.objects.filter(classname="lltest", keyname="obj14").count(), count_before, )