diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperation.java b/src/main/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperation.java
index 7776d6ba74..7dd3341fba 100644
--- a/src/main/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperation.java
+++ b/src/main/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperation.java
@@ -28,7 +28,6 @@
 import io.stargate.sgv2.jsonapi.config.constants.TableCommentConstants;
 import io.stargate.sgv2.jsonapi.exception.DatabaseException;
 import io.stargate.sgv2.jsonapi.exception.SchemaException;
-import io.stargate.sgv2.jsonapi.service.cqldriver.CQLSessionCache;
 import io.stargate.sgv2.jsonapi.service.cqldriver.executor.QueryExecutor;
 import io.stargate.sgv2.jsonapi.service.cqldriver.override.ExtendedCreateIndex;
 import io.stargate.sgv2.jsonapi.service.cqldriver.override.ExtendedVectorType;
@@ -50,22 +49,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public record CreateCollectionOperation(
-    CommandContext<KeyspaceSchemaObject> commandContext,
-    DatabaseLimitsConfig dbLimitsConfig,
-    CQLSessionCache cqlSessionCache,
-    CqlIdentifier collectionName,
-    int ddlDelayMillis,
-    boolean tooManyIndexesRollbackEnabled,
-    // nullable
-    CreateCollectionCommand.Options.DocIdDesc docIdDesc,
-    // nullable
-    CreateCollectionCommand.Options.IndexingDesc indexingDesc,
-    // nullable
-    CreateCollectionCommand.Options.VectorSearchDesc vectorDesc,
-    SchemaHolder<CollectionLexicalDef> lexicalDef,
-    SchemaHolder<CollectionRerankDef> rerankDef)
-    implements Operation<CollectionSchemaObject> {
+/** Creates a new collection in the target keyspace using the SuperShredding table model */
+public class CreateCollectionOperation implements Operation<KeyspaceSchemaObject> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(CreateCollectionOperation.class);
 
@@ -73,6 +58,109 @@ public record CreateCollectionOperation(
 
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
+  private final CommandContext<KeyspaceSchemaObject> commandContext;
+  private final DatabaseLimitsConfig dbLimitsConfig;
+  private final CqlIdentifier collectionName;
+  private final int ddlDelayMillis;
+  private final boolean tooManyIndexesRollbackEnabled;
+  // nullable
+  private final CreateCollectionCommand.Options.DocIdDesc docIdDesc;
+  // nullable
+  private final CreateCollectionCommand.Options.IndexingDesc indexingDesc;
+  // nullable
+  private final CreateCollectionCommand.Options.VectorSearchDesc vectorDesc;
+  private final SchemaHolder<CollectionLexicalDef> lexicalDef;
+  private final SchemaHolder<CollectionRerankDef> rerankDef;
+
+  public CreateCollectionOperation(
+      CommandContext<KeyspaceSchemaObject> commandContext,
+      DatabaseLimitsConfig dbLimitsConfig,
+      CqlIdentifier collectionName,
+      int ddlDelayMillis,
+      boolean tooManyIndexesRollbackEnabled,
+      // nullable
+      CreateCollectionCommand.Options.DocIdDesc docIdDesc,
+      // nullable
+      CreateCollectionCommand.Options.IndexingDesc indexingDesc,
+      // nullable
+      CreateCollectionCommand.Options.VectorSearchDesc vectorDesc,
+      SchemaHolder<CollectionLexicalDef> lexicalDef,
+      SchemaHolder<CollectionRerankDef> rerankDef) {
+    this.commandContext = commandContext;
+    this.dbLimitsConfig = dbLimitsConfig;
+    this.collectionName = collectionName;
+    this.ddlDelayMillis = ddlDelayMillis;
+    this.tooManyIndexesRollbackEnabled = tooManyIndexesRollbackEnabled;
+    this.docIdDesc = docIdDesc;
+    this.indexingDesc = indexingDesc;
+    this.vectorDesc = vectorDesc;
+    this.lexicalDef = lexicalDef;
+    this.rerankDef = rerankDef;
+  }
+
+  /**
+   * Present and visible because the old testing relied on this being a record, keeping until we
+   * review the testing in more detail and determine if needed.
+   */
+  @VisibleForTesting
+  public CqlIdentifier collectionName() {
+    return collectionName;
+  }
+
+  /**
+   * Present and visible because the old testing relied on this being a record, keeping until we
+   * review the testing in more detail and determine if needed.
+   */
+  @VisibleForTesting
+  public CommandContext<KeyspaceSchemaObject> commandContext() {
+    return commandContext;
+  }
+
+  /**
+   * Present and visible because the old testing relied on this being a record, keeping until we
+   * review the testing in more detail and determine if needed.
+   */
+  @VisibleForTesting
+  public CreateCollectionCommand.Options.DocIdDesc docIdDesc() {
+    return docIdDesc;
+  }
+
+  /**
+   * Present and visible because the old testing relied on this being a record, keeping until we
+   * review the testing in more detail and determine if needed.
+   */
+  @VisibleForTesting
+  public CreateCollectionCommand.Options.IndexingDesc indexingDesc() {
+    return indexingDesc;
+  }
+
+  /**
+   * Present and visible because the old testing relied on this being a record, keeping until we
+   * review the testing in more detail and determine if needed.
+   */
+  @VisibleForTesting
+  public CreateCollectionCommand.Options.VectorSearchDesc vectorDesc() {
+    return vectorDesc;
+  }
+
+  /**
+   * Present and visible because the old testing relied on this being a record, keeping until we
+   * review the testing in more detail and determine if needed.
+   */
+  @VisibleForTesting
+  public SchemaHolder<CollectionLexicalDef> lexicalDef() {
+    return lexicalDef;
+  }
+
+  /**
+   * Present and visible because the old testing relied on this being a record, keeping until we
+   * review the testing in more detail and determine if needed.
+   */
+  @VisibleForTesting
+  public SchemaHolder<CollectionRerankDef> rerankDef() {
+    return rerankDef;
+  }
+
   @Override
   public Uni<Supplier<CommandResult>> execute(
       RequestContext requestContext, QueryExecutor queryExecutor) {
@@ -112,7 +200,7 @@ public Uni<Supplier<CommandResult>> execute(
                     requestContext,
                     queryExecutor,
                     initialTableComment,
-                    lexicalDef().runningValue(),
+                    lexicalDef.runningValue(),
                     false);
               }
 
@@ -158,11 +246,11 @@ public Uni<Supplier<CommandResult>> execute(
               // NOTE: FROM NOW ON WE NEED TO USE THE OVERRIDEN VALUE, (which may or may not be
               // actually overidden)
               var overrideLexicalDef =
-                  lexicalDef()
+                  lexicalDef
                       .replaceIfMissing(existingCollectionSettings.lexicalDefSchemaValue())
                       .value();
               var overrideRerankDef =
-                  rerankDef()
+                  rerankDef
                       .replaceIfMissing(existingCollectionSettings.rerankDefSchemaValue())
                       .value();
 
@@ -212,7 +300,7 @@ public Uni<Supplier<CommandResult>> execute(
 
   @VisibleForTesting
   String generateTableComment() {
-    return generateTableComment(lexicalDef(), rerankDef());
+    return generateTableComment(lexicalDef, rerankDef);
   }
 
   @VisibleForTesting
diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/operation/embeddings/FindEmbeddingProvidersOperation.java b/src/main/java/io/stargate/sgv2/jsonapi/service/operation/embeddings/FindEmbeddingProvidersOperation.java
index d4c54ac0b3..7df7bfaf13 100644
--- a/src/main/java/io/stargate/sgv2/jsonapi/service/operation/embeddings/FindEmbeddingProvidersOperation.java
+++ b/src/main/java/io/stargate/sgv2/jsonapi/service/operation/embeddings/FindEmbeddingProvidersOperation.java
@@ -10,6 +10,7 @@
 import io.stargate.sgv2.jsonapi.service.embedding.configuration.EmbeddingProvidersConfig;
 import io.stargate.sgv2.jsonapi.service.operation.Operation;
 import io.stargate.sgv2.jsonapi.service.provider.ApiModelSupport;
+import io.stargate.sgv2.jsonapi.service.schema.DatabaseSchemaObject;
 import java.util.*;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
@@ -20,7 +21,9 @@
  * CommandStatus#EXISTING_EMBEDDING_PROVIDERS} command status.
  */
 public record FindEmbeddingProvidersOperation(
-    FindEmbeddingProvidersCommand command, EmbeddingProvidersConfig config) implements Operation {
+    FindEmbeddingProvidersCommand command, EmbeddingProvidersConfig config)
+    implements Operation<DatabaseSchemaObject> {
+
   @Override
   public Uni<Supplier<CommandResult>> execute(
       RequestContext dataApiRequestInfo, QueryExecutor queryExecutor) {
diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolver.java b/src/main/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolver.java
index 4041163fa8..7666eda66d 100644
--- a/src/main/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolver.java
+++ b/src/main/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolver.java
@@ -101,7 +101,6 @@ public Operation resolveKeyspaceCommand(
     return new CreateCollectionOperation(
         context,
         dbLimitsConfig,
-        context.cqlSessionCache(),
         collectionName,
         operationsConfig.databaseConfig().ddlDelayMillis(),
         operationsConfig.tooManyIndexesRollbackEnabled(),
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/TestConstants.java b/src/test/java/io/stargate/sgv2/jsonapi/TestConstants.java
index e0c42122bc..73b687c3f3 100644
--- a/src/test/java/io/stargate/sgv2/jsonapi/TestConstants.java
+++ b/src/test/java/io/stargate/sgv2/jsonapi/TestConstants.java
@@ -5,6 +5,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.stargate.sgv2.jsonapi.api.model.command.CommandConfig;
 import io.stargate.sgv2.jsonapi.api.model.command.CommandContext;
@@ -56,6 +57,7 @@ public class TestConstants {
   public final String APP_NAME;
   public final DatabaseType DATABASE_TYPE;
   public final TenantFactory SINGLETON_TENANT_FACTORY;
+  public final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
   // ============================================================
   // Names
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/api/model/command/impl/IndexingDescTest.java b/src/test/java/io/stargate/sgv2/jsonapi/api/model/command/impl/IndexingDescTest.java
new file mode 100644
index 0000000000..25063d3750
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/api/model/command/impl/IndexingDescTest.java
@@ -0,0 +1,344 @@
+package io.stargate.sgv2.jsonapi.api.model.command.impl;
+
+import static io.stargate.sgv2.jsonapi.util.asserts.DataAPIAsserts.assertThatSchemaException;
+import static org.assertj.core.api.Assertions.*;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import io.stargate.sgv2.jsonapi.TestConstants;
+import io.stargate.sgv2.jsonapi.exception.ErrorCode;
+import io.stargate.sgv2.jsonapi.exception.SchemaException;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Unit tests for the {@link CreateCollectionCommand.Options.IndexingDesc}
+ *
+ * <p>We are testing that when created from JSON and validated, exceptions are thrown and functions
+ * derived from the record state work. Not really testing that deserialisationw works.
+ */
+public class IndexingDescTest {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(IndexingDescTest.class);
+
+  private final TestConstants TEST_CONSTANTS = new TestConstants();
+
+  private CreateCollectionCommand.Options.IndexingDesc assertValidate(
+      String testName, String json, CreateCollectionCommand.Options.IndexingDesc expected) {
+
+    LOGGER.info("assertValidate() - testName: {}, json:{}", testName, json);
+
+    var actualIndexingDesc = deserialize(testName, json);
+    actualIndexingDesc.validate();
+
+    assertThat(actualIndexingDesc).as(testName).isEqualTo(expected);
+    return actualIndexingDesc;
+  }
+
+  /** Assert that validating the JSON for the IndexingDesc will result in a SchemaException */
+  private void assertValidateFails(
+      String testName, String json, ErrorCode<?> errorCode, String... message) {
+
+    LOGGER.info("assertValidateFails() - testName: {}, json:{}", testName, json);
+
+    var indexingDesc = deserialize(testName, json);
+    var throwable = catchThrowable(indexingDesc::validate);
+
+    assertThatSchemaException(throwable)
+        .as(testName)
+        .hasCode(errorCode)
+        .hasMessageSnippets(message);
+  }
+
+  @Test
+  public void failAllowAndDeny() {
+
+    var json =
+        """
+        {
+            "deny": [
+                "comment"
+            ],
+            "allow": [
+                "data"
+            ]
+        }""";
+
+    assertValidateFails(
+        "failAllowAndDeny()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "'createCollection' indexing definition invalid: 'allow' and 'deny' cannot be used together");
+  }
+
+  @Test
+  public void failNeitherAllowNorDeny() {
+
+    var json =
+        """
+        {}""";
+
+    assertValidateFails(
+        "failNeitherAllowNorDeny()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "'allow' or 'deny' should be provided");
+  }
+
+  @Test
+  public void failAllowDuplicates() {
+
+    var json =
+        """
+        {
+            "allow": [
+                "name",
+                "name"
+            ]
+        }""";
+
+    assertValidateFails(
+        "failAllowDuplicates()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "'allow' cannot contain duplicates");
+  }
+
+  @Test
+  public void failDenyDuplicates() {
+
+    var json =
+        """
+         {
+             "deny": [
+                 "name",
+                 "name"
+             ]
+         }""";
+
+    assertValidateFails(
+        "failDenyDuplicates()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "'deny' cannot contain duplicates");
+  }
+
+  @Test
+  public void failAllowEmptyPath() {
+
+    var json =
+        """
+        {
+            "allow": [
+                ""
+            ]
+        }""";
+
+    assertValidateFails(
+        "failAllowEmptyPath()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "path must be represented as a non-empty string");
+  }
+
+  @Test
+  public void failDenyEmptyPath() {
+
+    var json =
+        """
+        {
+            "deny": [
+                ""
+            ]
+        }""";
+
+    assertValidateFails(
+        "failDenyEmptyPath()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "path must be represented as a non-empty string");
+  }
+
+  @Test
+  public void failAllowDollarPrefix() {
+
+    var json =
+        """
+        {
+            "allow": [
+                "$score"
+            ]
+        }""";
+
+    assertValidateFails(
+        "failAllowDollarPrefix()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "path ('$score') must not start with '$'");
+  }
+
+  @Test
+  public void failDenyDollarPrefix() {
+
+    var json =
+        """
+        {
+            "deny": [
+                "$score"
+            ]
+        }""";
+
+    assertValidateFails(
+        "failDenyDollarPrefix()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "path ('$score') must not start with '$'");
+  }
+
+  @Test
+  public void failAllowInvalidPath() {
+
+    var json =
+        """
+        {
+            "allow": [
+                "bad&path"
+            ]
+        }""";
+
+    assertValidateFails(
+        "failAllowInvalidPath()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "indexing path ('bad&path') is not a valid path");
+  }
+
+  @Test
+  public void failDenyInvalidPath() {
+
+    var json =
+        """
+        {
+            "deny": [
+                "bad&path"
+            ]
+        }""";
+
+    assertValidateFails(
+        "failDenyInvalidPath()",
+        json,
+        SchemaException.Code.INVALID_INDEXING_DEFINITION,
+        "indexing path ('bad&path') is not a valid path");
+  }
+
+  @Test
+  public void successAllowStar() {
+
+    var json =
+        """
+        {
+            "allow": [
+                "*"
+            ]
+        }""";
+
+    var expected = new CreateCollectionCommand.Options.IndexingDesc(List.of("*"), null);
+    assertValidate("successAllowStar()", json, expected);
+  }
+
+  @Test
+  public void successDenyStar() {
+
+    var json =
+        """
+        {
+            "deny": [
+                "*"
+            ]
+        }""";
+
+    var expected = new CreateCollectionCommand.Options.IndexingDesc(null, List.of("*"));
+    var actual = assertValidate("successDenyStar()", json, expected);
+    assertThat(actual.denyAll()).as("delayAll true when deny '*'").isTrue();
+  }
+
+  @Test
+  public void successAllowVector() {
+
+    var json =
+        """
+        {
+            "allow": [
+                "$vector"
+            ]
+        }""";
+
+    var expected = new CreateCollectionCommand.Options.IndexingDesc(List.of("$vector"), null);
+    assertValidate("successAllowVector()", json, expected);
+  }
+
+  @Test
+  public void successDenyVector() {
+
+    var json =
+        """
+        {
+            "deny": [
+                "$vector"
+            ]
+        }""";
+
+    var expected = new CreateCollectionCommand.Options.IndexingDesc(null, List.of("$vector"));
+    assertValidate("successDenyVector()", json, expected);
+  }
+
+  @Test
+  public void successAllowValidPaths() {
+
+    var json =
+        """
+        {
+            "allow": [
+                "name",
+                "address.city",
+                "tags"
+            ]
+        }""";
+
+    var expected =
+        new CreateCollectionCommand.Options.IndexingDesc(
+            List.of("name", "address.city", "tags"), null);
+    assertValidate("successAllowValidPaths()", json, expected);
+  }
+
+  @Test
+  public void successDenyValidPaths() {
+
+    var json =
+        """
+        {
+            "deny": [
+                "name",
+                "address.city",
+                "tags"
+            ]
+        }""";
+
+    var expected =
+        new CreateCollectionCommand.Options.IndexingDesc(
+            null, List.of("name", "address.city", "tags"));
+    assertValidate("successDenyValidPaths()", json, expected);
+  }
+
+  private CreateCollectionCommand.Options.IndexingDesc deserialize(String testName, String json) {
+    LOGGER.info("deserialize() - testName: {}, json: {}", testName, json);
+
+    try {
+      return TEST_CONSTANTS.OBJECT_MAPPER.readValue(
+          json, CreateCollectionCommand.Options.IndexingDesc.class);
+    } catch (JsonProcessingException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/api/request/tenant/SubdomainTenantResolverTests.java b/src/test/java/io/stargate/sgv2/jsonapi/api/request/tenant/SubdomainTenantResolverTests.java
index c68ccb7cb5..ea99b756ca 100644
--- a/src/test/java/io/stargate/sgv2/jsonapi/api/request/tenant/SubdomainTenantResolverTests.java
+++ b/src/test/java/io/stargate/sgv2/jsonapi/api/request/tenant/SubdomainTenantResolverTests.java
@@ -31,6 +31,8 @@ public class SubdomainTenantResolverTests {
 
   @BeforeEach
   public void initTenantFactory() {
+    // when run in the full suite, another test may leave this initialized
+    TenantFactory.reset();
     TenantFactory.initialize(DatabaseType.ASTRA);
   }
 
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperationTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperationTest.java
index 65853c29e7..5933c86304 100644
--- a/src/test/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperationTest.java
+++ b/src/test/java/io/stargate/sgv2/jsonapi/service/operation/collections/CreateCollectionOperationTest.java
@@ -1,28 +1,22 @@
 package io.stargate.sgv2.jsonapi.service.operation.collections;
 
+import static io.stargate.sgv2.jsonapi.util.asserts.DataAPIAsserts.assertThatCommandResult;
+import static io.stargate.sgv2.jsonapi.util.asserts.DataAPIAsserts.assertThatSchemaException;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.argThat;
-import static org.mockito.ArgumentMatchers.eq;
+import static org.assertj.core.api.Assertions.catchThrowable;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.cql.AsyncResultSet;
-import com.datastax.oss.driver.api.core.cql.ColumnDefinitions;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
-import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata;
 import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
 import com.datastax.oss.driver.api.core.servererrors.InvalidQueryException;
-import com.datastax.oss.driver.api.core.type.DataType;
-import com.datastax.oss.driver.internal.core.metadata.schema.DefaultColumnMetadata;
 import com.datastax.oss.driver.internal.core.metadata.schema.DefaultKeyspaceMetadata;
-import com.datastax.oss.driver.internal.core.type.DefaultTupleType;
-import com.datastax.oss.driver.internal.core.type.PrimitiveType;
-import com.datastax.oss.protocol.internal.ProtocolConstants;
 import com.fasterxml.jackson.core.JacksonException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -30,13 +24,15 @@
 import io.quarkus.test.junit.TestProfile;
 import io.smallrye.mutiny.Uni;
 import io.smallrye.mutiny.helpers.test.UniAssertSubscriber;
+import io.stargate.sgv2.jsonapi.api.model.command.CommandStatus;
 import io.stargate.sgv2.jsonapi.api.model.command.impl.CreateCollectionCommand;
+import io.stargate.sgv2.jsonapi.api.request.RequestContext;
 import io.stargate.sgv2.jsonapi.config.DatabaseLimitsConfig;
 import io.stargate.sgv2.jsonapi.config.constants.TableCommentConstants;
-import io.stargate.sgv2.jsonapi.service.cqldriver.CQLSessionCache;
+import io.stargate.sgv2.jsonapi.exception.APIException;
+import io.stargate.sgv2.jsonapi.exception.SchemaException;
 import io.stargate.sgv2.jsonapi.service.cqldriver.executor.QueryExecutor;
 import io.stargate.sgv2.jsonapi.service.cqldriver.executor.VectorColumnDefinition;
-import io.stargate.sgv2.jsonapi.service.cqldriver.executor.VectorConfig;
 import io.stargate.sgv2.jsonapi.service.resolver.CreateCollectionCommandResolver;
 import io.stargate.sgv2.jsonapi.service.schema.EmbeddingSourceModel;
 import io.stargate.sgv2.jsonapi.service.schema.SimilarityFunction;
@@ -47,16 +43,20 @@
 import io.stargate.sgv2.jsonapi.service.testutil.MockRow;
 import io.stargate.sgv2.jsonapi.testresource.NoGlobalResourcesTestProfile;
 import jakarta.inject.Inject;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * NOTE: Example table comment string:
+ * Tests for how the {@link CreateCollectionOperation} makes calls to the DB.
+ *
+ * <p>NOTE: Example table comment string:
  *
  * <pre>
  *  {
@@ -98,92 +98,130 @@ public class CreateCollectionOperationTest extends OperationTestBase {
 
   // Comment to extract comment from the crete table cql statement.
   // Assume it is delineated by single quotes
-  private static final Pattern COMMENT_PATTERN = Pattern.compile("comment='(.*?)'");
-
-  private final ColumnDefinitions RESULT_COLUMNS =
-      buildColumnDefs(OperationTestBase.TestColumn.ofBoolean("[applied]"));
-
-  private AsyncResultSet mockSuccessSchemaResultset() {
-    List<Row> resultRows =
-        Arrays.asList(new MockRow(RESULT_COLUMNS, 0, Arrays.asList(byteBufferFrom(true))));
-
-    return new MockAsyncResultSet(RESULT_COLUMNS, resultRows, null);
+  private static final Pattern TABLE_COMMENT_PATTERN = Pattern.compile("comment='(.*?)'");
+
+  private SchemaChangeMemento assertOperation(
+      String testName,
+      CreateCollectionOperation operation,
+      int expectedCreateTable,
+      int expectedCreateIndex) {
+    return assertOperation(
+        testName, operation, expectedCreateTable, expectedCreateIndex, true, false);
   }
 
-  private record SchemaChangeMemento(AtomicInteger counter, List<String> cqlComments) {
-    SchemaChangeMemento {
-      counter = counter == null ? new AtomicInteger() : counter;
-      cqlComments = cqlComments == null ? new ArrayList<>() : cqlComments;
+  /**
+   * Calls {@link CreateCollectionOperation#execute(RequestContext, QueryExecutor)} and makes
+   * assertions on how it works calling the DB. Main re-usable set of assertions.
+   *
+   * @param testName name of the test, used for logging
+   * @param operation Operation to test.
+   * @param expectedCreateTable How many CQL statements should be sent to the DB with CREATE TABLE
+   * @param expectedCreateIndex How many CQL statements should be sent t the DB with CREATE CUSTOM
+   *     INDEX
+   * @param mockKeyspaceMetadata If true, fake metadata for the keyspace is added to the mock. If
+   *     false, we will add mock data but it is empty.
+   * @param awaitFailure if true, we expect an throwable out of the operation and wait for that.
+   * @return {@link SchemaChangeMemento} which describes all the DB schema change calls the
+   *     operation made
+   */
+  private SchemaChangeMemento assertOperation(
+      String testName,
+      CreateCollectionOperation operation,
+      int expectedCreateTable,
+      int expectedCreateIndex,
+      boolean mockKeyspaceMetadata,
+      boolean awaitFailure) {
+
+    LOGGER.info(
+        "assertOperation() - testName={}, expectedCreateTable={}, expectedCreateIndex={}",
+        testName,
+        expectedCreateTable,
+        expectedCreateIndex);
+
+    // track calls to change the schema and mock the keyspace exists for the new collection
+    var mockQueryExecutor = mock(QueryExecutor.class);
+    var schemaChangeMemento = addSchemaChangeMomento(mockQueryExecutor);
+    addMockKeyspaceMetadata(mockQueryExecutor, mockKeyspaceMetadata);
+
+    // IMPORTANT - we still need to confirm the DB calls even in an error condition
+    // so catch, check db activity, then rethrow
+    var subscriber =
+        awaitFailure
+            ? operation
+                .execute(requestContext, mockQueryExecutor)
+                .subscribe()
+                .withSubscriber(UniAssertSubscriber.create())
+                .awaitFailure()
+            : operation
+                .execute(requestContext, mockQueryExecutor)
+                .subscribe()
+                .withSubscriber(UniAssertSubscriber.create())
+                .awaitItem();
+
+    LOGGER.info(
+        "assertOperation() - testName={}, schemaChangeMemento={}", testName, schemaChangeMemento);
+
+    // Validate the change schema calls
+    assertThat(schemaChangeMemento.counter.get())
+        .as("%s - total schema calls is expected")
+        .isEqualTo(expectedCreateTable + expectedCreateIndex);
+
+    int actualCreateTables =
+        schemaChangeMemento.queries.stream()
+            .filter(query -> query.startsWith("CREATE TABLE"))
+            .toList()
+            .size();
+    assertThat(actualCreateTables)
+        .as("%s - expected create table calls", testName)
+        .isEqualTo(expectedCreateTable);
+
+    // create table is always first
+    if (expectedCreateTable > 0) {
+      var collectionComment = schemaChangeMemento.tableComments.getFirst();
+      assertThat(collectionComment)
+          .as("%s - collection comment is not blank", testName)
+          .isNotBlank()
+          .as("Collection comment is not blank");
     }
-  }
-
-  private SchemaChangeMemento addSchemaChangeMomento(QueryExecutor queryExecutor) {
-    var memento = new SchemaChangeMemento(null, null);
-
-    when(queryExecutor.executeCreateSchemaChange(eq(requestContext), any()))
-        .then(
-            invocation -> {
-              memento.counter.incrementAndGet();
-              SimpleStatement statement = invocation.getArgument(1);
-              var matcher = COMMENT_PATTERN.matcher(statement.getQuery());
-              memento.cqlComments.add(matcher.find() ? matcher.group(1) : null);
-              return Uni.createFrom().item(mockSuccessSchemaResultset());
-            });
-    return memento;
-  }
 
-  private void addKeyspaceSchema(QueryExecutor queryExecutor) {
-
-    var driverMetadata = mock(Metadata.class);
-    when(queryExecutor.getDriverMetadata(any())).thenReturn(Uni.createFrom().item(driverMetadata));
-
-    var allKeyspaces = new HashMap<CqlIdentifier, KeyspaceMetadata>();
-    var keyspaceMetadata =
-        new DefaultKeyspaceMetadata(
-            TEST_CONSTANTS.KEYSPACE_IDENTIFIER.keyspace(),
-            false,
-            false,
-            new HashMap<>(),
-            new HashMap<>(),
-            new HashMap<>(),
-            new HashMap<>(),
-            new HashMap<>(),
-            new HashMap<>());
-    allKeyspaces.put(keyspaceMetadata.getName(), keyspaceMetadata);
-    when(driverMetadata.getKeyspaces()).thenReturn(allKeyspaces);
-  }
+    int actualCreateIndexes =
+        schemaChangeMemento.queries.stream()
+            .filter(query -> query.startsWith("CREATE CUSTOM INDEX"))
+            .toList()
+            .size();
+    assertThat(actualCreateIndexes)
+        .as("%s - expected create index calls", testName)
+        .isEqualTo(expectedCreateIndex);
+
+    // if the call failed, then we re-throw
+    if (awaitFailure) {
+      var throwable = subscriber.getFailure();
+      LOGGER.info("assertOperation() - testName={}, throwable={}", testName, throwable.toString());
+
+      assertThat(throwable)
+          .as("%s - expected failure is APIException", testName)
+          .isInstanceOf(APIException.class);
+      throw (APIException) throwable;
+    }
 
-  private JsonNode collectionNodeFromTableComment(String testName, String tableComment) {
+    // no failure, so check the command result is success
+    var commandResult = subscriber.getItem().get();
+    LOGGER.info("assertOperation() - testName={}, commandResult={}", testName, commandResult);
+    assertThatCommandResult(commandResult)
+        .as(testName)
+        .isDDLSuccess()
+        .hasOnlyStatus(CommandStatus.OK, 1);
 
-    LOGGER.info("tableCommentToNode() - testName: {}, tableComment: {}", testName, tableComment);
-    try {
-      var root = objectMapper.readTree(tableComment);
-      // we always want the "collection" node, see example at the top
-      // let the null out, it will cause the calling test to fail loud
-      return root.get(TableCommentConstants.TOP_LEVEL_KEY);
-    } catch (JacksonException e) {
-      throw new RuntimeException(
-          "Invalid JSON in Table comment for Collection, problem: " + e.getMessage());
-    }
+    return schemaChangeMemento;
   }
 
-  @BeforeEach
-  public void init() {}
-
   @Test
-  public void createCollectionNoVector() {
+  public void successWithLexicalNoVector() {
 
-    var queryExecutor = mock(QueryExecutor.class);
-    var schemaChangeMemento = addSchemaChangeMomento(queryExecutor);
-    addKeyspaceSchema(queryExecutor);
-
-    // aaron - 19-nov-2025 - best I can tell the sessionCache is not used but we need to pass it
-    // :(
     var operation =
         new CreateCollectionOperation(
             KEYSPACE_CONTEXT,
             databaseLimitsConfig,
-            mock(CQLSessionCache.class),
             TEST_CONSTANTS.COLLECTION_IDENTIFIER.table(),
             10,
             false,
@@ -193,19 +231,11 @@ public void createCollectionNoVector() {
             CollectionLexicalDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null),
             CollectionRerankDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null));
 
-    operation
-        .execute(requestContext, queryExecutor)
-        .subscribe()
-        .withSubscriber(UniAssertSubscriber.create())
-        .awaitItem();
-
     // 1 create Table + 8 super shredder indexes + lexical index
-    assertThat(schemaChangeMemento.counter.get()).isEqualTo(10);
+    var schemaChangeMemento = assertOperation("successWithLexicalNoVector", operation, 1, 9);
 
-    var collectionComment = schemaChangeMemento.cqlComments.getFirst();
-    assertThat(collectionComment).isNotBlank().as("Collection comment is not blank");
-
-    var commentNode = collectionNodeFromTableComment("createCollectionNoVector", collectionComment);
+    var commentNode =
+        collectionNodeFromTableComment("successWithLexicalNoVector", schemaChangeMemento);
     var optionsNode = commentNode.get(TableCommentConstants.OPTIONS_KEY);
     assertThat(optionsNode.get(TableCommentConstants.COLLECTION_VECTOR_KEY))
         .as("Collection comment must not have a vector key")
@@ -213,14 +243,7 @@ public void createCollectionNoVector() {
   }
 
   @Test
-  public void createCollectionVector() {
-
-    var queryExecutor = mock(QueryExecutor.class);
-    var schemaChangeMemento = addSchemaChangeMomento(queryExecutor);
-    addKeyspaceSchema(queryExecutor);
-
-    // aaron - 19-nov-2025 - best I can tell the sessionCache is not used but we need to pass it
-    // :(
+  public void successWithLexicalWithVector() {
 
     var vectorDesc = new CreateCollectionCommand.Options.VectorSearchDesc(5, "cosine", null, null);
     // Must use validateVectorOptions() because it will cleanup defaults, the resolver normally does
@@ -231,7 +254,6 @@ public void createCollectionVector() {
         new CreateCollectionOperation(
             KEYSPACE_CONTEXT,
             databaseLimitsConfig,
-            mock(CQLSessionCache.class),
             TEST_CONSTANTS.COLLECTION_IDENTIFIER.table(),
             10,
             false,
@@ -241,26 +263,18 @@ public void createCollectionVector() {
             CollectionLexicalDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null),
             CollectionRerankDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null));
 
-    operation
-        .execute(requestContext, queryExecutor)
-        .subscribe()
-        .withSubscriber(UniAssertSubscriber.create())
-        .awaitItem();
-
     // 1 create Table + 8 super shredder indexes + 1 vector index + 1 lexical
-    assertThat(schemaChangeMemento.counter.get()).isEqualTo(11);
+    var schemaChangeMemento = assertOperation("successWithLexicalWithVector", operation, 1, 10);
 
-    var collectionComment = schemaChangeMemento.cqlComments.getFirst();
-    assertThat(collectionComment).isNotBlank().as("Collection comment is not blank");
-
-    var commentNode = collectionNodeFromTableComment("createCollectionVector", collectionComment);
+    var commentNode =
+        collectionNodeFromTableComment("successWithLexicalWithVector", schemaChangeMemento);
     var optionsNode = commentNode.get(TableCommentConstants.OPTIONS_KEY);
+
     var vectorNode = optionsNode.get(TableCommentConstants.COLLECTION_VECTOR_KEY);
     assertThat(vectorNode).as("Collection comment must have a vector key").isNotNull();
 
     // see CollectionSettingsV1Reader
     var vectorColumnDefinition = VectorColumnDefinition.fromJson(vectorNode, objectMapper);
-    var vectorConfig = VectorConfig.fromColumnDefinitions(List.of(vectorColumnDefinition));
 
     assertThat(vectorColumnDefinition.vectorSize())
         .as("Vector size from table comment matches")
@@ -277,20 +291,13 @@ public void createCollectionVector() {
   }
 
   @Test
-  public void denyAllCollectionNoVector() {
-    var queryExecutor = mock(QueryExecutor.class);
-    var schemaChangeMemento = addSchemaChangeMomento(queryExecutor);
-    addKeyspaceSchema(queryExecutor);
-
-    // aaron - 19-nov-2025 - best I can tell the sessionCache is not used but we need to pass it
-    // :(
+  public void successIndexingDenyAllWithLexical() {
 
     var indexingDesc = new CreateCollectionCommand.Options.IndexingDesc(null, List.of("*"));
     var operation =
         new CreateCollectionOperation(
             KEYSPACE_CONTEXT,
             databaseLimitsConfig,
-            mock(CQLSessionCache.class),
             TEST_CONSTANTS.COLLECTION_IDENTIFIER.table(),
             10,
             false,
@@ -300,21 +307,15 @@ public void denyAllCollectionNoVector() {
             CollectionLexicalDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null),
             CollectionRerankDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null));
 
-    operation
-        .execute(requestContext, queryExecutor)
-        .subscribe()
-        .withSubscriber(UniAssertSubscriber.create())
-        .awaitItem();
-
     // 1 create Table + 1 lexical index
-    assertThat(schemaChangeMemento.counter.get()).isEqualTo(2);
-
-    var collectionComment = schemaChangeMemento.cqlComments.getFirst();
-    assertThat(collectionComment).isNotBlank().as("Collection comment is not blank");
+    // NOTE: because of deny all we do not need any super shredding, but we do need the lexcial
+    // still
+    // for the $lexcial field
+    var schemaChangeMemento = assertOperation("successIndexingDenyAllWithLexical", operation, 1, 1);
 
     // see CollectionSettingsV1Reader
     var commentNode =
-        collectionNodeFromTableComment("denyAllCollectionNoVector", collectionComment);
+        collectionNodeFromTableComment("successIndexingDenyAllWithLexical", schemaChangeMemento);
     var optionsNode = commentNode.get(TableCommentConstants.OPTIONS_KEY);
 
     var indexingNode = optionsNode.get(TableCommentConstants.COLLECTION_INDEXING_KEY);
@@ -331,11 +332,7 @@ public void denyAllCollectionNoVector() {
   }
 
   @Test
-  public void denyAllCollectionVector() {
-
-    var queryExecutor = mock(QueryExecutor.class);
-    var schemaChangeMemento = addSchemaChangeMomento(queryExecutor);
-    addKeyspaceSchema(queryExecutor);
+  public void successIndexingDenyAllWithLexicalWithVector() {
 
     var vectorDesc = new CreateCollectionCommand.Options.VectorSearchDesc(5, "cosine", null, null);
     // Must use validateVectorOptions() because it will cleanup defaults, the resolver normally does
@@ -347,7 +344,6 @@ public void denyAllCollectionVector() {
         new CreateCollectionOperation(
             KEYSPACE_CONTEXT,
             databaseLimitsConfig,
-            mock(CQLSessionCache.class),
             TEST_CONSTANTS.COLLECTION_IDENTIFIER.table(),
             10,
             false,
@@ -357,28 +353,59 @@ public void denyAllCollectionVector() {
             CollectionLexicalDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null),
             CollectionRerankDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null));
 
-    operation
-        .execute(requestContext, queryExecutor)
-        .subscribe()
-        .withSubscriber(UniAssertSubscriber.create())
-        .awaitItem();
-
     // 1 create Table + 1 vector index + 1 lexical
-    assertThat(schemaChangeMemento.counter.get()).isEqualTo(3);
+    // NOTE: because of deny all we do not need any super shredding, but we do need the lexcial
+    // still
+    // for the $lexcial field
+    var schemaChangeMemento =
+        assertOperation("successIndexingDenyAllWithLexicalWithVector", operation, 1, 2);
 
     // NOTE: no need to test the table comment again, that is covered above
   }
 
   @Test
-  public void indexAlreadyDropTable() {
+  public void failMissingKeyspace() {
+
+    var operation =
+        new CreateCollectionOperation(
+            KEYSPACE_CONTEXT,
+            databaseLimitsConfig,
+            TEST_CONSTANTS.COLLECTION_IDENTIFIER.table(),
+            10,
+            false,
+            null,
+            null,
+            null,
+            CollectionLexicalDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null),
+            CollectionRerankDefSchemaFactory.FOR_TESTING_ENABLED.currentVersion(null));
+
+    // this should throw, not return Command
+    var exception =
+        catchThrowable(() -> assertOperation("failMissingKeyspace", operation, 0, 0, false, true));
+
+    assertThatSchemaException(exception)
+        .as("failMissingKeyspace()")
+        .hasCode(SchemaException.Code.UNKNOWN_KEYSPACE)
+        .hasMessageSnippets(
+            "The keyspace used by the command: %s."
+                .formatted(KEYSPACE_CONTEXT.schemaObject().identifier().keyspace()));
+  }
+
+  /**
+   * Test: create table works, but there is an index with the same name, the operation should then
+   * try to drop the table. More complicated than others.
+   */
+  @Test
+  public void failExistingIndexDropTable() {
 
     var queryExecutor = mock(QueryExecutor.class);
-    var successResultSet = mockSuccessSchemaResultset();
-    addKeyspaceSchema(queryExecutor);
+    var successResultSet = mockSchemaSuccessResultSet();
+    addMockKeyspaceMetadata(queryExecutor);
 
-    final AtomicInteger schemaChangeCounter = new AtomicInteger();
-    final AtomicInteger schemaDropCounter = new AtomicInteger();
+    var schemaChangeCounter = new AtomicInteger();
+    var schemaDropCounter = new AtomicInteger();
 
+    // count the first create table
     when(queryExecutor.executeCreateSchemaChange(
             eq(requestContext),
             argThat(
@@ -390,6 +417,7 @@ public void indexAlreadyDropTable() {
               return Uni.createFrom().item(successResultSet);
             });
 
+    // mock existing index from first create index
     when(queryExecutor.executeCreateSchemaChange(
             eq(requestContext),
             argThat(
@@ -400,6 +428,7 @@ public void indexAlreadyDropTable() {
               throw new InvalidQueryException(mock(Node.class), "Index xxxxx already exists");
             });
 
+    // count the drop table
     when(queryExecutor.executeDropSchemaChange(
             eq(requestContext),
             argThat(
@@ -410,13 +439,10 @@ public void indexAlreadyDropTable() {
               return Uni.createFrom().item(successResultSet);
             });
 
-    // aaron - 19-nov-2025 - best I can tell the sessionCache is not used but we need to pass it
-    // :(
     var operation =
         new CreateCollectionOperation(
             KEYSPACE_CONTEXT,
             databaseLimitsConfig,
-            mock(CQLSessionCache.class),
             TEST_CONSTANTS.COLLECTION_IDENTIFIER.table(),
             10,
             true,
@@ -431,25 +457,98 @@ public void indexAlreadyDropTable() {
         .subscribe()
         .withSubscriber(UniAssertSubscriber.create())
         .awaitItem();
+
     // 1 create Table + 1 index failure
     assertThat(schemaChangeCounter.get()).isEqualTo(2);
     // 1 drop table
     assertThat(schemaDropCounter.get()).isEqualTo(1);
   }
 
-  private List<ColumnMetadata> createCorrectPartitionColumn() {
-    List<DataType> tuple =
-        Arrays.asList(
-            new PrimitiveType(ProtocolConstants.DataType.TINYINT),
-            new PrimitiveType(ProtocolConstants.DataType.VARCHAR));
-    List<ColumnMetadata> partitionKey = new ArrayList<>();
-    partitionKey.add(
-        new DefaultColumnMetadata(
-            CqlIdentifier.fromInternal("keyspace"),
-            CqlIdentifier.fromInternal("collection"),
-            CqlIdentifier.fromInternal("key"),
-            new DefaultTupleType(tuple),
-            false));
-    return partitionKey;
+  private void addMockKeyspaceMetadata(QueryExecutor queryExecutor) {
+    addMockKeyspaceMetadata(queryExecutor, true);
+  }
+
+  /** Attaches mock KeyspaceMetadata to the queryExecutor so create tests can find the keyspace */
+  private void addMockKeyspaceMetadata(QueryExecutor queryExecutor, boolean addKeyspaceMetadata) {
+
+    var allKeyspaces = new HashMap<CqlIdentifier, KeyspaceMetadata>();
+    if (addKeyspaceMetadata) {
+      var keyspaceMetadata =
+          new DefaultKeyspaceMetadata(
+              TEST_CONSTANTS.KEYSPACE_IDENTIFIER.keyspace(),
+              false,
+              false,
+              new HashMap<>(),
+              new HashMap<>(),
+              new HashMap<>(),
+              new HashMap<>(),
+              new HashMap<>(),
+              new HashMap<>());
+      allKeyspaces.put(keyspaceMetadata.getName(), keyspaceMetadata);
+    }
+
+    var driverMetadata = mock(Metadata.class);
+    when(driverMetadata.getKeyspaces()).thenReturn(allKeyspaces);
+
+    when(queryExecutor.getDriverMetadata(any())).thenReturn(Uni.createFrom().item(driverMetadata));
+  }
+
+  private AsyncResultSet mockSchemaSuccessResultSet() {
+
+    var schemaColumns = buildColumnDefs(OperationTestBase.TestColumn.ofBoolean("[applied]"));
+    List<Row> resultRows = List.of(new MockRow(schemaColumns, 0, List.of(byteBufferFrom(true))));
+    return new MockAsyncResultSet(schemaColumns, resultRows, null);
+  }
+
+  /**
+   * Holds the calls the operation made against the DB
+   *
+   * @param counter Total number of calls that were intercepted.
+   * @param queries The CQL from the statements, in order the calls were intercepted.
+   * @param tableComments If the CQL had a command field (for create table) the extracted comment,
+   *     see TABLE_COMMENT_PATTERN
+   */
+  private record SchemaChangeMemento(
+      AtomicInteger counter, List<String> queries, List<String> tableComments) {
+    SchemaChangeMemento {
+      counter = counter == null ? new AtomicInteger() : counter;
+      queries = queries == null ? new ArrayList<>() : queries;
+      tableComments = tableComments == null ? new ArrayList<>() : tableComments;
+    }
+  }
+
+  private SchemaChangeMemento addSchemaChangeMomento(QueryExecutor queryExecutor) {
+    var memento = new SchemaChangeMemento(null, null, null);
+
+    when(queryExecutor.executeCreateSchemaChange(eq(requestContext), any()))
+        .then(
+            invocation -> {
+              memento.counter.incrementAndGet();
+              SimpleStatement statement = invocation.getArgument(1);
+              memento.queries.add(statement.getQuery());
+              var matcher = TABLE_COMMENT_PATTERN.matcher(statement.getQuery());
+              memento.tableComments.add(matcher.find() ? matcher.group(1) : null);
+              return Uni.createFrom().item(mockSchemaSuccessResultSet());
+            });
+    return memento;
+  }
+
+  private JsonNode collectionNodeFromTableComment(String testName, SchemaChangeMemento memento) {
+    // create table should be first
+    return collectionNodeFromTableComment(testName, memento.tableComments.getFirst());
+  }
+
+  private JsonNode collectionNodeFromTableComment(String testName, String tableComment) {
+
+    LOGGER.info("tableCommentToNode() - testName: {}, tableComment: {}", testName, tableComment);
+    try {
+      var root = objectMapper.readTree(tableComment);
+      // we always want the "collection" node, see example at the top
+      // let the null out, it will cause the calling test to fail loud
+      return root.get(TableCommentConstants.TOP_LEVEL_KEY);
+    } catch (JacksonException e) {
+      throw new RuntimeException(
+          "Invalid JSON in Table comment for Collection, problem: " + e.getMessage());
+    }
   }
 }
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CommandResolverTestBase.java b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CommandResolverTestBase.java
new file mode 100644
index 0000000000..a4af075bb6
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CommandResolverTestBase.java
@@ -0,0 +1,115 @@
+package io.stargate.sgv2.jsonapi.service.resolver;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import io.stargate.sgv2.jsonapi.TestConstants;
+import io.stargate.sgv2.jsonapi.api.model.command.Command;
+import io.stargate.sgv2.jsonapi.api.model.command.CommandContext;
+import io.stargate.sgv2.jsonapi.service.operation.Operation;
+import io.stargate.sgv2.jsonapi.service.operation.collections.CreateCollectionOperation;
+import io.stargate.sgv2.jsonapi.service.schema.SchemaObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base class for tests for a {@link CommandResolver} to reduce some boilerplate.
+ *
+ * <p>Notes for users:
+ *
+ * <ul>
+ *   <li>Godo exmaple of why this exists, see {@link #assertResolver(String, String)}
+ *   <li>We still need some CDI injection, so there are abstract properties to access objects
+ *       injected after construction on the subclass.
+ *   <li>Avoid adding actual @Test tests to this class, as they will be run for all subclasses. Add
+ *       features that do asserts etc, and then have the @Test test declared on subclasses
+ * </ul>
+ *
+ * @param <SCHEMA> the type of the schema under test, this is the type used by the operation created
+ * @param <COMMAND> the type of the command under test
+ * @param <RESOLVER> the type of the resolver we will pass the command for testing
+ * @param <OPERATION> the type of the operation that should be created by the resolver.
+ */
+abstract class CommandResolverTestBase<
+    SCHEMA extends SchemaObject,
+    COMMAND extends Command,
+    RESOLVER extends CommandResolver<COMMAND>,
+    OPERATION extends Operation<SCHEMA>> {
+
+  protected static final Logger LOGGER = LoggerFactory.getLogger(CommandResolverTestBase.class);
+
+  protected final TestConstants TEST_CONSTANTS = new TestConstants();
+
+  // =================================================================
+  // Required implementation for subclasses
+  // =================================================================
+
+  /**
+   * Implementations should return an instance of the resolver for test, may be a new instance every
+   * call or same instance.
+   */
+  protected abstract RESOLVER resolver();
+
+  /**
+   * Implementations should return the CommandContext to call the resolver with, this may be
+   * different for each call but recommend it is the same.
+   */
+  protected abstract CommandContext<SCHEMA> commandContext();
+
+  /** Class of {@link COMMAND}, the JSON of the command is deseralised into this class. */
+  protected abstract Class<COMMAND> commandClass();
+
+  // =================================================================
+  // Useful assertions for subclasses to use.
+  // =================================================================
+
+  /**
+   * Resolves the command described in the rawJson and does basic assertions on the operaton
+   * returned. Most subclasses testing a resolve need this.
+   *
+   * @param testName name for logging etc
+   * @param rawJson the raw command JSON to parse into a command, {@link
+   *     TestConstants#rawNamesSubstitutor()} is used to replace names such as <code>${collection}
+   *     </code>
+   * @return The operation the resolver created.
+   */
+  protected OPERATION assertResolver(String testName, String rawJson) {
+
+    var json = TEST_CONSTANTS.subsRawNames(rawJson);
+    LOGGER.info("assertResolver() - testName: {}, (substituted) json: {}", testName, json);
+
+    COMMAND command;
+    try {
+      command = TEST_CONSTANTS.OBJECT_MAPPER.readValue(json, commandClass());
+    } catch (JsonProcessingException e) {
+      throw new RuntimeException(e);
+    }
+
+    var localCommandContext = commandContext();
+    var operation = resolver().resolveCommand(localCommandContext, command);
+
+    assertThat(operation)
+        .isInstanceOfSatisfying(
+            CreateCollectionOperation.class,
+            op -> {
+              assertThat(op.commandContext())
+                  .as("%s - commandContext() is same object", testName)
+                  .isSameAs(localCommandContext);
+            });
+    return (OPERATION) operation;
+  }
+
+  /**
+   * Runs the command through the resolver, and assumes it will throw an exception.
+   *
+   * <p>See {@link #assertResolver(String, String)}
+   */
+  protected Throwable assertResolverThrows(String testName, String rawJson) {
+    var throwable = catchThrowable(() -> assertResolver(testName, rawJson));
+
+    assertThat(throwable).as("%s - exception is thrown", testName).isNotNull();
+
+    return throwable;
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverTest.java
index bf2fe808c3..0a06912319 100644
--- a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverTest.java
+++ b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverTest.java
@@ -1,429 +1,276 @@
 package io.stargate.sgv2.jsonapi.service.resolver;
 
 import static io.stargate.sgv2.jsonapi.util.CqlIdentifierUtil.cqlIdentifierFromUserInput;
+import static io.stargate.sgv2.jsonapi.util.asserts.DataAPIAsserts.assertThatSchemaException;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.catchThrowable;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.TestProfile;
-import io.stargate.sgv2.jsonapi.TestConstants;
-import io.stargate.sgv2.jsonapi.api.model.command.CommandContext;
 import io.stargate.sgv2.jsonapi.api.model.command.impl.CreateCollectionCommand;
-import io.stargate.sgv2.jsonapi.api.model.command.impl.VectorizeConfig;
 import io.stargate.sgv2.jsonapi.exception.SchemaException;
-import io.stargate.sgv2.jsonapi.service.operation.collections.CreateCollectionOperation;
 import io.stargate.sgv2.jsonapi.service.schema.EmbeddingSourceModel;
-import io.stargate.sgv2.jsonapi.service.schema.KeyspaceSchemaObject;
-import jakarta.inject.Inject;
+import io.stargate.sgv2.jsonapi.service.schema.SimilarityFunction;
+import io.stargate.sgv2.jsonapi.testresource.NoGlobalResourcesTestProfile;
 import java.util.List;
-import java.util.Map;
 import org.apache.commons.lang3.RandomStringUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
-
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests how the {@link CreateCollectionCommandResolver} handles inputs and the operation it
+ * creates.
+ *
+ * <p><b>NOTE:</b> subclassed atleast by {@link
+ * CreateCollectionCommandResolverVectorizeDisabledTest} to change the vectorize enabled setting
+ */
 @QuarkusTest
-@TestProfile(EnabledVectorizeProfile.class)
-class CreateCollectionCommandResolverTest {
-
-  @Inject ObjectMapper objectMapper;
-  @Inject CreateCollectionCommandResolver resolver;
-
-  private final TestConstants TEST_CONSTANTS = new TestConstants();
-
-  CommandContext<KeyspaceSchemaObject> commandContext;
-
-  @BeforeEach
-  public void beforeEach() {
-    commandContext = TEST_CONSTANTS.keyspaceContext();
+@TestProfile(NoGlobalResourcesTestProfile.Impl.class)
+class CreateCollectionCommandResolverTest extends CreateCollectionCommandResolverTestBase {
+
+  protected static final Logger LOGGER =
+      LoggerFactory.getLogger(CreateCollectionCommandResolverTest.class);
+
+  @Test
+  public void successWithDefaults() {
+    var operation =
+        assertResolver(
+            "successWithDefaults()",
+            """
+                        {
+                            "createCollection": {
+                                "name": "${collection}"
+                            }
+                        }
+                        """);
+
+    assertThat(operation.docIdDesc()).isNull();
+    assertThat(operation.indexingDesc()).isNull();
+    assertThat(operation.vectorDesc()).isNull();
+    assertThat(operation.indexingDesc()).isNull();
+    assertThat(operation.lexicalDef())
+        .isEqualTo(
+            COMMAND_CONTEXT.requestContext().schemaRegistry().lexicalDef().currentVersion(null));
+    assertThat(operation.rerankDef())
+        .isEqualTo(
+            COMMAND_CONTEXT.requestContext().schemaRegistry().rerankDef().currentVersion(null));
   }
 
-  @Nested
-  class CreateCollectionSuccess {
+  @Test
+  public void successWithSupportedNames() {
 
-    @Test
-    public void happyPath() throws Exception {
+    String[] supportedNames = {"a", "A", "0", "_", "a0", "0a_A", "_0a"};
+    for (String name : supportedNames) {
       var json =
-          TEST_CONSTANTS.subsRawNames(
               """
-              {
-                "createCollection": {
-                  "name" : "${collection}"
-                }
-              }
-              """);
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var operation = resolver.resolveCommand(commandContext, command);
-
-      assertThat(operation)
-          .isInstanceOfSatisfying(
-              CreateCollectionOperation.class,
-              op -> {
-                assertThat(op.collectionName())
-                    .isEqualTo(TEST_CONSTANTS.COLLECTION_IDENTIFIER.table());
-                assertThat(op.commandContext()).isEqualTo(commandContext);
-                assertThat(op.vectorDesc()).isNull();
-              });
-    }
-
-    @Test
-    public void happyPathVectorSearch() throws Exception {
-      var json =
-          TEST_CONSTANTS.subsRawNames(
-              """
-          {
-            "createCollection": {
-              "name" : "${collection}",
-              "options": {
-                "vector": {
-                  "dimension": 4,
-                  "metric": "cosine"
-                }
-              }
-            }
-          }
-          """);
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var operation = resolver.resolveCommand(commandContext, command);
-
-      assertThat(operation)
-          .isInstanceOfSatisfying(
-              CreateCollectionOperation.class,
-              op -> {
-                assertThat(op.collectionName())
-                    .isEqualTo(TEST_CONSTANTS.COLLECTION_IDENTIFIER.table());
-                assertThat(op.commandContext()).isEqualTo(commandContext);
-                assertThat(op.vectorDesc()).isNotNull();
-                assertThat(op.vectorDesc().dimension()).isEqualTo(4);
-                assertThat(op.vectorDesc().metric()).isEqualTo("cosine");
-              });
+                    {
+                        "createCollection": {
+                            "name": "%s"
+                        }
+                    }
+                    """
+              .formatted(name);
+      assertResolver("successWithSupportedNames()", json, cqlIdentifierFromUserInput(name));
     }
+  }
 
-    @Test
-    public void happyPathVectorizeSearch() throws Exception {
-      var json =
-          TEST_CONSTANTS.subsRawNames(
-              """
-            {
-                "createCollection": {
-                    "name": "${collection}",
-                    "options": {
-                        "vector": {
-                            "metric": "cosine",
-                            "dimension": 768,
-                            "service": {
-                                "provider": "azureOpenAI",
-                                "modelName": "text-embedding-3-small",
-                                "parameters": {
-                                    "resourceName": "test",
-                                    "deploymentId": "test"
+  @Test
+  public void successWithVector() {
+    var operation =
+        assertResolver(
+            "successWithVector()",
+            """
+                        {
+                            "createCollection": {
+                                "name": "${collection}",
+                                "options": {
+                                    "vector": {
+                                        "dimension": 4,
+                                        "metric": "cosine",
+                                        "sourceModel": "openai-v3-large"
+                                    }
                                 }
                             }
                         }
-                    }
-                }
-            }
-          """);
-      // NOTE: source model of null turns into DEFAULT
-      var expectedVectorDesc =
-          new CreateCollectionCommand.Options.VectorSearchDesc(
-              768,
-              "cosine",
-              EmbeddingSourceModel.DEFAULT.cqlName(),
-              new VectorizeConfig(
-                  "azureOpenAI",
-                  "text-embedding-3-small",
-                  null,
-                  Map.of("resourceName", "test", "deploymentId", "test")));
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var operation = resolver.resolveCommand(commandContext, command);
-
-      // NOTE: this used to check the table comment string that was created, that has moved to the
-      // CreateCollectionOperationTest
-      assertThat(operation)
-          .isInstanceOfSatisfying(
-              CreateCollectionOperation.class,
-              op -> {
-                assertThat(op.collectionName())
-                    .isEqualTo(TEST_CONSTANTS.COLLECTION_IDENTIFIER.table());
-                assertThat(op.commandContext()).isEqualTo(commandContext);
-                assertThat(op.vectorDesc()).isEqualTo(expectedVectorDesc);
-              });
-    }
+                        """);
 
-    @Test
-    public void happyPathIndexing() throws Exception {
+    var expectedVectorDesc =
+        new CreateCollectionCommand.Options.VectorSearchDesc(
+            4, "cosine", EmbeddingSourceModel.OPENAI_V3_LARGE.apiName(), null);
 
-      var json =
-          TEST_CONSTANTS.subsRawNames(
-              """
-          {
-            "createCollection": {
-              "name" : "${collection}",
-              "options": {
-                "indexing": {
-                  "deny" : ["comment"]
-                }
-              }
-            }
-          }
-          """);
-      var expectedIndexing =
-          new CreateCollectionCommand.Options.IndexingDesc(null, List.of("comment"));
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var operation = resolver.resolveCommand(commandContext, command);
-
-      assertThat(operation)
-          .isInstanceOfSatisfying(
-              CreateCollectionOperation.class,
-              op -> {
-                assertThat(op.collectionName())
-                    .isEqualTo(TEST_CONSTANTS.COLLECTION_IDENTIFIER.table());
-                assertThat(op.commandContext()).isEqualTo(commandContext);
-                assertThat(op.indexingDesc()).isEqualTo(expectedIndexing);
-              });
-    }
-
-    @Test
-    public void happyPathVectorSearchDefaultFunction() throws Exception {
-
-      var json =
-          TEST_CONSTANTS.subsRawNames(
-              """
-          {
-            "createCollection": {
-              "name" : "${collection}",
-              "options": {
-                "vector": {
-                  "dimension": 4
-                }
-              }
-            }
-          }
-          """);
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var operation = resolver.resolveCommand(commandContext, command);
-
-      assertThat(operation)
-          .isInstanceOfSatisfying(
-              CreateCollectionOperation.class,
-              op -> {
-                assertThat(op.collectionName())
-                    .isEqualTo(TEST_CONSTANTS.COLLECTION_IDENTIFIER.table());
-                assertThat(op.commandContext()).isEqualTo(commandContext);
-                assertThat(op.vectorDesc()).isNotNull();
-                assertThat(op.vectorDesc().dimension()).isEqualTo(4);
-                assertThat(op.vectorDesc().metric()).isEqualTo("COSINE");
-              });
-    }
-
-    @Test
-    public void createCollectionWithSupportedName() throws Exception {
-
-      String[] supportedName = {"a", "A", "0", "_", "a0", "0a_A", "_0a"};
-      for (String name : supportedName) {
-        String json =
-                """
-              {
-                "createCollection": {
-                  "name" : "%s"
-                }
-              }
-              """
-                .formatted(name);
-
-        var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-        var operation = resolver.resolveCommand(commandContext, command);
-
-        assertThat(operation)
-            .isInstanceOfSatisfying(
-                CreateCollectionOperation.class,
-                op -> {
-                  assertThat(op.collectionName()).isEqualTo(cqlIdentifierFromUserInput(name));
-                  assertThat(op.commandContext()).isEqualTo(commandContext);
-                  assertThat(op.vectorDesc()).isNull();
-                });
-      }
-    }
+    assertThat(operation.vectorDesc()).isEqualTo(expectedVectorDesc);
   }
 
-  @Nested
-  class CreateCollectionFailure {
-
-    @Test
-    public void indexingOptionsError() throws Exception {
-
-      var json =
-          TEST_CONSTANTS.subsRawNames(
-              """
-                  {
-                    "createCollection": {
-                      "name" : "${collection}",
-                      "options": {
-                        "vector": {
-                          "dimension": 4,
-                          "metric": "cosine"
-                        },
-                        "indexing": {
-                          "deny" : ["comment"],
-                          "allow" : ["data"]
+  @Test
+  public void successWithVectorDefaultMetric() {
+
+    var operation =
+        assertResolver(
+            "successWithVectorDefaultMetric()",
+            """
+                        {
+                            "createCollection": {
+                                "name": "${collection}",
+                                "options": {
+                                    "vector": {
+                                        "dimension": 4
+                                    }
+                                }
+                            }
                         }
-                      }
-                    }
-                  }
-                  """);
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var throwable = catchThrowable(() -> resolver.resolveCommand(commandContext, command));
-
-      assertThat(throwable)
-          .isInstanceOf(SchemaException.class)
-          .satisfies(
-              e -> {
-                SchemaException exception = (SchemaException) e;
-                assertThat(exception.getMessage())
-                    .containsSequence(
-                        "'createCollection' indexing definition invalid: 'allow' and 'deny' cannot be used together");
-                assertThat(exception.code)
-                    .isEqualTo(SchemaException.Code.INVALID_INDEXING_DEFINITION.name());
-              });
-    }
-
-    @Test
-    public void createCollectionWithNull() throws Exception {
-
-      var json =
-          """
-          {
-            "createCollection": {
-            }
-          }
-          """;
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var throwable = catchThrowable(() -> resolver.resolveCommand(commandContext, command));
-
-      verifySchemaException(
-          throwable,
-          SchemaException.Code.UNSUPPORTED_SCHEMA_NAME,
-          "The command attempted to create a Collection with a name that is not supported.",
-          "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
-          "The command used the unsupported Collection name: '(null)'.");
-    }
+                        """);
+
+    // aaron 15 june 26 - not sure why but we need the name of the source model must be the CQL name
+    // cql cares about capitals, we dont when processing this normally
+    var expectedVectorDesc =
+        new CreateCollectionCommand.Options.VectorSearchDesc(
+            4,
+            SimilarityFunction.COSINE.cqlIndexingFunction(),
+            EmbeddingSourceModel.OTHER.cqlName(),
+            null);
+
+    assertThat(operation.vectorDesc()).isEqualTo(expectedVectorDesc);
+  }
 
-    @Test
-    public void createCollectionWithEmptyName() throws Exception {
+  @Test
+  public void successWithDenyIndexing() {
+    var operation =
+        assertResolver(
+            "successWithDenyIndexing()",
+            """
+                        {
+                            "createCollection": {
+                                "name": "${collection}",
+                                "options": {
+                                    "indexing": {
+                                        "deny": [
+                                            "comment"
+                                        ]
+                                    }
+                                }
+                            }
+                        }
+                        """);
 
-      var json =
-          """
-          {
-            "createCollection": {
-              "name" : ""
-            }
-          }
-          """;
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var throwable = catchThrowable(() -> resolver.resolveCommand(commandContext, command));
-
-      verifySchemaException(
-          throwable,
-          SchemaException.Code.UNSUPPORTED_SCHEMA_NAME,
-          "The command attempted to create a Collection with a name that is not supported.",
-          "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
-          "The command used the unsupported Collection name: ''.");
-    }
+    var expectedIndexing =
+        new CreateCollectionCommand.Options.IndexingDesc(null, List.of("comment"));
 
-    @Test
-    public void createCollectionWithBlankName() throws Exception {
+    assertThat(operation.indexingDesc()).isEqualTo(expectedIndexing);
+  }
 
-      var json =
-          """
-          {
-            "createCollection": {
-              "name": " "
-            }
-          }
-          """;
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var throwable = catchThrowable(() -> resolver.resolveCommand(commandContext, command));
-
-      verifySchemaException(
-          throwable,
-          SchemaException.Code.UNSUPPORTED_SCHEMA_NAME,
-          "The command attempted to create a Collection with a name that is not supported.",
-          "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
-          "The command used the unsupported Collection name: ' '.");
-    }
+  @Test
+  public void failWithUndefinedName() {
 
-    @Test
-    public void createCollectionWithNameTooLong() throws Exception {
+    var throwable =
+        assertResolverThrows(
+            "failWithUndefinedName()",
+            """
+                        {
+                            "createCollection": {}
+                        }
+                        """);
+
+    assertThatSchemaException(throwable)
+        .as("failWithUndefinedName()")
+        .hasCode(SchemaException.Code.UNSUPPORTED_SCHEMA_NAME)
+        .hasMessageSnippets(
+            "The command attempted to create a Collection with a name that is not supported.",
+            "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
+            "The command used the unsupported Collection name: '(null)'.");
+  }
 
-      var name = RandomStringUtils.insecure().nextAlphabetic(49);
-      var json =
-              """
-          {
-            "createCollection": {
-              "name": "%s"
-            }
-          }
-          """
-              .formatted(name);
+  @Test
+  public void failWithEmptyName() {
 
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var throwable = catchThrowable(() -> resolver.resolveCommand(commandContext, command));
+    var throwable =
+        assertResolverThrows(
+            "failWithEmptyName()",
+            """
+                        {
+                            "createCollection": {
+                                "name": ""
+                            }
+                        }
+                        """);
+
+    assertThatSchemaException(throwable)
+        .as("failWithEmptyName()")
+        .hasCode(SchemaException.Code.UNSUPPORTED_SCHEMA_NAME)
+        .hasMessageSnippets(
+            "The command attempted to create a Collection with a name that is not supported.",
+            "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
+            "The command used the unsupported Collection name: ''.");
+  }
 
-      verifySchemaException(
-          throwable,
-          SchemaException.Code.UNSUPPORTED_SCHEMA_NAME,
-          "The command attempted to create a Collection with a name that is not supported.",
-          "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
-          "The command used the unsupported Collection name: '%s'.".formatted(name));
-    }
+  @Test
+  public void failWithBlankName() {
+
+    // Blank is a white space
+    var throwable =
+        assertResolverThrows(
+            "failWithBlankName()",
+            """
+                        {
+                            "createCollection": {
+                                "name": "  "
+                            }
+                        }
+                        """);
+
+    assertThatSchemaException(throwable)
+        .as("failWithBlankName()")
+        .hasCode(SchemaException.Code.UNSUPPORTED_SCHEMA_NAME)
+        .hasMessageSnippets(
+            "The command attempted to create a Collection with a name that is not supported.",
+            "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
+            "The command used the unsupported Collection name: '  '.");
+  }
 
-    @Test
-    public void createCollectionWithSpecialCharacter() throws Exception {
+  @Test
+  public void failWithNameTooLong() {
 
-      var json =
-          """
-          {
-            "createCollection": {
-              "name": "!@-"
-            }
-          }
-          """;
-
-      var command = objectMapper.readValue(json, CreateCollectionCommand.class);
-      var throwable = catchThrowable(() -> resolver.resolveCommand(commandContext, command));
-
-      verifySchemaException(
-          throwable,
-          SchemaException.Code.UNSUPPORTED_SCHEMA_NAME,
-          "The command attempted to create a Collection with a name that is not supported.",
-          "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
-          "The command used the unsupported Collection name: '!@-'.");
-    }
+    var longName = RandomStringUtils.insecure().nextAlphabetic(49);
+    var throwable =
+        assertResolverThrows(
+            "failWithNameTooLong()",
+                """
+                        {
+                            "createCollection": {
+                                "name": "%s"
+                            }
+                        }
+                        """
+                .formatted(longName));
+
+    assertThatSchemaException(throwable)
+        .as("failWithNameTooLong()")
+        .hasCode(SchemaException.Code.UNSUPPORTED_SCHEMA_NAME)
+        .hasMessageSnippets(
+            "The command attempted to create a Collection with a name that is not supported.",
+            "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
+            "The command used the unsupported Collection name: '%s'.".formatted(longName));
   }
 
-  private void verifySchemaException(
-      Throwable throwable, SchemaException.Code expectedErrorCode, String... messageSnippet) {
-
-    assertThat(throwable)
-        .isInstanceOf(SchemaException.class)
-        .satisfies(
-            e -> {
-              SchemaException exception = (SchemaException) e;
-              assertThat(exception.code).isEqualTo(expectedErrorCode.name());
-              for (String snippet : messageSnippet) {
-                assertThat(exception.getMessage()).contains(snippet);
-              }
-            });
+  @Test
+  public void failWithNameSpecialCharacter() {
+
+    var throwable =
+        assertResolverThrows(
+            "failWithNameSpecialCharacter()",
+            """
+                        {
+                            "createCollection": {
+                                "name": "!@-"
+                            }
+                        }
+                        """);
+
+    assertThatSchemaException(throwable)
+        .as("failWithNameSpecialCharacter()")
+        .hasCode(SchemaException.Code.UNSUPPORTED_SCHEMA_NAME)
+        .hasMessageSnippets(
+            "The command attempted to create a Collection with a name that is not supported.",
+            "The supported Collection names must not be empty, more than 48 characters long, or contain non-alphanumeric-underscore characters.",
+            "The command used the unsupported Collection name: '!@-'.");
   }
 }
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverTestBase.java b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverTestBase.java
new file mode 100644
index 0000000000..f05259a895
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverTestBase.java
@@ -0,0 +1,67 @@
+package io.stargate.sgv2.jsonapi.service.resolver;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.datastax.oss.driver.api.core.CqlIdentifier;
+import io.stargate.sgv2.jsonapi.api.model.command.CommandContext;
+import io.stargate.sgv2.jsonapi.api.model.command.impl.CreateCollectionCommand;
+import io.stargate.sgv2.jsonapi.service.operation.collections.CreateCollectionOperation;
+import io.stargate.sgv2.jsonapi.service.schema.KeyspaceSchemaObject;
+import jakarta.inject.Inject;
+
+/**
+ * Re-usable base class for tests for {@link CreateCollectionCommandResolver}.
+ *
+ * <p>Use {@link #assertResolver(String, String, CqlIdentifier)} in this class because it has more
+ * checks about creating a collection
+ *
+ * <p>Needed because we have tests that run with different profiles, such as with vectorize enabled
+ * or disabled, and this puts re-usable code in one place.
+ */
+abstract class CreateCollectionCommandResolverTestBase
+    extends CommandResolverTestBase<
+        KeyspaceSchemaObject,
+        CreateCollectionCommand,
+        CreateCollectionCommandResolver,
+        CreateCollectionOperation> {
+
+  @Inject CreateCollectionCommandResolver RESOLVER;
+
+  protected final CommandContext<KeyspaceSchemaObject> COMMAND_CONTEXT =
+      TEST_CONSTANTS.keyspaceContext();
+
+  @Override
+  protected CreateCollectionCommandResolver resolver() {
+    return RESOLVER;
+  }
+
+  @Override
+  protected CommandContext<KeyspaceSchemaObject> commandContext() {
+    return COMMAND_CONTEXT;
+  }
+
+  @Override
+  protected Class<CreateCollectionCommand> commandClass() {
+    return CreateCollectionCommand.class;
+  }
+
+  /**
+   * Run the command json through the resolver using the super class, then some extra common
+   * assertions for the createCollection command.
+   */
+  protected CreateCollectionOperation assertResolver(
+      String testName, String rawJson, CqlIdentifier collectionName) {
+
+    var operation = super.assertResolver(testName, rawJson);
+
+    assertThat(operation)
+        .isInstanceOfSatisfying(
+            CreateCollectionOperation.class,
+            op -> {
+              assertThat(op.collectionName())
+                  .as("%s - collectionName() matches command", testName)
+                  .isEqualTo(collectionName);
+            });
+    return operation;
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverVectorizeDisabledTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverVectorizeDisabledTest.java
new file mode 100644
index 0000000000..1f71757477
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverVectorizeDisabledTest.java
@@ -0,0 +1,44 @@
+package io.stargate.sgv2.jsonapi.service.resolver;
+
+import static io.stargate.sgv2.jsonapi.util.asserts.DataAPIAsserts.assertThatSchemaException;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.stargate.sgv2.jsonapi.exception.SchemaException;
+import io.stargate.sgv2.jsonapi.util.profiles.DisableVectorizeProfile;
+import org.junit.jupiter.api.Test;
+
+@QuarkusTest
+@TestProfile(DisableVectorizeProfile.class)
+public class CreateCollectionCommandResolverVectorizeDisabledTest
+    extends CreateCollectionCommandResolverTestBase {
+
+  @Test
+  public void failVectorizeSearchDisabled() {
+
+    var throwable =
+        assertResolverThrows(
+            "failVectorizeSearchDisabled()",
+            """
+                            {
+                              	"createCollection": {
+                              		"name": "my_collection",
+                              		"options": {
+                              			"vector": {
+                              				"metric": "cosine",
+                              				"dimension": 1024,
+                              				"service": {
+                              					"provider": "nvidia",
+                              					"modelName": "NV-Embed-QA"
+                              				}
+                              			}
+                              		}
+                              	}
+                              }
+                                """);
+
+    assertThatSchemaException(throwable)
+        .as("failVectorizeSearchDisabled()")
+        .hasCode(SchemaException.Code.VECTORIZE_FEATURE_NOT_AVAILABLE);
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverVectorizeEnabledTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverVectorizeEnabledTest.java
new file mode 100644
index 0000000000..015a88e27d
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionCommandResolverVectorizeEnabledTest.java
@@ -0,0 +1,69 @@
+package io.stargate.sgv2.jsonapi.service.resolver;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.stargate.sgv2.jsonapi.api.model.command.impl.CreateCollectionCommand;
+import io.stargate.sgv2.jsonapi.api.model.command.impl.VectorizeConfig;
+import io.stargate.sgv2.jsonapi.service.schema.EmbeddingSourceModel;
+import io.stargate.sgv2.jsonapi.util.profiles.EnabledVectorizeProfile;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests how the {@link CreateCollectionCommandResolver} handles inputs and the operation it
+ * creates.
+ *
+ * <p><b>NOTE:</b> subclassed atleast by {@link
+ * CreateCollectionCommandResolverVectorizeDisabledTest} to change the vectorize enabled setting
+ */
+@QuarkusTest
+@TestProfile(EnabledVectorizeProfile.class)
+class CreateCollectionCommandResolverVectorizeEnabledTest
+    extends CreateCollectionCommandResolverTestBase {
+
+  @Test
+  public void successWithVectorize() {
+    var operation =
+        assertResolver(
+            "successWithVector()",
+            """
+                        {
+                            "createCollection": {
+                                "name": "${collection}",
+                                "options": {
+                                    "vector": {
+                                        "metric": "cosine",
+                                        "dimension": 768,
+                                        "service": {
+                                            "provider": "azureOpenAI",
+                                            "modelName": "text-embedding-3-small",
+                                            "parameters": {
+                                                "resourceName": "testResourceName",
+                                                "deploymentId": "testResourceName"
+                                            }
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                        """);
+
+    // NOTE: source model of null turns into DEFAULT
+    // aaron 15 june 26 - not sure why but we need the name of the source model must be the CQL name
+    // cql cares about capitals, we dont when processing this normally
+    var expectedVectorDesc =
+        new CreateCollectionCommand.Options.VectorSearchDesc(
+            768,
+            "cosine",
+            EmbeddingSourceModel.DEFAULT.cqlName(),
+            new VectorizeConfig(
+                "azureOpenAI",
+                "text-embedding-3-small",
+                null,
+                Map.of("resourceName", "testResourceName", "deploymentId", "testResourceName")));
+
+    assertThat(operation.vectorDesc()).isEqualTo(expectedVectorDesc);
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionResolverVectorizeDisabledTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionResolverVectorizeDisabledTest.java
deleted file mode 100644
index e42f6d7d7a..0000000000
--- a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/CreateCollectionResolverVectorizeDisabledTest.java
+++ /dev/null
@@ -1,78 +0,0 @@
-package io.stargate.sgv2.jsonapi.service.resolver;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.catchException;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import io.quarkus.test.junit.QuarkusTest;
-import io.quarkus.test.junit.TestProfile;
-import io.stargate.sgv2.jsonapi.TestConstants;
-import io.stargate.sgv2.jsonapi.api.model.command.impl.CreateCollectionCommand;
-import io.stargate.sgv2.jsonapi.api.model.command.impl.FindEmbeddingProvidersCommand;
-import io.stargate.sgv2.jsonapi.exception.SchemaException;
-import jakarta.inject.Inject;
-import org.junit.jupiter.api.Test;
-
-@QuarkusTest
-@TestProfile(DisableVectorizeProfile.class)
-public class CreateCollectionResolverVectorizeDisabledTest {
-  @Inject ObjectMapper objectMapper;
-  @Inject CreateCollectionCommandResolver createCollectionCommandResolver;
-  @Inject FindEmbeddingProvidersCommandResolver findEmbeddingProvidersCommandResolver;
-  private final TestConstants testConstants = new TestConstants();
-
-  @Test
-  public void vectorizeSearchDisabled() throws Exception {
-    String json =
-        """
-                {
-                    "createCollection": {
-                        "name": "my_collection",
-                        "options": {
-                            "vector": {
-                                "metric": "cosine",
-                                "dimension": 1024,
-                                "service": {
-                                    "provider": "nvidia",
-                                    "modelName": "NV-Embed-QA"
-                                }
-                            }
-                        }
-                    }
-                }
-              """;
-
-    CreateCollectionCommand command = objectMapper.readValue(json, CreateCollectionCommand.class);
-    Exception e =
-        catchException(
-            () ->
-                createCollectionCommandResolver.resolveCommand(
-                    testConstants.keyspaceContext(), command));
-    assertThat(e)
-        .hasFieldOrPropertyWithValue(
-            "code", SchemaException.Code.VECTORIZE_FEATURE_NOT_AVAILABLE.name());
-  }
-
-  @Test
-  public void findEmbeddingProvidersWithVectorizeSearchDisabled() throws Exception {
-    // TODO: This test should be moved, these are rests for create collection NOT for
-    // findEmbeddingProviders.
-
-    String json =
-        """
-                  {
-                      "findEmbeddingProviders": {}
-                  }
-                  """;
-    FindEmbeddingProvidersCommand command =
-        objectMapper.readValue(json, FindEmbeddingProvidersCommand.class);
-    Exception e =
-        catchException(
-            () ->
-                findEmbeddingProvidersCommandResolver.resolveCommand(
-                    testConstants.databaseContext(), command));
-    assertThat(e)
-        .hasFieldOrPropertyWithValue(
-            "code", SchemaException.Code.VECTORIZE_FEATURE_NOT_AVAILABLE.name());
-  }
-}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/FindEmbeddingProvidersResolverVectorizeDisabledTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/FindEmbeddingProvidersResolverVectorizeDisabledTest.java
new file mode 100644
index 0000000000..d785a89937
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/FindEmbeddingProvidersResolverVectorizeDisabledTest.java
@@ -0,0 +1,67 @@
+package io.stargate.sgv2.jsonapi.service.resolver;
+
+import static io.stargate.sgv2.jsonapi.util.asserts.DataAPIAsserts.assertThatSchemaException;
+
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+import io.stargate.sgv2.jsonapi.api.model.command.CommandContext;
+import io.stargate.sgv2.jsonapi.api.model.command.impl.FindEmbeddingProvidersCommand;
+import io.stargate.sgv2.jsonapi.exception.SchemaException;
+import io.stargate.sgv2.jsonapi.service.operation.embeddings.FindEmbeddingProvidersOperation;
+import io.stargate.sgv2.jsonapi.service.schema.DatabaseSchemaObject;
+import io.stargate.sgv2.jsonapi.util.profiles.DisableVectorizeProfile;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration tests for
+ *
+ * <pre>findEmbeddingProviders</pre>
+ *
+ * when vectorize is disabled.
+ */
+@QuarkusTest
+@TestProfile(DisableVectorizeProfile.class)
+public class FindEmbeddingProvidersResolverVectorizeDisabledTest
+    extends CommandResolverTestBase<
+        DatabaseSchemaObject,
+        FindEmbeddingProvidersCommand,
+        FindEmbeddingProvidersCommandResolver,
+        FindEmbeddingProvidersOperation> {
+
+  @Inject FindEmbeddingProvidersCommandResolver RESOLVER;
+  private final CommandContext<DatabaseSchemaObject> COMMAND_CONTEXT =
+      TEST_CONSTANTS.databaseContext();
+
+  @Override
+  protected FindEmbeddingProvidersCommandResolver resolver() {
+    return RESOLVER;
+  }
+
+  @Override
+  protected CommandContext<DatabaseSchemaObject> commandContext() {
+    return COMMAND_CONTEXT;
+  }
+
+  @Override
+  protected Class<FindEmbeddingProvidersCommand> commandClass() {
+    return FindEmbeddingProvidersCommand.class;
+  }
+
+  @Test
+  public void failVectorizeSearchDisabled() {
+
+    var throwable =
+        assertResolverThrows(
+            "failVectorizeSearchDisabled()",
+            """
+                            {
+                              "findEmbeddingProviders": {}
+                            }
+                            """);
+
+    assertThatSchemaException(throwable)
+        .as("failVectorizeSearchDisabled()")
+        .hasCode(SchemaException.Code.VECTORIZE_FEATURE_NOT_AVAILABLE);
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/APIExceptionAssert.java b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/APIExceptionAssert.java
new file mode 100644
index 0000000000..a8f35f2bd9
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/APIExceptionAssert.java
@@ -0,0 +1,106 @@
+package io.stargate.sgv2.jsonapi.util.asserts;
+
+import static io.stargate.sgv2.jsonapi.util.ClassUtils.classSimpleName;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+import io.stargate.sgv2.jsonapi.exception.APIException;
+import io.stargate.sgv2.jsonapi.exception.ErrorCode;
+import java.util.function.Function;
+import org.assertj.core.api.AbstractAssert;
+
+/**
+ * Base for Assert classes that are designed to assert against {@link APIException} sub-classes.
+ * Most of the work is done in this class, the subclass just brings the type declarations.
+ *
+ * <p>For example, from {@link SchemaExceptionAssert}, the subclass sets the types, and declares a
+ * factory:
+ *
+ * <pre>
+ *   public class SchemaExceptionAssert
+ *     extends APIExceptionAssert<SchemaExceptionAssert, SchemaException> {
+ *
+ *   private SchemaExceptionAssert(SchemaException actual) {
+ *     super(actual, SchemaExceptionAssert.class);
+ *   }
+ *
+ *   protected static SchemaExceptionAssert assertThatSchemaException(SchemaException schemaException) {
+ *     return assertThatAPIException(
+ *         SchemaExceptionAssert::new, SchemaException.class, schemaException);
+ *   }
+ *   protected static SchemaExceptionAssert assertThatSchemaException(Throwable throwable) {
+ *     return assertThatAPIException(SchemaExceptionAssert::new, SchemaException.class, throwable);
+ *   }
+ *   ...
+ * </pre>
+ *
+ * See {@link #assertThatAPIException(Function, Class, Throwable)}
+ *
+ * @param <SELF> type of the subclass, used for fluent API
+ * @param <ERROR> Type of the API Exception class, e.g. {@link
+ *     io.stargate.sgv2.jsonapi.exception.SchemaException}
+ */
+public abstract class APIExceptionAssert<
+        SELF extends APIExceptionAssert<SELF, ERROR>, ERROR extends APIException>
+    extends AbstractAssert<SELF, ERROR> {
+
+  /**
+   * Instantiate with values from the subclass.
+   *
+   * @param actual The actual error instance the assertions will run against.
+   * @param selfClazz Class for {@link SELF}
+   */
+  protected APIExceptionAssert(ERROR actual, Class<SELF> selfClazz) {
+    super(actual, selfClazz);
+  }
+
+  /**
+   * Factory that subclasses can reuse for all logic to create a new instance, checks that actual
+   * throwable is of the expected class.
+   *
+   * @param assertCtor Constructor for the subclass, called to create an instance after checking
+   * @param errorClass Class for {@link ERROR} that throwable must match.
+   * @param throwable Throwable we want to asset, must be instance of {@link ERROR}
+   * @return New assert sub class.
+   * @param <SELF> Type of subclass, same as for the class.
+   * @param <ERROR> type of the APIException, same as for class.
+   */
+  protected static <SELF extends APIExceptionAssert<SELF, ERROR>, ERROR extends APIException>
+      SELF assertThatAPIException(
+          Function<ERROR, SELF> assertCtor, Class<ERROR> errorClass, Throwable throwable) {
+
+    assertThat(throwable)
+        .as("Throwable is instance of: " + classSimpleName(errorClass))
+        .isInstanceOf(errorClass);
+
+    return assertCtor.apply(errorClass.cast(throwable));
+  }
+
+  /**
+   * Asserts that {@link APIException#code} matches the supplied {@link ErrorCode}
+   *
+   * @param expected Expected error code.
+   * @return Self
+   */
+  public SELF hasCode(ErrorCode<?> expected) {
+    isNotNull();
+    assertThat(actual.code).as("APIException.code() is expected").isEqualTo(expected.name());
+    return myself;
+  }
+
+  /**
+   * Asserts that the {@link APIException#getMessage()} string contains each of the provided message
+   * snippets.
+   *
+   * @param snippets vararg strings to check for
+   * @return Self
+   */
+  public SELF hasMessageSnippets(String... snippets) {
+    isNotNull();
+    for (String snippet : snippets) {
+      assertThat(actual.getMessage())
+          .as("APIException.message() contains expected message snippet")
+          .contains(snippet);
+    }
+    return myself;
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/CommandResultAssert.java b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/CommandResultAssert.java
new file mode 100644
index 0000000000..e52b1d8507
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/CommandResultAssert.java
@@ -0,0 +1,85 @@
+package io.stargate.sgv2.jsonapi.util.asserts;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.stargate.sgv2.jsonapi.api.model.command.CommandResult;
+import io.stargate.sgv2.jsonapi.api.model.command.CommandStatus;
+import io.stargate.sgv2.jsonapi.api.v1.ResponseAssertions;
+import io.stargate.sgv2.jsonapi.exception.ErrorCode;
+import org.assertj.core.api.AbstractAssert;
+
+/**
+ * Assertions for matching the {@link CommandResult}
+ *
+ * <p>NOTE: there is some logic duplication with {@link ResponseAssertions} which is used for
+ * integration tests which tie in with RestEasy. It would be good to have some underlying logic but
+ * that can be future refactoring
+ */
+public class CommandResultAssert extends AbstractAssert<CommandResultAssert, CommandResult> {
+
+  private CommandResultAssert(CommandResult commandResult) {
+    super(commandResult, CommandResultAssert.class);
+  }
+
+  protected static CommandResultAssert assertThatCommandResult(CommandResult commandResult) {
+
+    // must always be non null
+    assertThat(commandResult).as("CommandResult is not null").isNotNull();
+    return new CommandResultAssert(commandResult);
+  }
+
+  public CommandResultAssert isDDLSuccess() {
+
+    assertThat(actual.data()).as("isDDLSuccess() - data is null").isNull();
+    assertThat(actual.status()).as("isError() - status is no empty").isNotEmpty();
+    assertThat(actual.errors()).as("isDDLSuccess() - errors is empty").isEmpty();
+    return this;
+  }
+
+  public CommandResultAssert isError() {
+
+    assertThat(actual.data()).as("isError() - data is null").isNull();
+    assertThat(actual.status()).as("isError() - status is empty").isEmpty();
+    assertThat(actual.errors()).as("isError() - errors is not null").isNotNull();
+    assertThat(actual.errors()).as("isError() - errors is not empty").isNotEmpty();
+    return this;
+  }
+
+  public CommandResultAssert hasOnlyStatus(CommandStatus key, Object value) {
+    assertThat(actual.status()).as("status is not null").isNotNull();
+    assertThat(actual.status()).as("status is only one").hasSize(1);
+    assertThat(actual.status().get(key)).as("status['%s']", key).isEqualTo(value);
+
+    return this;
+  }
+
+  public CommandResultAssert hasNoStatus() {
+    assertThat(actual.status()).as("status is empty").isEmpty();
+    return this;
+  }
+
+  public CommandResultAssert hasNoErrors() {
+    assertThat(actual.errors()).as("errors is empty").isEmpty();
+    return this;
+  }
+
+  public CommandResultAssert hasErrorCount(int count) {
+    assertThat(actual.errors()).as("errors is not null").isNotNull();
+    assertThat(actual.errors()).as("errors size").hasSize(count);
+    return this;
+  }
+
+  public CommandResultAssert hasOnlyError(ErrorCode<?> errorCode, String... snippets) {
+
+    assertThat(actual.errors()).as("errors is not null").isNotNull();
+    assertThat(actual.errors()).as("errors size").hasSize(1);
+
+    var commandError = actual.errors().getFirst();
+    assertThat(commandError.errorCode()).as("errorCode").isEqualTo(errorCode.name());
+
+    for (var snippet : snippets) {
+      assertThat(commandError.message()).as("message").contains(snippet);
+    }
+    return this;
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/DataAPIAsserts.java b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/DataAPIAsserts.java
new file mode 100644
index 0000000000..df43ae29f3
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/DataAPIAsserts.java
@@ -0,0 +1,28 @@
+package io.stargate.sgv2.jsonapi.util.asserts;
+
+import io.stargate.sgv2.jsonapi.api.model.command.CommandResult;
+import io.stargate.sgv2.jsonapi.exception.SchemaException;
+
+/**
+ * Central place to export all new assertions added for the Data API, e.g. same as {@link
+ * org.assertj.core.api.Assertions}.
+ *
+ * <p>Assertion helpers in this package should write package-protected factories in their class, and
+ * then add them here as public.
+ */
+public class DataAPIAsserts {
+
+  private DataAPIAsserts() {}
+
+  public static CommandResultAssert assertThatCommandResult(CommandResult commandResult) {
+    return CommandResultAssert.assertThatCommandResult(commandResult);
+  }
+
+  public static SchemaExceptionAssert assertThatSchemaException(Throwable throwable) {
+    return SchemaExceptionAssert.assertThatSchemaException(throwable);
+  }
+
+  public static SchemaExceptionAssert assertThatSchemaException(SchemaException schemaException) {
+    return SchemaExceptionAssert.assertThatSchemaException(schemaException);
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/SchemaExceptionAssert.java b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/SchemaExceptionAssert.java
new file mode 100644
index 0000000000..41882fcdc9
--- /dev/null
+++ b/src/test/java/io/stargate/sgv2/jsonapi/util/asserts/SchemaExceptionAssert.java
@@ -0,0 +1,24 @@
+package io.stargate.sgv2.jsonapi.util.asserts;
+
+import io.stargate.sgv2.jsonapi.exception.SchemaException;
+
+/** See {@link APIExceptionAssert} and {@link DataAPIAsserts} */
+public class SchemaExceptionAssert
+    extends APIExceptionAssert<SchemaExceptionAssert, SchemaException> {
+
+  private SchemaExceptionAssert(SchemaException actual) {
+    super(actual, SchemaExceptionAssert.class);
+  }
+
+  protected static SchemaExceptionAssert assertThatSchemaException(
+      SchemaException schemaException) {
+
+    return assertThatAPIException(
+        SchemaExceptionAssert::new, SchemaException.class, schemaException);
+  }
+
+  protected static SchemaExceptionAssert assertThatSchemaException(Throwable throwable) {
+
+    return assertThatAPIException(SchemaExceptionAssert::new, SchemaException.class, throwable);
+  }
+}
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/DisableVectorizeProfile.java b/src/test/java/io/stargate/sgv2/jsonapi/util/profiles/DisableVectorizeProfile.java
similarity index 75%
rename from src/test/java/io/stargate/sgv2/jsonapi/service/resolver/DisableVectorizeProfile.java
rename to src/test/java/io/stargate/sgv2/jsonapi/util/profiles/DisableVectorizeProfile.java
index ddf69aaf18..46318decba 100644
--- a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/DisableVectorizeProfile.java
+++ b/src/test/java/io/stargate/sgv2/jsonapi/util/profiles/DisableVectorizeProfile.java
@@ -1,10 +1,14 @@
-package io.stargate.sgv2.jsonapi.service.resolver;
+package io.stargate.sgv2.jsonapi.util.profiles;
 
 import com.google.common.collect.ImmutableMap;
 import io.quarkus.test.junit.QuarkusTestProfile;
 import java.util.Map;
 
+/**
+ * Disables vectorize via operations config, while also disables spinning up DB as a test resource.
+ */
 public class DisableVectorizeProfile implements QuarkusTestProfile {
+
   @Override
   public boolean disableGlobalTestResources() {
     return true;
diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/EnabledVectorizeProfile.java b/src/test/java/io/stargate/sgv2/jsonapi/util/profiles/EnabledVectorizeProfile.java
similarity index 75%
rename from src/test/java/io/stargate/sgv2/jsonapi/service/resolver/EnabledVectorizeProfile.java
rename to src/test/java/io/stargate/sgv2/jsonapi/util/profiles/EnabledVectorizeProfile.java
index 64fe0da76a..6e2808e9c9 100644
--- a/src/test/java/io/stargate/sgv2/jsonapi/service/resolver/EnabledVectorizeProfile.java
+++ b/src/test/java/io/stargate/sgv2/jsonapi/util/profiles/EnabledVectorizeProfile.java
@@ -1,10 +1,14 @@
-package io.stargate.sgv2.jsonapi.service.resolver;
+package io.stargate.sgv2.jsonapi.util.profiles;
 
 import com.google.common.collect.ImmutableMap;
 import io.quarkus.test.junit.QuarkusTestProfile;
 import java.util.Map;
 
+/**
+ * Enables vectorize via operations config, while also disables spinning up DB as a test resource.
+ */
 public class EnabledVectorizeProfile implements QuarkusTestProfile {
+
   @Override
   public boolean disableGlobalTestResources() {
     return true;