# (c) cavaliba.com - tests / migrator / legacy user-group-role dataformat rewrite """ migrate_legacy_schema_dataformat() (app_home/migrator.py, 5.3.0 block) rewrites data_field rows still holding a pre-4.0 reference dataformat - "user"/"group"/ "role" or "_user"/"_group"/"_role" - to the modern dataformat="schema" plus dataformat_ext= pair, and repairs the data_cache rows those fields wrote under the wrong format. The legacy state can only be produced by writing data_field directly: the import path (Schema.update_from_dict) normalizes it, which is exactly why existing rows were never fixed. force_legacy() below does that, then re-saves the instance so its EAV rows are rebuilt by the real code path (base Field.get_eav_format(), which emits the raw dataformat) rather than being faked. """ from django.test import TestCase, override_settings import app_home.cache as cache from app_data.data import Instance from app_data.models import DataEAV, DataSchema from app_data.related import get_related from app_home.migrator import migrate_legacy_schema_dataformat from tests import helper @override_settings(CAVALIBA_AUTH_MODE="unittest") class TestMigratorLegacyDataformat(TestCase): def setUp(self): """Two schemas referencing "user": "legacyref" (forced into the legacy state by each test) and "modernref" (already dataformat=schema/user, so it can witness that the migration leaves correct rows alone). Plus one user to be referenced.""" cache.clear() helper.add_admin_user(login="unittest") helper.add_user(login="jmurray") helper.add_schema( classname="legacyref", field_definition={ "owner": {"dataformat": "string", "displayname": "Owner"}, "label": {"dataformat": "string", "displayname": "Label"}, }, ) helper.add_instance( classname="legacyref", keyname="obj01", fields={"owner": "jmurray", "label": "kept"}, ) helper.add_schema( classname="modernref", field_definition={ "owner": {"dataformat": "schema", "dataformat_ext": "user"}, }, ) helper.add_instance(classname="modernref", keyname="obj02", fields={"owner": "jmurray"}) def force_legacy(self, dataformat=None, dataformat_ext=None, classname="legacyref"): """Write a legacy dataformat/dataformat_ext straight into data_field for .owner, bypassing Schema.update_from_dict()'s normalization, then re-save the instance so EavBatch rebuilds its data_cache rows from that legacy dataformat (the real origin of the stale format values).""" values = {} if dataformat is not None: values["dataformat"] = dataformat if dataformat_ext is not None: values["dataformat_ext"] = dataformat_ext DataSchema.objects.filter(classname=classname, keyname="owner").update(**values) cache.clear() instance = Instance.from_keyname(classname=classname, keyname="obj01", expand=True) instance.save(force=True, skip_revision=True) cache.clear() def field_row(self, classname="legacyref", keyname="owner"): return DataSchema.objects.get(classname=classname, keyname=keyname) def eav_formats(self, classname="legacyref", fieldname="owner"): return set( DataEAV.objects.filter(classname=classname, fieldname=fieldname).values_list( "format", flat=True ) ) # ---- data_field rewrite ---- def test_legacy_user_dataformat_is_rewritten(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "schema") self.assertEqual(row.dataformat_ext, "user") def test_legacy_group_dataformat_is_rewritten(self): self.force_legacy(dataformat="group") migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "schema") self.assertEqual(row.dataformat_ext, "group") def test_legacy_role_dataformat_is_rewritten(self): self.force_legacy(dataformat="role") migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "schema") self.assertEqual(row.dataformat_ext, "role") def test_underscore_prefixed_dataformat_is_rewritten(self): self.force_legacy(dataformat="_user") migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "schema") self.assertEqual(row.dataformat_ext, "user") def test_underscore_prefixed_dataformat_ext_is_stripped(self): self.force_legacy(dataformat="schema", dataformat_ext="_group") migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "schema") self.assertEqual(row.dataformat_ext, "group") def test_unrelated_field_is_untouched(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() row = self.field_row(keyname="label") self.assertEqual(row.dataformat, "string") self.assertEqual(row.dataformat_ext, "") def test_already_modern_field_is_untouched(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() row = self.field_row(classname="modernref") self.assertEqual(row.dataformat, "schema") self.assertEqual(row.dataformat_ext, "user") # ---- data_cache repair ---- def test_legacy_field_really_wrote_a_bad_eav_format(self): # guard against this file going vacuous: if the EAV write path ever starts # normalizing, the data_cache repair stops being what's under test self.force_legacy(dataformat="user") self.assertEqual(self.eav_formats(), {"user"}) def test_eav_format_is_repaired(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() self.assertEqual(self.eav_formats(), {"schema:user"}) def test_eav_format_is_repaired_for_underscored_ext(self): self.force_legacy(dataformat="schema", dataformat_ext="_group") self.assertEqual(self.eav_formats(), {"schema:_group"}) migrate_legacy_schema_dataformat() self.assertEqual(self.eav_formats(), {"schema:group"}) def test_modern_field_eav_format_is_untouched(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() self.assertEqual(self.eav_formats(classname="modernref"), {"schema:user"}) def test_related_lookup_is_broken_before_the_migration(self): # the actual user-visible damage: related.py filters on format="schema:user", # so the legacy field contributes nothing. get_related() always returns an # entry per schema (build_related_reply), hence the assert on count. self.force_legacy(dataformat="user") related = get_related(classname="user", keyname="jmurray") self.assertEqual(related["legacyref"]["count"], 0) # the modern schema referencing the same user is found all along self.assertEqual(related["modernref"]["count"], 1) def test_related_lookup_works_after_the_migration(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() related = get_related(classname="user", keyname="jmurray") self.assertEqual(related["legacyref"]["count"], 1) # ---- engine / UI recovery ---- def test_field_resolves_to_fieldschema_after_the_migration(self): from app_data.fieldtypes.field_schema import FieldSchema self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() instance = Instance.from_keyname(classname="legacyref", keyname="obj01", expand=True) self.assertIsInstance(instance.fields["owner"], FieldSchema) def test_ui_dataformat_is_no_longer_unknown(self): self.force_legacy(dataformat="user") instance = Instance.from_keyname(classname="legacyref", keyname="obj01", expand=True) self.assertEqual(instance.fields["owner"].get_ui_dataformat(), "unknown") migrate_legacy_schema_dataformat() instance = Instance.from_keyname(classname="legacyref", keyname="obj01", expand=True) self.assertEqual(instance.fields["owner"].get_ui_dataformat(), "schema") def test_stored_value_survives_the_migration(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() instance = Instance.from_keyname(classname="legacyref", keyname="obj01", expand=True) self.assertEqual(instance.fields["owner"].get_value(), ["jmurray"]) # ---- json: report only ---- def test_legacy_json_dataformat_is_left_alone(self): self.force_legacy(dataformat="json") migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "json") # ---- idempotency ---- def test_running_twice_is_a_noop(self): self.force_legacy(dataformat="user") migrate_legacy_schema_dataformat() migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "schema") self.assertEqual(row.dataformat_ext, "user") self.assertEqual(self.eav_formats(), {"schema:user"}) def test_running_on_a_clean_db_changes_nothing(self): migrate_legacy_schema_dataformat() row = self.field_row() self.assertEqual(row.dataformat, "string") self.assertEqual(self.field_row(classname="modernref").dataformat_ext, "user")