# (c) cavaliba.com - test - schema loader from django.test import TestCase import app_home.cache as cache from app_data.loader import load_broker from app_data.models import DataClass, DataSchema from app_data.schema import Schema class TestSchemaLoader(TestCase): def setUp(self): cache.clear() def test_schema_minimal(self): yaml_entry = { "classname": "_schema", "keyname": "testschema_minimal", } datalist = [yaml_entry] aaa = {"perms": ["p_schema_write"]} load_broker(datalist=datalist, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschema_minimal")) schema = Schema.from_name("testschema_minimal") self.assertIsNotNone(schema) self.assertEqual(schema.classname, "testschema_minimal") # Verify DataClass exists in database dataclass = DataClass.objects.filter(keyname="testschema_minimal").first() self.assertIsNotNone(dataclass) def test_load_create(self): yaml_entry = { "classname": "_schema", "_action": "create", "keyname": "testschema", "displayname": "Test Schema", } datalist = [yaml_entry] aaa = { "perms": [ "p_schema_write", ] } load_broker(datalist=datalist, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschema")) schema = Schema.from_name("testschema") self.assertIsNotNone(schema) self.assertEqual(schema.classname, "testschema") def test_load_create_update(self): # Create initial schema yaml_entry = { "classname": "_schema", "_action": "create", "keyname": "testschemaupdated", } datalist = [yaml_entry] aaa = { "perms": [ "p_schema_write", ] } load_broker(datalist=datalist, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschemaupdated")) schema = Schema.from_name("testschemaupdated") self.assertIsNotNone(schema) # Update schema with new displayname using create action yaml_entry_update = { "classname": "_schema", "_action": "create", "keyname": "testschemaupdated", "displayname": "Updated Schema Display Name", } datalist_update = [yaml_entry_update] load_broker(datalist=datalist_update, aaa=aaa) # Verify displayname was updated schema_updated = Schema.from_name("testschemaupdated") self.assertIsNotNone(schema_updated) self.assertEqual(schema_updated.displayname, "Updated Schema Display Name") def test_load_init(self): yaml_entry = { "classname": "_schema", "_action": "init", "keyname": "testschemainit", "displayname": "Test Schema Init", } datalist = [yaml_entry] aaa = { "perms": [ "p_schema_write", ] } # First init - should create schema load_broker(datalist=datalist, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschemainit")) schema = Schema.from_name("testschemainit") self.assertIsNotNone(schema) self.assertEqual(schema.classname, "testschemainit") self.assertEqual(schema.displayname, "Test Schema Init") # Second init with different displayname - should NOT update (init doesn't update) yaml_entry_second = { "classname": "_schema", "_action": "init", "keyname": "testschemainit", "displayname": "Different Display Name", } datalist_second = [yaml_entry_second] load_broker(datalist=datalist_second, aaa=aaa) # Verify displayname was NOT changed (init action skips if exists) schema_second = Schema.from_name("testschemainit") self.assertIsNotNone(schema_second) self.assertEqual(schema_second.displayname, "Test Schema Init") def test_load_update(self): yaml_entry = { "classname": "_schema", "_action": "update", "keyname": "testschemaupdate", "displayname": "Update Test Schema", } datalist = [yaml_entry] aaa = { "perms": [ "p_schema_write", ] } # Try to update non-existent schema - should fail silently _ = load_broker(datalist=datalist, aaa=aaa) # Verify schema was NOT created self.assertFalse(Schema.exists("testschemaupdate")) # Create schema first using create action yaml_entry_create = { "classname": "_schema", "_action": "create", "keyname": "testschemaupdate", "displayname": "Original Display Name", } datalist_create = [yaml_entry_create] load_broker(datalist=datalist_create, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschemaupdate")) schema = Schema.from_name("testschemaupdate") self.assertEqual(schema.displayname, "Original Display Name") # Now update with update action - should succeed yaml_entry_update = { "classname": "_schema", "_action": "update", "keyname": "testschemaupdate", "displayname": "Updated via Update Action", } datalist_update = [yaml_entry_update] load_broker(datalist=datalist_update, aaa=aaa) # Verify displayname was updated schema_updated = Schema.from_name("testschemaupdate") self.assertIsNotNone(schema_updated) self.assertEqual(schema_updated.displayname, "Updated via Update Action") def test_load_delete(self): # Create schema first yaml_entry_create = { "classname": "_schema", "_action": "create", "keyname": "testschemadelete", "displayname": "Schema to Delete", } datalist_create = [yaml_entry_create] aaa = { "perms": [ "p_schema_write", ] } load_broker(datalist=datalist_create, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschemadelete")) schema = Schema.from_name("testschemadelete") self.assertIsNotNone(schema) # Delete the schema using delete action yaml_entry_delete = { "classname": "_schema", "_action": "delete", "keyname": "testschemadelete", } datalist_delete = [yaml_entry_delete] load_broker(datalist=datalist_delete, aaa=aaa) # Verify schema was deleted self.assertFalse(Schema.exists("testschemadelete")) schema_deleted = Schema.from_name("testschemadelete") self.assertIsNone(schema_deleted) def test_load_disable(self): # Create schema first yaml_entry_create = { "classname": "_schema", "_action": "create", "keyname": "testschemadisable", "displayname": "Schema to Disable", } datalist_create = [yaml_entry_create] aaa = { "perms": [ "p_schema_write", ] } load_broker(datalist=datalist_create, aaa=aaa) # Verify schema was created and is enabled self.assertTrue(Schema.exists("testschemadisable")) schema = Schema.from_name("testschemadisable") self.assertIsNotNone(schema) self.assertTrue(schema.is_enabled) # Disable the schema yaml_entry_disable = { "classname": "_schema", "_action": "disable", "keyname": "testschemadisable", } datalist_disable = [yaml_entry_disable] load_broker(datalist=datalist_disable, aaa=aaa) # Verify schema is still there but disabled self.assertTrue(Schema.exists("testschemadisable")) schema_disabled = Schema.from_name("testschemadisable") self.assertIsNotNone(schema_disabled) self.assertFalse(schema_disabled.is_enabled) def test_load_enable(self): # Create and disable schema first yaml_entry_create = { "classname": "_schema", "_action": "create", "keyname": "testschemaenable", "displayname": "Schema to Enable", } datalist_create = [yaml_entry_create] aaa = { "perms": [ "p_schema_write", ] } load_broker(datalist=datalist_create, aaa=aaa) # Disable it yaml_entry_disable = { "classname": "_schema", "_action": "disable", "keyname": "testschemaenable", } datalist_disable = [yaml_entry_disable] load_broker(datalist=datalist_disable, aaa=aaa) # Verify schema is disabled schema = Schema.from_name("testschemaenable") self.assertFalse(schema.is_enabled) # Enable the schema yaml_entry_enable = { "classname": "_schema", "_action": "enable", "keyname": "testschemaenable", } datalist_enable = [yaml_entry_enable] load_broker(datalist=datalist_enable, aaa=aaa) # Verify schema is now enabled schema_enabled = Schema.from_name("testschemaenable") self.assertIsNotNone(schema_enabled) self.assertTrue(schema_enabled.is_enabled) def test_load_unknown_action(self): yaml_entry = { "classname": "_schema", "_action": "unknown_action", "keyname": "testschema", "displayname": "Test Schema", } datalist = [yaml_entry] aaa = { "perms": [ "p_schema_write", ] } result = load_broker(datalist=datalist, aaa=aaa) # Verify no schema was created (result should be 0 for unknown action) # load_broker returns count of processed items, which is 0 for unknown action self.assertEqual(result["count_ok"], 0) self.assertFalse(Schema.exists("testschema")) def test_load_attributes(self): yaml_entry = { "classname": "_schema", "_action": "init", "keyname": "testschemaattributes", "displayname": "Test Schema Attributes", "is_enabled": True, "page": "TestPage", "order": 10, "_options": { "icon": "fa-star", "keyname_mode": "auto", "keyname_label": "ID", "displayname_label": "Name", "p_admin": "p_admin_test", "p_create": "p_create_test", "p_read": "p_read_test", "p_update": "p_update_test", "p_delete": "p_delete_test", }, } datalist = [yaml_entry] aaa = { "perms": [ "p_schema_write", ] } load_broker(datalist=datalist, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschemaattributes")) schema = Schema.from_name("testschemaattributes") self.assertIsNotNone(schema) # Verify all attributes were set correctly self.assertEqual(schema.classname, "testschemaattributes") self.assertEqual(schema.displayname, "Test Schema Attributes") self.assertTrue(schema.is_enabled) self.assertEqual(schema.page, "TestPage") self.assertEqual(schema.order, 10) # Verify options self.assertEqual(schema.icon, "fa-star") self.assertEqual(schema.keyname_mode, "auto") self.assertEqual(schema.keyname_label, "ID") self.assertEqual(schema.displayname_label, "Name") # Verify permissions self.assertEqual(schema.p_admin, "p_admin_test") self.assertEqual(schema.p_create, "p_create_test") self.assertEqual(schema.p_read, "p_read_test") self.assertEqual(schema.p_update, "p_update_test") self.assertEqual(schema.p_delete, "p_delete_test") def test_load_field_init(self): yaml_entry = { "classname": "_schema", "keyname": "testschemafield", "displayname": "Test Schema Field", "is_enabled": True, "myfield": { "_action": "init", "dataformat": "string", "displayname": "My Field Name", "description": "Field description", }, } datalist = [yaml_entry] aaa = { "perms": [ "p_schema_write", ] } load_broker(datalist=datalist, aaa=aaa) # Verify schema was created self.assertTrue(Schema.exists("testschemafield")) schema = Schema.from_name("testschemafield") self.assertIsNotNone(schema) self.assertEqual(schema.displayname, "Test Schema Field") self.assertTrue(schema.is_enabled) # Verify field was created self.assertIn("myfield", schema.fields) field = schema.fields["myfield"] self.assertEqual(field["dataformat"], "string") self.assertEqual(field["displayname"], "My Field Name") self.assertEqual(field["description"], "Field description") def test_load_delete_with_fields(self): # Step 1: Create schema with multiple fields yaml_entry_create = { "classname": "_schema", "_action": "create", "keyname": "testschemafields", "displayname": "Schema with Multiple Fields", "field1": { "dataformat": "string", "displayname": "Field 1", }, "field2": { "dataformat": "int", "displayname": "Field 2", }, "field3": { "dataformat": "text", "displayname": "Field 3", }, } datalist_create = [yaml_entry_create] aaa = {"perms": ["p_schema_write"]} load_broker(datalist=datalist_create, aaa=aaa) # Step 2: Verify schema and fields are created in database # Check DataClass exists dataclass = DataClass.objects.filter(keyname="testschemafields").first() self.assertIsNotNone(dataclass, "DataClass 'testschemafields' should exist after creation") # Check DataSchema fields exist dataschema_fields = DataSchema.objects.filter(classname="testschemafields").count() self.assertEqual(dataschema_fields, 3, "Should have 3 fields in DataSchema") # Verify via Schema object schema = Schema.from_name("testschemafields") self.assertIsNotNone(schema) self.assertEqual(len(schema.fields), 3, "Schema should have 3 fields") self.assertIn("field1", schema.fields) self.assertIn("field2", schema.fields) self.assertIn("field3", schema.fields) # Step 3: Delete the schema using delete action yaml_entry_delete = { "classname": "_schema", "_action": "delete", "keyname": "testschemafields", } datalist_delete = [yaml_entry_delete] load_broker(datalist=datalist_delete, aaa=aaa) # Step 4: Verify schema and fields are removed from database # Check DataClass is deleted dataclass_deleted = DataClass.objects.filter(keyname="testschemafields").first() self.assertIsNone(dataclass_deleted, "DataClass 'testschemafields' should be deleted") # Check DataSchema fields are deleted dataschema_fields_deleted = DataSchema.objects.filter(classname="testschemafields").count() self.assertEqual(dataschema_fields_deleted, 0, "All DataSchema fields should be deleted") # Verify via Schema object schema_deleted = Schema.from_name("testschemafields") self.assertIsNone(schema_deleted, "Schema should return None after deletion") self.assertFalse(Schema.exists("testschemafields"), "Schema.exists() should return False")