# (c) cavaliba.com - test - schema loader 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 from django.test import TestCase 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")