test(fts): add integration tests for FTS control plane

Add comprehensive integration tests for the new Full-Text Search (FTS) control
plane functionality including:

- Schema-based index creation (dict format, field type classes, SchemaBuilder)
- Deployment configuration (explicit ServerlessDeployment, default deployment)
- describe_index compatibility (.spec.serverless.cloud, dimension, metric)
- FTS-only indexes (return None for vector properties)
- Error handling (spec/schema mutual exclusion, validation)
- Legacy spec-based creation compatibility

These tests verify the new schema-based API for creating Pinecone indexes
with full-text search and vector capabilities.

Relates to: SDK-113

Author: jhamon

tests/integration/rest_sync/db/control/fts/__init__.py

@@ -0,0 +1 @@
+# FTS control plane integration tests

tests/integration/rest_sync/db/control/fts/conftest.py

@@ -0,0 +1,54 @@
+"""Fixtures for FTS control plane integration tests."""
+
+import pytest
+import uuid
+import logging
+import dotenv
+from pinecone import Pinecone
+from tests.integration.helpers import delete_indexes_from_run, index_tags, get_environment_var
+
+dotenv.load_dotenv()
+
+logger = logging.getLogger(__name__)
+
+# Generate a unique ID for the entire test run
+RUN_ID = str(uuid.uuid4())
+
+
+@pytest.fixture()
+def pc():
+    """Create a Pinecone client."""
+    return Pinecone()
+
+
+@pytest.fixture()
+def index_name():
+    """Generate a unique index name for each test."""
+    return f"{str(uuid.uuid4())}"
+
+
+@pytest.fixture()
+def index_name_and_tags(request):
+    """Generate a unique index name and tags for each test."""
+    name = f"{str(uuid.uuid4())}"
+    tags = index_tags(request, RUN_ID)
+    return name, tags
+
+
+@pytest.fixture()
+def serverless_cloud():
+    """Get the serverless cloud provider."""
+    return get_environment_var("SERVERLESS_CLOUD", "aws")
+
+
+@pytest.fixture()
+def serverless_region():
+    """Get the serverless region."""
+    return get_environment_var("SERVERLESS_REGION", "us-east-1")
+
+
+def pytest_sessionfinish(session, exitstatus):
+    """Clean up indexes created during the test session."""
+    logger.info("Running final cleanup after FTS control plane tests...")
+    pc = Pinecone()
+    delete_indexes_from_run(pc, RUN_ID)

tests/integration/rest_sync/db/control/fts/test_create_index_deployment.py

@@ -0,0 +1,90 @@
+"""Integration tests for index deployment configurations.
+
+These tests verify the deployment parameter for schema-based index creation,
+including explicit ServerlessDeployment and default deployment behavior.
+"""
+
+from pinecone import Pinecone, DenseVectorField, ServerlessDeployment
+
+
+class TestCreateIndexWithDefaultDeployment:
+    """Test creating indexes with default deployment (aws/us-east-1)."""
+
+    def test_create_index_with_default_deployment(self, pc: Pinecone, index_name_and_tags):
+        """Test that omitting deployment defaults to aws/us-east-1."""
+        index_name, tags = index_name_and_tags
+        schema = {"embedding": DenseVectorField(dimension=512, metric="cosine")}
+        resp = pc.db.index.create(name=index_name, schema=schema, tags=tags)
+        assert resp.name == index_name
+
+        desc = pc.db.index.describe(name=index_name)
+        assert desc.name == index_name
+        assert desc.spec.serverless.cloud == "aws"
+        assert desc.spec.serverless.region == "us-east-1"
+
+
+class TestCreateIndexWithExplicitDeployment:
+    """Test creating indexes with explicit ServerlessDeployment."""
+
+    def test_create_index_with_serverless_deployment_aws(self, pc: Pinecone, index_name_and_tags):
+        """Test creating an index with explicit AWS ServerlessDeployment."""
+        index_name, tags = index_name_and_tags
+        schema = {"embedding": DenseVectorField(dimension=512, metric="cosine")}
+        resp = pc.db.index.create(
+            name=index_name,
+            schema=schema,
+            deployment=ServerlessDeployment(cloud="aws", region="us-east-1"),
+            tags=tags,
+        )
+        assert resp.name == index_name
+
+        desc = pc.db.index.describe(name=index_name)
+        assert desc.name == index_name
+        assert desc.spec.serverless.cloud == "aws"
+        assert desc.spec.serverless.region == "us-east-1"
+
+    def test_create_index_with_serverless_deployment_us_west(
+        self, pc: Pinecone, index_name_and_tags
+    ):
+        """Test creating an index with AWS us-west-2 ServerlessDeployment."""
+        index_name, tags = index_name_and_tags
+        schema = {"embedding": DenseVectorField(dimension=512, metric="cosine")}
+        resp = pc.db.index.create(
+            name=index_name,
+            schema=schema,
+            deployment=ServerlessDeployment(cloud="aws", region="us-west-2"),
+            tags=tags,
+        )
+        assert resp.name == index_name
+
+        desc = pc.db.index.describe(name=index_name)
+        assert desc.name == index_name
+        assert desc.spec.serverless.cloud == "aws"
+        assert desc.spec.serverless.region == "us-west-2"
+
+
+class TestDeploymentCompatibilityAccess:
+    """Test that deployment info is accessible via compatibility shim."""
+
+    def test_spec_serverless_access_pattern(self, pc: Pinecone, index_name_and_tags):
+        """Test that .spec.serverless.cloud access pattern works."""
+        index_name, tags = index_name_and_tags
+        schema = {"embedding": DenseVectorField(dimension=512, metric="cosine")}
+        pc.db.index.create(
+            name=index_name,
+            schema=schema,
+            deployment=ServerlessDeployment(cloud="aws", region="us-east-1"),
+            tags=tags,
+        )
+
+        desc = pc.db.index.describe(name=index_name)
+
+        # Verify the compatibility shim provides .spec.serverless access
+        assert desc.spec is not None
+        assert desc.spec.serverless is not None
+        assert desc.spec.serverless.cloud == "aws"
+        assert desc.spec.serverless.region == "us-east-1"
+
+        # Verify .spec.pod and .spec.byoc return None for serverless index
+        assert desc.spec.pod is None
+        assert desc.spec.byoc is None

tests/integration/rest_sync/db/control/fts/test_create_index_errors.py

@@ -0,0 +1,118 @@
+"""Integration tests for schema-based index creation error handling.
+
+These tests verify proper error handling for invalid configurations
+and mutual exclusion of spec and schema parameters.
+"""
+
+import pytest
+from pinecone import Pinecone, DenseVectorField, ServerlessSpec, CloudProvider, AwsRegion
+
+
+class TestSpecAndSchemaMutualExclusion:
+    """Test that spec and schema parameters are mutually exclusive."""
+
+    def test_error_when_both_spec_and_schema_provided(self, pc: Pinecone, index_name_and_tags):
+        """Test that providing both spec and schema raises ValueError."""
+        index_name, tags = index_name_and_tags
+        schema = {"embedding": DenseVectorField(dimension=512, metric="cosine")}
+        spec = ServerlessSpec(cloud=CloudProvider.AWS, region=AwsRegion.US_EAST_1)
+
+        with pytest.raises(ValueError) as exc_info:
+            pc.db.index.create(name=index_name, spec=spec, schema=schema, tags=tags)
+
+        assert "spec" in str(exc_info.value).lower()
+        assert "schema" in str(exc_info.value).lower()
+
+    def test_error_when_neither_spec_nor_schema_provided(self, pc: Pinecone, index_name_and_tags):
+        """Test that providing neither spec nor schema raises ValueError."""
+        index_name, tags = index_name_and_tags
+
+        with pytest.raises(ValueError) as exc_info:
+            pc.db.index.create(name=index_name, dimension=512, metric="cosine", tags=tags)
+
+        assert "spec" in str(exc_info.value).lower() or "schema" in str(exc_info.value).lower()
+
+
+class TestSchemaValidation:
+    """Test schema field configuration validation."""
+
+    def test_error_when_schema_is_empty(self, pc: Pinecone, index_name_and_tags):
+        """Test that an empty schema raises an error from the API."""
+        index_name, tags = index_name_and_tags
+        schema = {}
+
+        with pytest.raises(Exception):
+            pc.db.index.create(name=index_name, schema=schema, tags=tags)
+
+
+class TestLegacySpecBasedCreation:
+    """Test that legacy spec-based creation still works correctly."""
+
+    def test_legacy_serverless_spec_creation(self, pc: Pinecone, index_name_and_tags):
+        """Test that legacy ServerlessSpec-based creation still works."""
+        index_name, tags = index_name_and_tags
+        resp = pc.db.index.create(
+            name=index_name,
+            dimension=1024,
+            metric="cosine",
+            spec=ServerlessSpec(cloud=CloudProvider.AWS, region=AwsRegion.US_EAST_1),
+            tags=tags,
+        )
+        assert resp.name == index_name
+
+        desc = pc.db.index.describe(name=index_name)
+        assert desc.name == index_name
+        assert desc.dimension == 1024
+        assert desc.metric == "cosine"
+        assert desc.spec.serverless.cloud == "aws"
+        assert desc.spec.serverless.region == "us-east-1"
+
+    def test_legacy_dict_spec_creation(self, pc: Pinecone, index_name_and_tags):
+        """Test that legacy dict-based spec creation still works."""
+        index_name, tags = index_name_and_tags
+        resp = pc.db.index.create(
+            name=index_name,
+            dimension=768,
+            metric="euclidean",
+            spec={"serverless": {"cloud": "aws", "region": "us-east-1"}},
+            tags=tags,
+        )
+        assert resp.name == index_name
+
+        desc = pc.db.index.describe(name=index_name)
+        assert desc.name == index_name
+        assert desc.dimension == 768
+        assert desc.metric == "euclidean"
+
+    def test_legacy_vector_type_parameter(self, pc: Pinecone, index_name_and_tags):
+        """Test that legacy vector_type parameter works with spec."""
+        index_name, tags = index_name_and_tags
+        resp = pc.db.index.create(
+            name=index_name,
+            dimension=512,
+            metric="cosine",
+            vector_type="dense",
+            spec=ServerlessSpec(cloud=CloudProvider.AWS, region=AwsRegion.US_EAST_1),
+            tags=tags,
+        )
+        assert resp.name == index_name
+
+        desc = pc.db.index.describe(name=index_name)
+        assert desc.vector_type == "dense"
+
+    def test_legacy_sparse_index_creation(self, pc: Pinecone, index_name_and_tags):
+        """Test that legacy sparse index creation still works."""
+        index_name, tags = index_name_and_tags
+        resp = pc.db.index.create(
+            name=index_name,
+            metric="dotproduct",
+            vector_type="sparse",
+            spec=ServerlessSpec(cloud=CloudProvider.AWS, region=AwsRegion.US_EAST_1),
+            tags=tags,
+        )
+        assert resp.name == index_name
+
+        desc = pc.db.index.describe(name=index_name)
+        assert desc.metric == "dotproduct"
+        assert desc.vector_type == "sparse"
+        assert desc.dimension is None  # Sparse indexes have no dimension