test(migrate): add migration planner, wizard, and integration tests

- Unit tests for MigrationPlanner diff classification
- Unit tests for MigrationWizard (41 tests incl. adversarial inputs)
- Integration test for drop_recreate flow
- Field modifier ordering integration tests (INDEXEMPTY, INDEXMISSING, etc.)

Author: nkanu17

tests/integration/test_field_modifier_ordering_integration.py

@@ -399,6 +399,233 @@ def test_indexmissing_enables_ismissing_query(self, client, redis_url, worker_id
         index.delete(drop=True)
 
 
+class TestIndexEmptyIntegration:
+    """Integration tests for INDEXEMPTY functionality."""
+
+    def test_text_field_index_empty_creates_successfully(
+        self, client, redis_url, worker_id
+    ):
+        """Test that INDEXEMPTY on text field allows index creation."""
+        skip_if_search_version_below_for_indexmissing(client)
+        schema_dict = {
+            "index": {
+                "name": f"test_text_empty_{worker_id}",
+                "prefix": f"textempty_{worker_id}:",
+                "storage_type": "hash",
+            },
+            "fields": [
+                {
+                    "name": "description",
+                    "type": "text",
+                    "attrs": {"index_empty": True},
+                }
+            ],
+        }
+
+        schema = IndexSchema.from_dict(schema_dict)
+        index = SearchIndex(schema=schema, redis_url=redis_url)
+        index.create(overwrite=True)
+
+        # Verify index was created
+        info = client.execute_command("FT.INFO", f"test_text_empty_{worker_id}")
+        assert info is not None
+
+        # Create documents with empty and non-empty values
+        client.hset(f"textempty_{worker_id}:1", "description", "has content")
+        client.hset(f"textempty_{worker_id}:2", "description", "")
+        client.hset(f"textempty_{worker_id}:3", "description", "more content")
+
+        # Search should work, empty string doc should be indexed
+        result = client.execute_command(
+            "FT.SEARCH",
+            f"test_text_empty_{worker_id}",
+            "*",
+        )
+        # All 3 docs should be found
+        assert result[0] == 3
+
+        # Cleanup
+        client.delete(
+            f"textempty_{worker_id}:1",
+            f"textempty_{worker_id}:2",
+            f"textempty_{worker_id}:3",
+        )
+        index.delete(drop=True)
+
+    def test_tag_field_index_empty_creates_successfully(
+        self, client, redis_url, worker_id
+    ):
+        """Test that INDEXEMPTY on tag field allows index creation."""
+        skip_if_search_version_below_for_indexmissing(client)
+        schema_dict = {
+            "index": {
+                "name": f"test_tag_empty_{worker_id}",
+                "prefix": f"tagempty_{worker_id}:",
+                "storage_type": "hash",
+            },
+            "fields": [
+                {
+                    "name": "category",
+                    "type": "tag",
+                    "attrs": {"index_empty": True},
+                }
+            ],
+        }
+
+        schema = IndexSchema.from_dict(schema_dict)
+        index = SearchIndex(schema=schema, redis_url=redis_url)
+        index.create(overwrite=True)
+
+        # Verify index was created
+        info = client.execute_command("FT.INFO", f"test_tag_empty_{worker_id}")
+        assert info is not None
+
+        # Create documents with empty and non-empty values
+        client.hset(f"tagempty_{worker_id}:1", "category", "electronics")
+        client.hset(f"tagempty_{worker_id}:2", "category", "")
+        client.hset(f"tagempty_{worker_id}:3", "category", "books")
+
+        # Search should work
+        result = client.execute_command(
+            "FT.SEARCH",
+            f"test_tag_empty_{worker_id}",
+            "*",
+        )
+        # All 3 docs should be found
+        assert result[0] == 3
+
+        # Cleanup
+        client.delete(
+            f"tagempty_{worker_id}:1",
+            f"tagempty_{worker_id}:2",
+            f"tagempty_{worker_id}:3",
+        )
+        index.delete(drop=True)
+
+
+class TestUnfModifierIntegration:
+    """Integration tests for UNF (un-normalized form) modifier."""
+
+    def test_text_field_unf_requires_sortable(self, client, redis_url, worker_id):
+        """Test that UNF on text field works only when sortable is also True."""
+        skip_if_search_version_below_for_indexmissing(client)
+        schema_dict = {
+            "index": {
+                "name": f"test_text_unf_{worker_id}",
+                "prefix": f"textunf_{worker_id}:",
+                "storage_type": "hash",
+            },
+            "fields": [
+                {
+                    "name": "title",
+                    "type": "text",
+                    "attrs": {"sortable": True, "unf": True},
+                }
+            ],
+        }
+
+        schema = IndexSchema.from_dict(schema_dict)
+        index = SearchIndex(schema=schema, redis_url=redis_url)
+
+        # Should create successfully
+        index.create(overwrite=True)
+
+        info = client.execute_command("FT.INFO", f"test_text_unf_{worker_id}")
+        assert info is not None
+
+        index.delete(drop=True)
+
+    def test_numeric_field_unf_with_sortable(self, client, redis_url, worker_id):
+        """Test that UNF on numeric field works when sortable is True."""
+        skip_if_search_version_below_for_indexmissing(client)
+        schema_dict = {
+            "index": {
+                "name": f"test_num_unf_{worker_id}",
+                "prefix": f"numunf_{worker_id}:",
+                "storage_type": "hash",
+            },
+            "fields": [
+                {
+                    "name": "price",
+                    "type": "numeric",
+                    "attrs": {"sortable": True, "unf": True},
+                }
+            ],
+        }
+
+        schema = IndexSchema.from_dict(schema_dict)
+        index = SearchIndex(schema=schema, redis_url=redis_url)
+
+        # Should create successfully
+        index.create(overwrite=True)
+
+        info = client.execute_command("FT.INFO", f"test_num_unf_{worker_id}")
+        assert info is not None
+
+        index.delete(drop=True)
+
+
+class TestNoIndexModifierIntegration:
+    """Integration tests for NOINDEX modifier."""
+
+    def test_noindex_with_sortable_allows_sorting_not_searching(
+        self, client, redis_url, worker_id
+    ):
+        """Test that NOINDEX field can be sorted but not searched."""
+        schema_dict = {
+            "index": {
+                "name": f"test_noindex_{worker_id}",
+                "prefix": f"noindex_{worker_id}:",
+                "storage_type": "hash",
+            },
+            "fields": [
+                {
+                    "name": "searchable",
+                    "type": "text",
+                },
+                {
+                    "name": "sort_only",
+                    "type": "numeric",
+                    "attrs": {"sortable": True, "no_index": True},
+                },
+            ],
+        }
+
+        schema = IndexSchema.from_dict(schema_dict)
+        index = SearchIndex(schema=schema, redis_url=redis_url)
+        index.create(overwrite=True)
+
+        # Add test documents
+        client.hset(
+            f"noindex_{worker_id}:1", mapping={"searchable": "hello", "sort_only": 10}
+        )
+        client.hset(
+            f"noindex_{worker_id}:2", mapping={"searchable": "world", "sort_only": 5}
+        )
+        client.hset(
+            f"noindex_{worker_id}:3", mapping={"searchable": "test", "sort_only": 15}
+        )
+
+        # Sorting by no_index field should work
+        result = client.execute_command(
+            "FT.SEARCH",
+            f"test_noindex_{worker_id}",
+            "*",
+            "SORTBY",
+            "sort_only",
+            "ASC",
+        )
+        assert result[0] == 3
+
+        # Cleanup
+        client.delete(
+            f"noindex_{worker_id}:1",
+            f"noindex_{worker_id}:2",
+            f"noindex_{worker_id}:3",
+        )
+        index.delete(drop=True)
+
+
 class TestFieldTypeModifierSupport:
     """Test that field types only support their documented modifiers."""
 

tests/integration/test_migration_v1.py

@@ -0,0 +1,127 @@
+import uuid
+
+import yaml
+
+from redisvl.index import SearchIndex
+from redisvl.migration import MigrationExecutor, MigrationPlanner, MigrationValidator
+from redisvl.migration.utils import load_migration_plan, schemas_equal
+from redisvl.redis.utils import array_to_buffer
+
+
+def test_drop_recreate_plan_apply_validate_flow(redis_url, worker_id, tmp_path):
+    unique_id = str(uuid.uuid4())[:8]
+    index_name = f"migration_v1_{worker_id}_{unique_id}"
+    prefix = f"migration_v1:{worker_id}:{unique_id}"
+
+    source_index = SearchIndex.from_dict(
+        {
+            "index": {
+                "name": index_name,
+                "prefix": prefix,
+                "storage_type": "hash",
+            },
+            "fields": [
+                {"name": "doc_id", "type": "tag"},
+                {"name": "title", "type": "text"},
+                {"name": "price", "type": "numeric"},
+                {
+                    "name": "embedding",
+                    "type": "vector",
+                    "attrs": {
+                        "algorithm": "hnsw",
+                        "dims": 3,
+                        "distance_metric": "cosine",
+                        "datatype": "float32",
+                    },
+                },
+            ],
+        },
+        redis_url=redis_url,
+    )
+
+    docs = [
+        {
+            "doc_id": "1",
+            "title": "alpha",
+            "price": 1,
+            "category": "news",
+            "embedding": array_to_buffer([0.1, 0.2, 0.3], "float32"),
+        },
+        {
+            "doc_id": "2",
+            "title": "beta",
+            "price": 2,
+            "category": "sports",
+            "embedding": array_to_buffer([0.2, 0.1, 0.4], "float32"),
+        },
+    ]
+
+    source_index.create(overwrite=True)
+    source_index.load(docs, id_field="doc_id")
+
+    patch_path = tmp_path / "schema_patch.yaml"
+    patch_path.write_text(
+        yaml.safe_dump(
+            {
+                "version": 1,
+                "changes": {
+                    "add_fields": [
+                        {
+                            "name": "category",
+                            "type": "tag",
+                            "attrs": {"separator": ","},
+                        }
+                    ],
+                    "remove_fields": ["price"],
+                    "update_fields": [{"name": "title", "attrs": {"sortable": True}}],
+                },
+            },
+            sort_keys=False,
+        )
+    )
+
+    plan_path = tmp_path / "migration_plan.yaml"
+    planner = MigrationPlanner()
+    plan = planner.create_plan(
+        index_name,
+        redis_url=redis_url,
+        schema_patch_path=str(patch_path),
+    )
+    assert plan.diff_classification.supported is True
+    planner.write_plan(plan, str(plan_path))
+
+    query_check_path = tmp_path / "query_checks.yaml"
+    query_check_path.write_text(
+        yaml.safe_dump({"fetch_ids": ["1", "2"]}, sort_keys=False)
+    )
+
+    executor = MigrationExecutor()
+    report = executor.apply(
+        load_migration_plan(str(plan_path)),
+        redis_url=redis_url,
+        query_check_file=str(query_check_path),
+    )
+
+    assert report.result == "succeeded"
+    assert report.validation.schema_match is True
+    assert report.validation.doc_count_match is True
+    assert report.validation.key_sample_exists is True
+    assert report.validation.indexing_failures_delta == 0
+    assert not report.validation.errors
+    assert report.benchmark_summary.documents_indexed_per_second is not None
+
+    live_index = SearchIndex.from_existing(index_name, redis_url=redis_url)
+    assert schemas_equal(live_index.schema.to_dict(), plan.merged_target_schema)
+
+    validator = MigrationValidator()
+    validation, _target_info, _duration = validator.validate(
+        load_migration_plan(str(plan_path)),
+        redis_url=redis_url,
+        query_check_file=str(query_check_path),
+    )
+    assert validation.schema_match is True
+    assert validation.doc_count_match is True
+    assert validation.key_sample_exists is True
+    assert not validation.errors
+
+    live_index.delete(drop=True)