# (c) cavaliba.com - tests / data / instance from django.test import TestCase from django.utils import timezone import app_home.cache as cache from app_data.data import Instance from app_data.group import Group from app_data.models import DataEAV, DataInstance, DataRevision from app_data.user import User from tests import helper AAA = {"username": "testuser", "perms": ["p_data_admin"]} class InstanceGetDnTest(TestCase): def setUp(self): cache.clear() helper.add_schema(classname="testdn") helper.add_instance( classname="testdn", keyname="obj01", fields={"displayname": "Object One"} ) helper.add_instance(classname="testdn", keyname="obj02", fields={"displayname": ""}) helper.add_instance(classname="testdn", keyname="obj03") def test_returns_displayname(self): result = Instance.get_dn(classname="testdn", keyname="obj01") self.assertEqual(result, "Object One") def test_returns_none_for_missing_keyname(self): result = Instance.get_dn(classname="testdn", keyname="does_not_exist") self.assertIsNone(result) def test_returns_none_when_keyname_is_none(self): result = Instance.get_dn(classname="testdn", keyname=None) self.assertIsNone(result) def test_returns_none_when_classname_is_none(self): result = Instance.get_dn(classname=None, keyname="obj01") self.assertIsNone(result) def test_subclass_uses_own_classname(self): helper.add_user(login="userdn01", displayname="User DN One") result = User.get_dn(keyname="userdn01") self.assertEqual(result, "User DN One") def test_subclass_returns_none_for_missing(self): result = User.get_dn(keyname="no_such_user") self.assertIsNone(result) class InstanceGetAllKeynameTest(TestCase): def setUp(self): cache.clear() helper.add_schema(classname="testkn") helper.add_instance(classname="testkn", keyname="kn01") helper.add_instance(classname="testkn", keyname="kn02") helper.add_instance(classname="testkn", keyname="kn03") def test_returns_list(self): result = Instance.get_all_keyname(classname="testkn") self.assertIsInstance(result, list) def test_returns_all_keynames(self): result = Instance.get_all_keyname(classname="testkn") self.assertIn("kn01", result) self.assertIn("kn02", result) self.assertIn("kn03", result) def test_returns_empty_for_unknown_classname(self): result = Instance.get_all_keyname(classname="no_such_class") self.assertEqual(result, []) def test_returns_empty_when_classname_is_none(self): result = Instance.get_all_keyname(classname=None) self.assertEqual(result, []) def test_excludes_other_classname(self): helper.add_schema(classname="testkn_other") helper.add_instance(classname="testkn_other", keyname="other01") result = Instance.get_all_keyname(classname="testkn") self.assertNotIn("other01", result) def test_subclass_uses_own_classname(self): helper.add_group(name="grp_kn01") helper.add_group(name="grp_kn02") result = Group.get_all_keyname() self.assertIn("grp_kn01", result) self.assertIn("grp_kn02", result) class TestLastSync(TestCase): def setUp(self): cache.clear() helper.add_schema(classname="lstest") def test_last_sync_null_on_new_instance(self): instance = Instance(classname="lstest") self.assertIsNone(instance.last_sync) def test_last_sync_null_after_create_without_sync(self): instance = Instance(classname="lstest") instance.merge_import({"classname": "lstest", "keyname": "obj01"}, aaa=AAA) instance.save(aaa=AAA, action="create") iobj = DataInstance.objects.get(classname="lstest", keyname="obj01") self.assertIsNone(iobj.last_sync) def test_last_sync_set_when_assigned_before_create(self): instance = Instance(classname="lstest") instance.merge_import({"classname": "lstest", "keyname": "obj02"}, aaa=AAA) now = timezone.now() instance.last_sync = now instance.save(aaa=AAA, action="create") iobj = DataInstance.objects.get(classname="lstest", keyname="obj02") self.assertIsNotNone(iobj.last_sync) def test_last_sync_not_in_data_json(self): instance = Instance(classname="lstest") instance.merge_import({"classname": "lstest", "keyname": "obj03"}, aaa=AAA) instance.last_sync = timezone.now() instance.save(aaa=AAA, action="create") iobj = DataInstance.objects.get(classname="lstest", keyname="obj03") self.assertNotIn("last_sync", iobj.data_json) def test_last_sync_not_in_eav(self): instance = Instance(classname="lstest") instance.merge_import({"classname": "lstest", "keyname": "obj04"}, aaa=AAA) instance.last_sync = timezone.now() instance.save(aaa=AAA, action="create") eav_entries = DataEAV.objects.filter( classname="lstest", keyname="obj04", fieldname="last_sync" ) self.assertEqual(eav_entries.count(), 0) def test_last_sync_loaded_from_iobj(self): instance = Instance(classname="lstest") instance.merge_import({"classname": "lstest", "keyname": "obj05"}, aaa=AAA) now = timezone.now() instance.last_sync = now instance.save(aaa=AAA, action="create") loaded = Instance.from_keyname("lstest", "obj05") self.assertIsNotNone(loaded.last_sync) def test_last_update_unchanged_when_no_diff(self): helper.add_instance(classname="lstest", keyname="obj06") instance = Instance.from_keyname("lstest", "obj06") last_update_before = instance.last_update instance.merge_import({"classname": "lstest", "keyname": "obj06"}, aaa=AAA) instance.save(aaa=AAA, action="update") instance2 = Instance.from_keyname("lstest", "obj06") self.assertEqual(instance2.last_update, last_update_before) class TestWriteToDb(TestCase): def setUp(self): cache.clear() helper.add_schema(classname="wtest") def test_write_to_db_saves_instance(self): instance = Instance(classname="wtest") instance.keyname = "obj01" instance.displayname = "Obj 01" r = instance.save(action="create") self.assertTrue(r) self.assertTrue(DataInstance.objects.filter(classname="wtest", keyname="obj01").exists()) def test_write_to_db_not_called_by_update_on_no_change(self): helper.add_instance(classname="wtest", keyname="obj01") instance = Instance.from_keyname("wtest", "obj01") last_update_before = instance.last_update # merge_import with identical data — save() should detect no diff vs DB instance.merge_import({"classname": "wtest", "keyname": "obj01"}, aaa=AAA) r = instance.save(aaa=AAA, action="update") self.assertTrue(r) # last_update should not have changed since no diff instance2 = Instance.from_keyname("wtest", "obj01") self.assertEqual(instance2.last_update, last_update_before) def test_create_adds_revision(self): instance = Instance(classname="wtest") instance.merge_import({"classname": "wtest", "keyname": "obj02"}, aaa=AAA) r = instance.save(aaa=AAA, action="create") self.assertTrue(r) self.assertEqual( DataRevision.objects.filter( classname="wtest", keyname="obj02", action="create" ).count(), 1, ) def test_update_adds_revision_on_change(self): helper.add_instance(classname="wtest", keyname="obj03") instance = Instance.from_keyname("wtest", "obj03") instance.merge_import( {"classname": "wtest", "keyname": "obj03", "displayname": "Changed"}, aaa=AAA ) r = instance.save(aaa=AAA, action="update") self.assertTrue(r) self.assertEqual( DataRevision.objects.filter( classname="wtest", keyname="obj03", action="update" ).count(), 1, ) def test_update_skips_revision_on_no_change(self): helper.add_instance(classname="wtest", keyname="obj04", fields={"displayname": "Same"}) instance = Instance.from_keyname("wtest", "obj04") instance.merge_import( {"classname": "wtest", "keyname": "obj04", "displayname": "Same"}, aaa=AAA ) r = instance.save(aaa=AAA, action="update") self.assertTrue(r) self.assertEqual( DataRevision.objects.filter( classname="wtest", keyname="obj04", action="update" ).count(), 0, ) def test_delete_adds_revision(self): helper.add_instance(classname="wtest", keyname="obj06") instance = Instance.from_keyname("wtest", "obj06") instance.aaa_revision = AAA r = instance.delete() self.assertTrue(r) self.assertEqual( DataRevision.objects.filter( classname="wtest", keyname="obj06", action="delete" ).count(), 1, )