diff --git a/layers/admin.py b/layers/admin.py index 81d85f2..a074e91 100644 --- a/layers/admin.py +++ b/layers/admin.py @@ -237,6 +237,7 @@ def export_theme_details(self, request, queryset): class ThemeAdmin(ImportExportMixin,admin.ModelAdmin): + fixture_import_session_key = "layers.theme_fixture_import_rows" list_display = ('display_name', 'name', 'order', 'date_modified', 'is_top_theme', 'primary_site', 'preview_site') search_fields = ['display_name', 'name',] form = ThemeForm @@ -308,6 +309,144 @@ class Media: js = ['theme_admin.js',] change_form_template = os.path.join(CURRENT_DIR, 'templates', 'admin', 'layers', 'Theme', 'change_form.html') + change_list_template = os.path.join(CURRENT_DIR, 'templates', 'admin', 'layers', 'Theme', 'change_list.html') + + def values_match(self, current_value, new_value): + from datetime import date, datetime + from uuid import UUID + + if current_value == new_value: + return True + if isinstance(current_value, UUID): + return current_value == UUID(str(new_value)) + if isinstance(current_value, (date, datetime)): + return current_value == type(current_value).fromisoformat(str(new_value)) + return False + + def _fixture_preview_rows(self, rows): + preview_rows = [] + for row in rows: + model_label = row['model'] + uuid_value = row['uuid'] + existing_record = None + if uuid_value: + model = apps.get_model(model_label) + if any(field.name == 'uuid' for field in model._meta.fields): + manager = getattr(model, 'all_objects', model._base_manager) + existing_record = manager.filter(uuid=uuid_value).first() + + changes = [] + if existing_record is not None: + for field_name, new_value in row['fields'].items(): + current_value = getattr(existing_record, field_name) + if not self.values_match(current_value, new_value): + changes.append({ + 'name': field_name, + 'current_value': current_value, + 'new_value': new_value, + }) + + if existing_record is not None: + action = f'Update existing record: "{existing_record}"' + elif uuid_value: + action = 'Create new record' + else: + action = 'Create or merge relationship record' + + preview_rows.append({ + 'model': model_label, + 'uuid': uuid_value, + 'action': action, + 'fields': row['fields'], + 'changes': changes, + }) + return preview_rows + + def get_urls(self): + urls = super().get_urls() + custom_urls = [ + path( + 'import-fixture/', + self.admin_site.admin_view(self.import_fixture), + name='layers_theme_import_fixture', + ), + ] + return custom_urls + urls + + def import_fixture(self, request): + if not self.has_change_permission(request): + raise PermissionDenied + + changelist_url = reverse('admin:layers_theme_changelist') + context = { + **self.admin_site.each_context(request), + 'opts': self.model._meta, + 'title': 'Import theme fixture', + 'changelist_url': changelist_url, + } + + if request.method == 'POST' and 'cancel' in request.POST: + request.session.pop(self.fixture_import_session_key, None) + return redirect(changelist_url) + + if request.method == 'POST' and 'confirm' in request.POST: + rows = request.session.get(self.fixture_import_session_key) + if rows is None: + context['error'] = 'No validated fixture is available to import.' + return render(request, 'admin/layers/Theme/import_theme_fixture.html', context) + + try: + result = import_fixture_rows( + rows, + dry_run=False, + associate_all_sites=True, + missing_ref_policy='error', + duplicate_uuid_policy='error', + ) + except ValueError as error: + context['error'] = str(error) + context['fixture_rows'] = rows + context['preview_rows'] = self._fixture_preview_rows(rows) + return render(request, 'admin/layers/Theme/import_theme_fixture.html', context) + + request.session.pop(self.fixture_import_session_key, None) + self.message_user(request, 'Imported {} fixture rows.'.format(result['imported'])) + return redirect(changelist_url) + + if request.method == 'POST': + fixture_file = request.FILES.get('fixture_file') + if fixture_file is None: + context['error'] = 'Choose a fixture JSON file to import.' + return render(request, 'admin/layers/Theme/import_theme_fixture.html', context) + + try: + rows = json.loads(fixture_file.read().decode('utf-8')) + if not isinstance(rows, list): + raise ValueError('Fixture JSON must contain a list of rows.') + for row in rows: + validate_node_shape(row) + except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error: + context['error'] = 'Upload valid JSON fixture data: {}'.format(error) + return render(request, 'admin/layers/Theme/import_theme_fixture.html', context) + + try: + result = import_fixture_rows( + rows, + dry_run=True, + associate_all_sites=True, + missing_ref_policy='error', + duplicate_uuid_policy='error', + ) + except ValueError as error: + context['error'] = str(error) + return render(request, 'admin/layers/Theme/import_theme_fixture.html', context) + + request.session[self.fixture_import_session_key] = rows + context['fixture_rows'] = rows + context['preview_result'] = result + context['preview_rows'] = self._fixture_preview_rows(rows) + + return render(request, 'admin/layers/Theme/import_theme_fixture.html', context) def get_queryset(self, request): @@ -946,8 +1085,9 @@ def _fixture_preview_rows(self, rows): existing_record = None if uuid_value: model = apps.get_model(model_label) - manager = getattr(model, 'all_objects', model._base_manager) - existing_record = manager.filter(uuid=uuid_value).first() + if any(field.name == 'uuid' for field in model._meta.fields): + manager = getattr(model, 'all_objects', model._base_manager) + existing_record = manager.filter(uuid=uuid_value).first() changes = [] if existing_record is not None: diff --git a/layers/fixture_import.py b/layers/fixture_import.py index 53305bb..4ef9a55 100644 --- a/layers/fixture_import.py +++ b/layers/fixture_import.py @@ -8,6 +8,7 @@ from __future__ import annotations from django.apps import apps +from django.contrib.contenttypes.models import ContentType from django.contrib.sites.models import Site from django.db import transaction @@ -22,6 +23,8 @@ LAYER_MODEL = "layers.layer" +THEME_MODEL = "layers.theme" +CHILD_ORDER_MODEL = "layers.childorder" MULTILAYER_ASSOCIATION_MODEL = "layers.multilayerassociation" MULTILAYER_DIMENSION_MODEL = "layers.multilayerdimension" MULTILAYER_DIMENSION_VALUE_MODEL = "layers.multilayerdimensionvalue" @@ -151,7 +154,7 @@ def _execute_import(): # First pass: upsert UUID-keyed rows that do not require relation remaps. for row in rows: model_label = row.get(NODE_MODEL_KEY) - if model_label not in {LAYER_MODEL, ATTRIBUTE_INFO_MODEL, LOOKUP_INFO_MODEL}: + if model_label not in {LAYER_MODEL, THEME_MODEL, ATTRIBUTE_INFO_MODEL, LOOKUP_INFO_MODEL}: continue row_uuid = normalize_uuid(row.get(NODE_UUID_KEY)) @@ -175,6 +178,38 @@ def _execute_import(): if model_label == LAYER_MODEL and associate_all_sites: row_obj.site.set(Site.objects.all()) + elif model_label == THEME_MODEL and associate_all_sites: + row_obj.site.set(Site.objects.all()) + + # Second pass: resolve theme child-order relations by UUID. + for row in rows: + if row.get(NODE_MODEL_KEY) != CHILD_ORDER_MODEL: + continue + + relations = row.get(NODE_RELATIONS_KEY, {}) + parent_theme = _resolve_ref_instance( + relations.get("parent_theme"), + missing_ref_policy, + ) + content_object = _resolve_ref_instance( + relations.get("content_object"), + missing_ref_policy, + ) + ChildOrder = apps.get_model(CHILD_ORDER_MODEL) + content_type = ContentType.objects.get_for_model(content_object) + child_order = ChildOrder.objects.filter( + parent_theme=parent_theme, + content_type=content_type, + object_id=content_object.pk, + ).first() + if child_order is None: + child_order = ChildOrder( + parent_theme=parent_theme, + content_type=content_type, + object_id=content_object.pk, + ) + _apply_fields(child_order, row.get(NODE_FIELDS_KEY, {})) + child_order.save() # Second pass: resolve layer m2m attribute refs by UUID. for row in rows: @@ -347,7 +382,8 @@ def _execute_import(): "dry_run": True, } - _execute_import() + with transaction.atomic(): + _execute_import() return { "imported": len(rows), "dry_run": False, diff --git a/layers/templates/admin/layers/Theme/change_list.html b/layers/templates/admin/layers/Theme/change_list.html new file mode 100644 index 0000000..979be0d --- /dev/null +++ b/layers/templates/admin/layers/Theme/change_list.html @@ -0,0 +1,8 @@ +{% extends "admin/change_list.html" %} + +{% block object-tools-items %} +
  • + Import theme fixture +
    • + {{ block.super }} +{% endblock %} diff --git a/layers/templates/admin/layers/Theme/import_theme_fixture.html b/layers/templates/admin/layers/Theme/import_theme_fixture.html new file mode 100644 index 0000000..3babbec --- /dev/null +++ b/layers/templates/admin/layers/Theme/import_theme_fixture.html @@ -0,0 +1,52 @@ +{% extends "admin/base_site.html" %} + +{% block content %} +

    {{ title }}

    + + {% if error %} +

    {{ error }}

    + {% endif %} + + {% if preview_rows %} +

    Preview

    + + + + + + + + + + + + {% for row in preview_rows %} + + + + + + + + {% endfor %} + +
    ModelUUIDActionNameChanges
    {{ row.model }}{{ row.uuid }}{{ row.action }}{{ row.fields.name }} + {% for change in row.changes %} + {{ change.name }}: {{ change.current_value }} -> {{ change.new_value }}{% if not forloop.last %}; {% endif %} + {% empty %} + None + {% endfor %} +
    + {% endif %} + +
    + {% csrf_token %} + {% if preview_rows %} + + + {% else %} +

    + + {% endif %} +
    +{% endblock %} diff --git a/layers/tests/test_fixture_import.py b/layers/tests/test_fixture_import.py index f2c03a5..0df2e76 100644 --- a/layers/tests/test_fixture_import.py +++ b/layers/tests/test_fixture_import.py @@ -7,6 +7,7 @@ from layers.models import ( AttributeInfo, Companionship, + ChildOrder, Layer, LayerArcFeatureService, LayerArcREST, @@ -17,6 +18,7 @@ MultilayerAssociation, MultilayerDimension, MultilayerDimensionValue, + Theme, ) try: @@ -229,6 +231,94 @@ def test_duplicate_uuid_rows_with_conflicting_fields_raise_error(self): import_fixture_rows(fixture_rows, **self._import_kwargs()) +class ThemeFixtureImportPR09Test(TestCase): + """Theme fixture import tests for non-UUID ChildOrder identity.""" + + def _require_importer(self): + self.assertIsNotNone( + import_fixture_rows, + "importer API missing: expected layers.fixture_import.import_fixture_rows", + ) + + def _import_kwargs(self): + return { + "dry_run": False, + "associate_all_sites": True, + "missing_ref_policy": "error", + "duplicate_uuid_policy": "error", + } + + def test_child_order_source_id_collision_creates_new_relationship(self): + self._require_importer() + + parent_theme = Theme.all_objects.create( + name="Existing Theme", + display_name="Existing Theme", + ) + layer = Layer.all_objects.create(name="Existing Layer", layer_type="WMS") + imported_parent_theme = Theme.all_objects.create( + name="Imported Theme", + display_name="Imported Theme", + ) + imported_layer = Layer.all_objects.create( + name="Imported Layer", + layer_type="WMS", + ) + child_order = ChildOrder.objects.create( + parent_theme=parent_theme, + content_object=layer, + order=3, + ) + original_pk = child_order.pk + original_date_created = child_order.date_created + original_date_modified = child_order.date_modified + original_parent_theme_id = child_order.parent_theme_id + original_content_type_id = child_order.content_type_id + original_object_id = child_order.object_id + + fixture_rows = [ + build_node( + model="layers.childorder", + source_pk=original_pk, + uuid_value=None, + fields={"order": 17}, + relations={ + "parent_theme": build_ref( + model="layers.theme", + source_pk=1001, + uuid_value=imported_parent_theme.uuid, + ), + "content_object": build_ref( + model="layers.layer", + source_pk=1002, + uuid_value=imported_layer.uuid, + ), + }, + ) + ] + + import_fixture_rows(fixture_rows, **self._import_kwargs()) + + child_orders = ChildOrder.objects.all() + self.assertEqual(child_orders.count(), 2) + + child_order.refresh_from_db() + self.assertEqual(child_order.pk, original_pk) + self.assertEqual(child_order.order, 3) + self.assertEqual(child_order.parent_theme_id, original_parent_theme_id) + self.assertEqual(child_order.content_type_id, original_content_type_id) + self.assertEqual(child_order.object_id, original_object_id) + self.assertEqual(child_order.date_created, original_date_created) + self.assertEqual(child_order.date_modified, original_date_modified) + + imported_child_order = ChildOrder.objects.get( + parent_theme=imported_parent_theme, + object_id=imported_layer.pk, + ) + self.assertNotEqual(imported_child_order.pk, original_pk) + self.assertEqual(imported_child_order.order, 17) + + class LayerFixtureImportPR06Test(TestCase): """PR06 contract tests for associated model import behavior.""" diff --git a/layers/tests/test_theme_fixture_import_admin.py b/layers/tests/test_theme_fixture_import_admin.py new file mode 100644 index 0000000..55091a6 --- /dev/null +++ b/layers/tests/test_theme_fixture_import_admin.py @@ -0,0 +1,268 @@ +import json +from unittest.mock import Mock, patch +from uuid import uuid4 + +from django.contrib.auth import get_user_model +from django.core.files.uploadedfile import SimpleUploadedFile +from django.test import TestCase +from django.urls import reverse + +from layers.fixture_contract import build_node +from layers.admin import export_theme_details +from layers.models import ChildOrder, Layer, Theme + + +class ThemeFixtureImportAdminTest(TestCase): + """Contract tests for the Theme fixture import admin workflow.""" + + upload_url_name = "admin:layers_theme_import_fixture" + + def setUp(self): + self.upload_url = reverse(self.upload_url_name) + self.superuser = get_user_model().objects.create_superuser( + username="theme-fixture-admin", + email="theme-fixture-admin@example.com", + password="password", + ) + self.theme_uuid = uuid4() + self.valid_rows = [ + build_node( + model="layers.theme", + source_pk=1001, + uuid_value=self.theme_uuid, + fields={ + "name": "Uploaded Fixture Theme", + "display_name": "Uploaded Fixture Theme", + "theme_type": "checkbox", + "order": 10, + "is_visible": True, + "is_top_theme": False, + }, + relations={}, + ) + ] + + def _fixture_file(self, rows=None, name="themes.json"): + return SimpleUploadedFile( + name, + json.dumps(rows if rows is not None else self.valid_rows).encode("utf-8"), + content_type="application/json", + ) + + def test_upload_view_requires_theme_change_permission(self): + response = self.client.get(self.upload_url) + self.assertEqual(response.status_code, 302) + + user = get_user_model().objects.create_user( + username="no-theme-permission", + password="password", + is_staff=True, + ) + self.client.force_login(user) + response = self.client.get(self.upload_url) + self.assertEqual(response.status_code, 403) + + @patch("layers.admin.import_fixture_rows") + def test_valid_upload_runs_dry_run_and_shows_confirmation(self, import_fixture_rows): + self.client.force_login(self.superuser) + import_fixture_rows.return_value = {"imported": 1, "dry_run": True} + + response = self.client.post( + self.upload_url, + {"fixture_file": self._fixture_file()}, + ) + + self.assertEqual(response.status_code, 200) + import_fixture_rows.assert_called_once_with( + self.valid_rows, + dry_run=True, + associate_all_sites=True, + missing_ref_policy="error", + duplicate_uuid_policy="error", + ) + self.assertContains(response, "Uploaded Fixture Theme") + self.assertContains(response, "Confirm") + self.assertEqual(Theme.all_objects.count(), 0) + + @patch("layers.admin.import_fixture_rows") + def test_exported_theme_fixture_preview_has_no_field_differences(self, import_fixture_rows): + self.client.force_login(self.superuser) + theme = Theme.all_objects.create( + name="Preview Theme", + display_name="Preview Theme", + ) + layer = Layer.all_objects.create(name="Preview Layer", layer_type="WMS") + ChildOrder.objects.create( + parent_theme=theme, + content_object=layer, + order=4, + ) + + export_response = export_theme_details( + Mock(), + Mock(), + Theme.all_objects.filter(pk=theme.pk), + ) + fixture_file = SimpleUploadedFile( + "exported-theme.json", + export_response.content, + content_type="application/json", + ) + import_fixture_rows.return_value = {"imported": 0, "dry_run": True} + + response = self.client.post( + self.upload_url, + {"fixture_file": fixture_file}, + ) + + self.assertEqual(response.status_code, 200) + import_fixture_rows.assert_called_once() + self.assertTrue(response.context["preview_rows"]) + self.assertTrue(all(not row["changes"] for row in response.context["preview_rows"])) + self.assertContains(response, "Create or merge relationship record") + + @patch("layers.admin.import_fixture_rows") + def test_preview_reports_uuid_matched_theme_updates(self, import_fixture_rows): + self.client.force_login(self.superuser) + existing_theme = Theme.all_objects.create( + name="Current Theme Name", + display_name="Current Theme Name", + uuid=self.theme_uuid, + ) + fixture_rows = [ + build_node( + model="layers.theme", + source_pk=1001, + uuid_value=existing_theme.uuid, + fields={ + "name": "Imported Theme Name", + "display_name": "Imported Theme Name", + "theme_type": "checkbox", + "order": 10, + "is_visible": True, + "is_top_theme": False, + }, + relations={}, + ) + ] + import_fixture_rows.return_value = {"imported": 1, "dry_run": True} + + response = self.client.post( + self.upload_url, + {"fixture_file": self._fixture_file(fixture_rows)}, + ) + + self.assertEqual(response.status_code, 200) + self.assertContains(response, "Update existing record") + self.assertContains(response, "Current Theme Name") + self.assertContains(response, "Imported Theme Name") + + @patch("layers.admin.import_fixture_rows") + def test_source_id_collision_does_not_change_uuid_first_import_policy(self, import_fixture_rows): + self.client.force_login(self.superuser) + Theme.all_objects.create( + name="Existing Theme", + display_name="Existing Theme", + uuid=uuid4(), + id=1001, + ) + import_fixture_rows.return_value = {"imported": 1, "dry_run": True} + + response = self.client.post( + self.upload_url, + {"fixture_file": self._fixture_file()}, + ) + + self.assertEqual(response.status_code, 200) + import_fixture_rows.assert_called_once_with( + self.valid_rows, + dry_run=True, + associate_all_sites=True, + missing_ref_policy="error", + duplicate_uuid_policy="error", + ) + + @patch("layers.admin.import_fixture_rows") + def test_confirmation_executes_staged_fixture_not_posted_payload(self, import_fixture_rows): + self.client.force_login(self.superuser) + import_fixture_rows.return_value = {"imported": 1, "dry_run": True} + + preview_response = self.client.post( + self.upload_url, + {"fixture_file": self._fixture_file()}, + ) + self.assertEqual(preview_response.status_code, 200) + + import_fixture_rows.reset_mock() + import_fixture_rows.return_value = {"imported": 1, "dry_run": False} + tampered_rows = [ + build_node( + model="layers.theme", + source_pk=9999, + uuid_value=uuid4(), + fields={ + "name": "Tampered Theme", + "display_name": "Tampered Theme", + "theme_type": "checkbox", + "order": 10, + "is_visible": True, + "is_top_theme": False, + }, + relations={}, + ) + ] + + response = self.client.post( + self.upload_url, + {"confirm": "1", "fixture_file": self._fixture_file(tampered_rows)}, + ) + + self.assertEqual(response.status_code, 302) + import_fixture_rows.assert_called_once_with( + self.valid_rows, + dry_run=False, + associate_all_sites=True, + missing_ref_policy="error", + duplicate_uuid_policy="error", + ) + + @patch("layers.admin.import_fixture_rows") + def test_cancel_discards_staged_fixture_without_import(self, import_fixture_rows): + self.client.force_login(self.superuser) + import_fixture_rows.return_value = {"imported": 1, "dry_run": True} + + self.client.post(self.upload_url, {"fixture_file": self._fixture_file()}) + import_fixture_rows.reset_mock() + + response = self.client.post(self.upload_url, {"cancel": "1"}) + + self.assertRedirects(response, reverse("admin:layers_theme_changelist")) + import_fixture_rows.assert_not_called() + self.assertEqual(Theme.all_objects.count(), 0) + + @patch("layers.admin.import_fixture_rows") + def test_confirm_without_preview_does_not_write(self, import_fixture_rows): + self.client.force_login(self.superuser) + + response = self.client.post(self.upload_url, {"confirm": "1"}) + + self.assertEqual(response.status_code, 200) + import_fixture_rows.assert_not_called() + self.assertContains(response, "No validated fixture") + self.assertEqual(Theme.all_objects.count(), 0) + + @patch("layers.admin.import_fixture_rows") + def test_invalid_json_does_not_write(self, import_fixture_rows): + self.client.force_login(self.superuser) + invalid_file = SimpleUploadedFile( + "broken.json", + b"{not valid json", + content_type="application/json", + ) + + response = self.client.post(self.upload_url, {"fixture_file": invalid_file}) + + self.assertEqual(response.status_code, 200) + import_fixture_rows.assert_not_called() + self.assertContains(response, "valid JSON") + self.assertEqual(Theme.all_objects.count(), 0)