BaseEntityIT.java

package org.openmetadata.it.tests;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import lombok.extern.slf4j.Slf4j;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
import org.openmetadata.it.bootstrap.SharedEntities;
import org.openmetadata.it.util.EntityValidation;
import org.openmetadata.it.util.SdkClients;
import org.openmetadata.it.util.TestNamespace;
import org.openmetadata.it.util.TestNamespaceExtension;
import org.openmetadata.it.util.UpdateType;
import org.openmetadata.schema.EntityInterface;
import org.openmetadata.schema.type.ApiStatus;
import org.openmetadata.schema.type.ChangeDescription;
import org.openmetadata.schema.type.TagLabel;
import org.openmetadata.schema.type.api.BulkOperationResult;
import org.openmetadata.schema.type.api.BulkResponse;
import org.openmetadata.schema.type.csv.CsvImportResult;
import org.openmetadata.schema.utils.JsonUtils;
import org.openmetadata.sdk.client.OpenMetadataClient;
import org.openmetadata.sdk.exceptions.InvalidRequestException;
import org.openmetadata.sdk.fluent.Users;
import org.openmetadata.sdk.network.HttpMethod;
import org.openmetadata.sdk.network.RequestOptions;
import org.openmetadata.service.Entity;
import org.openmetadata.service.util.EntityUtil;
import org.openmetadata.service.util.TestUtils;

/**
 * Base class for all entity integration tests.
 *
 * Migrated from: org.openmetadata.service.resources.EntityResourceTest
 * Purpose: Provides common test methods that apply to ALL entities (CRUD, permissions, tags, etc.)
 *
 * This class replicates the pattern from EntityResourceTest but uses:
 * - SDK client instead of WebTarget
 * - TestNamespace for isolation instead of TestInfo
 * - Typed exceptions instead of HTTP status codes
 * - Feature flags to conditionally run tests based on entity capabilities
 *
 * Subclasses must implement:
 * - createMinimalRequest() - Create a minimal valid create request
 * - getEntityClient() - Return the appropriate entity client
 * - validateCreatedEntity() - Entity-specific validation
 *
 * @param <T> Entity type (e.g., Database, Table, User)
 * @param <K> Create request type (e.g., CreateDatabase, CreateTable)
 */
@Slf4j
@ExtendWith(TestNamespaceExtension.class)
public abstract class BaseEntityIT<T extends EntityInterface, K> {

  // ===================================================================
  // ABSTRACT METHODS - Must be implemented by subclasses
  // ===================================================================

  /**
   * Create a minimal valid create request for this entity.
   * This should include all required fields but minimal optional fields.
   */
  protected abstract K createMinimalRequest(TestNamespace ns);

  /**
   * Create a request with a specific name.
   */
  protected abstract K createRequest(String name, TestNamespace ns);

  /**
   * Create the entity using the fluent API.
   */
  protected abstract T createEntity(K createRequest);

  /**
   * Get entity by ID.
   */
  protected abstract T getEntity(String id);

  /**
   * Get entity by fully qualified name.
   */
  protected abstract T getEntityByName(String fqn);

  /**
   * Update entity using PATCH (returns updated entity).
   */
  protected abstract T patchEntity(String id, T entity);

  /**
   * Delete entity (soft delete if supported).
   */
  protected abstract void deleteEntity(String id);

  /**
   * Restore a soft-deleted entity.
   */
  protected abstract void restoreEntity(String id);

  /**
   * Hard delete an entity (permanently remove).
   */
  protected abstract void hardDeleteEntity(String id);

  /**
   * Get the entity type name (e.g., "database", "table", "user").
   */
  protected abstract String getEntityType();

  /**
   * Get the resource path for this entity (e.g., "/v1/tables", "/v1/databases").
   * Used for making raw HTTP calls to endpoints not exposed through the SDK.
   */
  protected String getResourcePath() {
    return "/v1/" + TestUtils.plurializeEntityType(getEntityType()) + "/";
  }

  /**
   * Validate that the created entity matches the create request.
   * Subclasses should add entity-specific validations.
   */
  protected abstract void validateCreatedEntity(T entity, K createRequest);

  // ===================================================================
  // FEATURE FLAGS - Control which tests run for this entity
  // ===================================================================

  protected boolean supportsFollowers = true;
  protected boolean supportsOwners = true;
  protected boolean supportsTags = true;
  protected boolean supportsDomains = true;
  protected boolean supportsPatchDomains = true; // Can domains be changed via PATCH after creation?
  protected boolean supportsDataProducts = true;
  protected boolean supportsDataContract = false;
  protected boolean supportsSoftDelete = true;
  protected boolean supportsCustomExtension = true;
  protected boolean supportsFieldsQueryParam = true;
  protected boolean supportsPatch = true;
  protected boolean supportsEmptyDescription = true;
  protected boolean supportsNameLengthValidation = true;
  protected boolean supportsBulkAPI = false; // Override in subclasses that support bulk API
  protected boolean supportsSearchIndex = true; // Override in subclasses that don't support search
  protected boolean supportsVersionHistory =
      true; // Override in subclasses that don't support version history
  protected boolean supportsGetByVersion =
      true; // Override if get specific version is not supported
  protected boolean supportsIncludeDeleted =
      true; // Override if include=deleted query param not supported
  protected boolean supportsImportExport =
      false; // Override in subclasses that support CSV import/export
  protected boolean supportsListHistoryByTimestamp =
      false; // Override in subclasses that support listing all versions by timestamp

  // ===================================================================
  // CHANGE TYPE - Controls how version changes are validated
  // ===================================================================

  /**
   * Get the default change type for updates in session.
   * Most entities use MINOR_UPDATE, but some may need CHANGE_CONSOLIDATED
   * when changes happen within the same session window.
   *
   * <p>Matches EntityResourceTest.getChangeType() pattern.
   *
   * @return The default UpdateType for changes made within the same session
   */
  public UpdateType getChangeType() {
    return UpdateType.MINOR_UPDATE;
  }

  // ===================================================================
  // SHARED ENTITY ACCESSORS - Session-scoped entities for cross-test use
  // ===================================================================

  /**
   * Access to session-scoped shared entities.
   * These entities are created ONCE at session start and shared across all tests.
   * Tests should use these as read-only references for ownership, team membership, etc.
   */
  protected SharedEntities shared() {
    return SharedEntities.get();
  }

  /** Convenience accessor for shared USER1 */
  protected org.openmetadata.schema.entity.teams.User testUser1() {
    return shared().USER1;
  }

  /** Convenience accessor for shared USER2 */
  protected org.openmetadata.schema.entity.teams.User testUser2() {
    return shared().USER2;
  }

  /** Convenience accessor for shared USER3 */
  protected org.openmetadata.schema.entity.teams.User testUser3() {
    return shared().USER3;
  }

  /** Convenience accessor for shared USER1 EntityReference */
  protected org.openmetadata.schema.type.EntityReference testUser1Ref() {
    return shared().USER1_REF;
  }

  /** Convenience accessor for shared USER2 EntityReference */
  protected org.openmetadata.schema.type.EntityReference testUser2Ref() {
    return shared().USER2_REF;
  }

  /** Convenience accessor for shared TEAM1 */
  protected org.openmetadata.schema.entity.teams.Team testTeam1() {
    return shared().TEAM1;
  }

  /** Convenience accessor for shared TEAM2 */
  protected org.openmetadata.schema.entity.teams.Team testTeam2() {
    return shared().TEAM2;
  }

  /** Convenience accessor for shared TEAM11 (Group type - can own entities) */
  protected org.openmetadata.schema.entity.teams.Team testGroupTeam() {
    return shared().TEAM11;
  }

  /** Convenience accessor for shared DOMAIN */
  protected org.openmetadata.schema.entity.domains.Domain testDomain() {
    return shared().DOMAIN;
  }

  /** Convenience accessor for shared SUB_DOMAIN */
  protected org.openmetadata.schema.entity.domains.Domain testSubDomain() {
    return shared().SUB_DOMAIN;
  }

  /** Convenience accessor for shared DATA_STEWARD_ROLE */
  protected org.openmetadata.schema.entity.teams.Role dataStewardRole() {
    return shared().DATA_STEWARD_ROLE;
  }

  /** Convenience accessor for shared DATA_CONSUMER_ROLE */
  protected org.openmetadata.schema.entity.teams.Role dataConsumerRole() {
    return shared().DATA_CONSUMER_ROLE;
  }

  /** Convenience accessor for shared PERSONAL_DATA_TAG_LABEL */
  protected TagLabel personalDataTagLabel() {
    return shared().PERSONAL_DATA_TAG_LABEL;
  }

  /** Convenience accessor for shared PII_SENSITIVE_TAG_LABEL */
  protected TagLabel piiSensitiveTagLabel() {
    return shared().PII_SENSITIVE_TAG_LABEL;
  }

  /** Convenience accessor for shared GLOSSARY1_TERM1_LABEL */
  protected TagLabel glossaryTermLabel() {
    return shared().GLOSSARY1_TERM1_LABEL;
  }

  /** Convenience accessor for shared MYSQL_SERVICE reference */
  protected org.openmetadata.schema.type.EntityReference mysqlServiceRef() {
    return shared().MYSQL_REFERENCE;
  }

  // ===================================================================
  // VALIDATION HELPER METHODS
  // ===================================================================

  /**
   * Helper method to patch entity and validate version, ChangeDescription.
   * Similar to patchEntityAndCheck in EntityResourceTest.
   *
   * @param entity Entity with updated values
   * @param updateType Expected type of update
   * @param expectedChange Expected ChangeDescription (can be null for simple cases)
   * @return Updated entity
   */
  protected T patchEntityAndCheck(
      T entity, UpdateType updateType, ChangeDescription expectedChange) {
    // Capture version before update
    Double previousVersion = entity.getVersion();

    // Perform the update
    T updated = patchEntity(entity.getId().toString(), entity);

    // Validate version changed correctly
    EntityValidation.validateVersion(updated, updateType, previousVersion);

    // Validate ChangeDescription
    EntityValidation.validateChangeDescription(updated, updateType, expectedChange);

    // Verify changes persisted by getting entity again
    T fetched = getEntity(updated.getId().toString());
    assertEquals(updated.getVersion(), fetched.getVersion(), "Version mismatch after fetch");

    return updated;
  }

  // ===================================================================
  // COMMON CRUD TESTS (Phase 1 - 10 tests)
  // ===================================================================

  /**
   * Test: Create entity with minimal required fields
   * Equivalent to: post_entityCreate_200_OK in EntityResourceTest
   */
  @Test
  void post_entityCreate_200_OK(TestNamespace ns) {

    // Create entity with minimal fields
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Common validations
    assertNotNull(entity.getId(), "Entity ID should not be null");
    assertNotNull(entity.getFullyQualifiedName(), "Entity FQN should not be null");
    assertNotNull(entity.getName(), "Entity name should not be null");

    // Validate version and ChangeDescription for newly created entity
    EntityValidation.validateVersion(entity, UpdateType.CREATED, null);
    EntityValidation.validateChangeDescription(entity, UpdateType.CREATED, null);

    // Entity-specific validations
    validateCreatedEntity(entity, createRequest);

    // Verify entity can be retrieved
    T fetched = getEntity(entity.getId().toString());
    assertEquals(entity.getId(), fetched.getId());
    assertEquals(entity.getFullyQualifiedName(), fetched.getFullyQualifiedName());
    assertEquals(entity.getVersion(), fetched.getVersion(), "Version should match after fetch");
  }

  /**
   * Test: Create entity with invalid name (special characters, too long, etc.)
   * Equivalent to: post_entityCreateWithInvalidName_400 in EntityResourceTest
   */
  @Test
  void post_entityCreateWithInvalidName_400(TestNamespace ns) {

    // Test invalid name patterns that OpenMetadata actually rejects
    // Empty name
    K emptyNameRequest = createRequest("", ns);
    assertThrows(
        InvalidRequestException.class,
        () -> createEntity(emptyNameRequest),
        "Expected InvalidRequestException for empty name");

    // Name with newline character
    K newlineNameRequest = createRequest("name with\nnewline", ns);
    assertThrows(
        InvalidRequestException.class,
        () -> createEntity(newlineNameRequest),
        "Expected InvalidRequestException for name with newline");

    // Name too long (>256 chars typically) - only if entity supports this validation
    if (supportsNameLengthValidation) {
      K longNameRequest = createRequest("a".repeat(300), ns);
      assertThrows(
          InvalidRequestException.class,
          () -> createEntity(longNameRequest),
          "Expected InvalidRequestException for name too long");
    }
  }

  /**
   * Test: Create duplicate entity should fail
   * Equivalent to: post_entityAlreadyExists_409_conflict in EntityResourceTest
   */
  @Test
  void post_duplicateEntity_409(TestNamespace ns) {

    // Create first entity with a specific request
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    assertNotNull(entity.getId());

    // Attempt to create duplicate using the SAME create request
    // This ensures we're truly creating a duplicate (same name, same parent service)
    assertThrows(
        Exception.class, // May be ConflictException or similar
        () -> createEntity(createRequest),
        "Creating duplicate entity should fail");
  }

  /**
   * Test: Get entity by ID
   * Equivalent to: get_entity_200_OK in EntityResourceTest
   */
  @Test
  void get_entity_200_OK(TestNamespace ns) {

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Get entity by ID
    T fetched = getEntity(created.getId().toString());

    assertNotNull(fetched);
    assertEquals(created.getId(), fetched.getId());
    assertEquals(created.getName(), fetched.getName());
    assertEquals(created.getFullyQualifiedName(), fetched.getFullyQualifiedName());
  }

  /**
   * Test: Get entity by fully qualified name
   * Equivalent to: get_entityByName_200_OK in EntityResourceTest
   */
  @Test
  void get_entityByName_200_OK(TestNamespace ns) {

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Get entity by FQN
    T fetched = getEntityByName(created.getFullyQualifiedName());

    assertNotNull(fetched);
    assertEquals(created.getId(), fetched.getId());
    assertEquals(created.getFullyQualifiedName(), fetched.getFullyQualifiedName());
  }

  /**
   * Test: Get non-existent entity should fail
   * Equivalent to: get_entityNotFound_404 in EntityResourceTest
   */
  @Test
  void get_entityNotFound_404(TestNamespace ns) {

    // Try to get non-existent entity by ID
    String fakeId = "00000000-0000-0000-0000-000000000000";
    assertThrows(
        Exception.class, () -> getEntity(fakeId), "Getting non-existent entity should fail");

    // Try to get non-existent entity by name
    String fakeName = ns.prefix("nonExistent");
    assertThrows(
        Exception.class,
        () -> getEntityByName(fakeName),
        "Getting non-existent entity by name should fail");
  }

  /**
   * Test: Update entity description
   * Equivalent to: patch_entityAttributes_200_OK in EntityResourceTest
   */
  @Test
  void patch_entityDescription_200_OK(TestNamespace ns) {
    if (!supportsPatch) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Validate initial version
    assertEquals(0.1, created.getVersion(), 0.001, "Initial version should be 0.1");

    // Update description
    String oldDescription = created.getDescription();
    String newDescription = "Updated description via integration test";
    created.setDescription(newDescription);

    // Create expected ChangeDescription
    ChangeDescription expectedChange =
        EntityValidation.getChangeDescription(created, UpdateType.MINOR_UPDATE);
    EntityValidation.fieldUpdated(expectedChange, "description", oldDescription, newDescription);

    // Perform update with validation
    T updated = patchEntityAndCheck(created, UpdateType.MINOR_UPDATE, expectedChange);

    // Validate description was updated
    assertEquals(newDescription, updated.getDescription());
    assertEquals(0.2, updated.getVersion(), 0.001, "Version should increment to 0.2");

    // Verify ChangeDescription is present and correct
    assertNotNull(updated.getChangeDescription(), "ChangeDescription should be present");
    assertEquals(
        0.1,
        updated.getChangeDescription().getPreviousVersion(),
        0.001,
        "Previous version should be 0.1");
  }

  /**
   * Test: Soft delete entity
   * Equivalent to: delete_entity_soft_200 in EntityResourceTest
   */
  @Test
  void delete_entity_soft_200(TestNamespace ns) {
    if (!supportsSoftDelete) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    String entityId = created.getId().toString();

    // Soft delete
    deleteEntity(entityId);

    // Entity should not be retrievable by default
    assertThrows(
        Exception.class, () -> getEntity(entityId), "Deleted entity should not be retrievable");
  }

  // ===================================================================
  // AUTHORIZATION TESTS (Phase 1 - JWT-based)
  // ===================================================================

  /**
   * Test: Non-admin user cannot create entity
   * Equivalent to: post_entity_as_non_admin_401 in EntityResourceTest
   *
   * Uses JWT-based authentication with a regular user (no admin role).
   */
  // TODO: Authorization tests need rework for fluent API pattern
  // These tests require using different clients (admin vs non-admin)
  // which conflicts with the static default client approach.
  // Subclasses can implement entity-specific authorization tests if needed.

  //  @Test
  //  void post_entity_as_non_admin_401(TestNamespace ns) { ... }
  //  @Test
  //  void delete_entity_as_non_admin_401(TestNamespace ns) { ... }

  /**
   * Test: Creating duplicate entity should fail with conflict
   * Equivalent to: post_entityAlreadyExists_409_conflict in EntityResourceTest
   */
  @Test
  public void post_entityAlreadyExists_409_conflict(TestNamespace ns) {

    // Create first entity
    K createRequest = createRequest(ns.prefix("duplicate"), ns);
    T created = createEntity(createRequest);
    assertNotNull(created.getId());

    // Attempt to create duplicate - should fail
    assertThrows(
        Exception.class,
        () -> createEntity(createRequest),
        "Creating duplicate entity should fail with conflict");
  }

  /**
   * Test: Entity names with dots should be properly quoted in FQN
   * Equivalent to: post_entityWithDots_200 in EntityResourceTest
   */
  @Test
  void post_entityWithDots_200(TestNamespace ns) {

    // Create entity with dots in name
    String nameWithDots = ns.prefix("foo.bar");
    K createRequest = createRequest(nameWithDots, ns);
    T created = createEntity(createRequest);

    // Verify entity created
    assertNotNull(created.getId());
    assertEquals(nameWithDots, created.getName());

    // FQN should contain quotes if hierarchical, or exact name if not
    String fqn = created.getFullyQualifiedName();
    boolean isHierarchical = !fqn.equals(created.getName());
    if (isHierarchical) {
      assertTrue(fqn.contains("\""), "Hierarchical FQN with dots should contain quotes: " + fqn);
    }
  }

  // ===================================================================
  // PUT (UPSERT) TESTS
  // ===================================================================

  /**
   * Test: PUT can create new entity (upsert)
   * Equivalent to: put_entityCreate_200 in EntityResourceTest
   */
  @Test
  void put_entityCreate_200(TestNamespace ns) {

    // Create entity using PUT (upsert with no existing entity)
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Verify entity was created
    assertNotNull(created.getId());
    assertEquals(0.1, created.getVersion(), 0.001);
    validateCreatedEntity(created, createRequest);
  }

  /**
   * Test: PUT with no changes should not increment version
   * Equivalent to: put_entityUpdateWithNoChange_200 in EntityResourceTest
   */
  @Test
  void put_entityUpdateWithNoChange_200(TestNamespace ns) {
    if (!supportsPatch) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    Double originalVersion = created.getVersion();

    // Update with same data (no change)
    T updated = patchEntity(created.getId().toString(), created);

    // Version should NOT change when there's no actual change
    assertEquals(
        originalVersion, updated.getVersion(), 0.001, "Version should not change for no-op update");
  }

  // ===================================================================
  // DELETE RESTORE TESTS
  // ===================================================================

  /**
   * Test: Soft deleted entity can be restored
   * Equivalent to: delete_restore_entity_200 in EntityResourceTest
   *
   * Note: This test verifies the restore functionality for soft-deleted entities.
   * The original test uses a PUT request to restore. Since we're using the SDK,
   * we'll verify that after soft delete, the entity can be restored and is functional.
   */
  @Test
  void delete_restore_entity_200(TestNamespace ns) {
    if (!supportsSoftDelete) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    String entityId = created.getId().toString();

    // Soft delete entity
    deleteEntity(entityId);

    // Verify entity is deleted (should not be retrievable by default)
    assertThrows(
        Exception.class,
        () -> getEntity(entityId),
        "Deleted entity should not be retrievable without include=deleted");

    // TODO: Add restore functionality once SDK supports it
    // For now, this test verifies soft delete works correctly
  }

  // ===================================================================
  // OWNER VALIDATION TESTS
  // ===================================================================

  /**
   * Test: PATCH entity with non-existent owner should fail
   * Equivalent to: post_entityWithNonExistentOwner_4xx in EntityResourceTest
   */
  @Test
  void patch_entityWithNonExistentOwner_4xx(TestNamespace ns) {
    if (!supportsOwners) return;

    // Create entity without owner
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Try to set a non-existent owner
    org.openmetadata.schema.type.EntityReference nonExistentOwner =
        new org.openmetadata.schema.type.EntityReference()
            .withId(java.util.UUID.randomUUID())
            .withType("user")
            .withName("nonexistent@example.com");

    created.setOwners(List.of(nonExistentOwner));

    String entityId = created.getId().toString();
    assertThrows(
        Exception.class,
        () -> patchEntity(entityId, created),
        "Patching entity with non-existent owner should fail");
  }

  /**
   * Test: PATCH to update owner
   * Equivalent to: patch_entityUpdateOwner_200 in EntityResourceTest
   */
  @Test
  void patch_entityUpdateOwner_200(TestNamespace ns) {
    if (!supportsOwners || !supportsPatch) return;

    // Create entity without owner
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Verify no owner initially
    T fetched = getEntityWithFields(created.getId().toString(), "owners");
    assertTrue(
        fetched.getOwners() == null || fetched.getOwners().isEmpty(),
        "Entity should not have owner initially");

    // Get ingestion-bot user to use as owner (bots are auto-created)
    org.openmetadata.schema.entity.teams.User botUser = Users.getByName("ingestion-bot");
    org.openmetadata.schema.type.EntityReference ownerRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(botUser.getId())
            .withType("user")
            .withName(botUser.getName())
            .withFullyQualifiedName(botUser.getFullyQualifiedName());

    // Set owner via PATCH
    fetched.setOwners(List.of(ownerRef));
    T updated = patchEntity(fetched.getId().toString(), fetched);

    // Verify owner was set
    T updatedFetched = getEntityWithFields(updated.getId().toString(), "owners");
    assertNotNull(updatedFetched.getOwners(), "Entity should have owners");
    assertEquals(1, updatedFetched.getOwners().size(), "Entity should have 1 owner");
    assertEquals(
        botUser.getId(),
        updatedFetched.getOwners().get(0).getId(),
        "Owner should be ingestion-bot user");
  }

  /**
   * Test: Change owner from one user to another
   * Equivalent to: put_entityUpdateOwner_200 in EntityResourceTest
   */
  @Test
  void patch_entityChangeOwner_200(TestNamespace ns) {
    if (!supportsOwners || !supportsPatch) return;

    // Create entity and set initial owner
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Set ingestion-bot as owner
    org.openmetadata.schema.entity.teams.User botUser = Users.getByName("ingestion-bot");
    org.openmetadata.schema.type.EntityReference botRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(botUser.getId())
            .withType("user")
            .withName(botUser.getName());

    created.setOwners(List.of(botRef));
    T withOwner = patchEntity(created.getId().toString(), created);

    // Verify owner was set
    T fetchedWithOwner = getEntityWithFields(withOwner.getId().toString(), "owners");
    assertNotNull(fetchedWithOwner.getOwners());
    assertEquals(1, fetchedWithOwner.getOwners().size());
    assertEquals(botUser.getId(), fetchedWithOwner.getOwners().get(0).getId());

    // Clear the owner by setting empty list
    fetchedWithOwner.setOwners(new ArrayList<>());
    T withoutOwner = patchEntity(fetchedWithOwner.getId().toString(), fetchedWithOwner);

    // Verify owner was removed
    T finalFetch = getEntityWithFields(withoutOwner.getId().toString(), "owners");
    assertTrue(
        finalFetch.getOwners() == null || finalFetch.getOwners().isEmpty(),
        "Owner should be removed");
  }

  // ===================================================================
  // VERSION-BASED CONCURRENCY TESTS (Using entity version for optimistic locking)
  // ===================================================================

  /**
   * Test: Concurrent updates - version acts as optimistic lock
   * Equivalent to: patch_concurrent_updates_with_etag in EntityResourceTest
   *
   * In OpenMetadata, entity version acts as optimistic lock. When two concurrent updates
   * happen, the second one will see a different version and should handle it appropriately.
   */
  @Test
  void patch_concurrentUpdates_optimisticLock(TestNamespace ns) {
    if (!supportsPatch) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    Double originalVersion = created.getVersion();

    // First update - change description to unique value
    String firstDesc = "Concurrent update 1 - " + System.currentTimeMillis();
    created.setDescription(firstDesc);
    T afterFirst = patchEntity(created.getId().toString(), created);

    // Version should have changed
    assertTrue(
        afterFirst.getVersion() > originalVersion,
        "Version should increment after update, was: " + afterFirst.getVersion());

    // Get fresh copy for second update
    T fresh = getEntity(afterFirst.getId().toString());

    // Second update with different description
    String secondDesc = "Concurrent update 2 - " + System.currentTimeMillis();
    fresh.setDescription(secondDesc);
    T afterSecond = patchEntity(fresh.getId().toString(), fresh);

    // Version should increment or stay same (depending on consolidation window)
    assertTrue(
        afterSecond.getVersion() >= afterFirst.getVersion(),
        "Version should be >= previous, got: "
            + afterSecond.getVersion()
            + " vs "
            + afterFirst.getVersion());
  }

  /**
   * Test: Version changes on updates (optimistic locking verification)
   * Equivalent to: patch_with_valid_etag / patch_with_stale_etag concepts
   *
   * This test verifies that version tracking works correctly for detecting changes.
   */
  @Test
  void patch_versionTracking_200(TestNamespace ns) {
    if (!supportsPatch) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    assertEquals(0.1, created.getVersion(), 0.001, "Initial version should be 0.1");

    // Update 1: Change description
    created.setDescription("Version tracking update 1 - " + System.currentTimeMillis());
    T updated1 = patchEntity(created.getId().toString(), created);
    assertEquals(0.2, updated1.getVersion(), 0.001, "Version should be 0.2 after first update");

    // Update 2: Change description to a DIFFERENT value
    updated1.setDescription("Version tracking update 2 - " + System.currentTimeMillis());
    T updated2 = patchEntity(updated1.getId().toString(), updated1);

    // Version should increment (will be 0.3 if it's a new change, or same if consolidated)
    assertTrue(
        updated2.getVersion() >= 0.2,
        "Version should be at least 0.2, got: " + updated2.getVersion());

    // Fetch entity and verify version persisted
    T fetched = getEntity(updated2.getId().toString());
    assertEquals(
        updated2.getVersion(),
        fetched.getVersion(),
        0.001,
        "Fetched version should match updated version");
  }

  // ===================================================================
  // PHASE 2: Advanced GET Operations
  // ===================================================================

  @Test
  void get_entityWithDifferentFields_200_OK(TestNamespace ns) {
    if (!supportsOwners && !supportsTags) {
      return; // Skip if entity doesn't support fields testing
    }

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // GET with no fields - should return basic fields only
    T entityWithoutFields = getEntity(entity.getId().toString());
    assertNotNull(entityWithoutFields);
    assertNotNull(entityWithoutFields.getId());

    // GET with specific fields - owners, tags (if supported)
    String fields = buildFieldsParam();
    if (fields != null && !fields.isEmpty()) {
      T entityWithFields = getEntityWithFields(entity.getId().toString(), fields);
      assertNotNull(entityWithFields);
      assertNotNull(entityWithFields.getId());

      // Validate by name as well
      T entityByName = getEntityByNameWithFields(entity.getFullyQualifiedName(), fields);
      assertNotNull(entityByName);
      assertEquals(entity.getId(), entityByName.getId());
    }
  }

  @Test
  void get_entityIncludeDeleted_200(TestNamespace ns) {
    Assumptions.assumeTrue(supportsSoftDelete, "Entity does not support soft delete");
    Assumptions.assumeTrue(supportsIncludeDeleted, "Entity does not support include=deleted");

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    String entityId = entity.getId().toString();

    // Soft delete the entity
    deleteEntity(entityId);

    // GET without include=deleted should throw exception
    assertThrows(
        Exception.class,
        () -> getEntity(entityId),
        "Getting deleted entity without include=deleted should fail");

    // GET with include=deleted should succeed
    T deletedEntity = getEntityIncludeDeleted(entityId);
    assertNotNull(deletedEntity);
    assertEquals(entity.getId(), deletedEntity.getId());
    assertTrue(deletedEntity.getDeleted(), "Entity should be marked as deleted");
  }

  // ===================================================================
  // HELPER METHODS FOR PHASE 2
  // ===================================================================

  /**
   * Build fields parameter based on entity capabilities.
   */
  private String buildFieldsParam() {
    List<String> fields = new ArrayList<>();
    if (supportsOwners) fields.add("owners");
    if (supportsTags) fields.add("tags");
    if (supportsFollowers) fields.add("followers");
    return String.join(",", fields);
  }

  /**
   * Get entity with specific fields parameter.
   * Subclasses can override if they have a different mechanism.
   */
  protected T getEntityWithFields(String id, String fields) {
    // Default implementation - subclasses should override
    return getEntity(id);
  }

  /**
   * Get entity by name with specific fields parameter.
   * Subclasses can override if they have a different mechanism.
   */
  protected T getEntityByNameWithFields(String fqn, String fields) {
    // Default implementation - subclasses should override
    return getEntityByName(fqn);
  }

  /**
   * Get entity with include=deleted parameter.
   * Subclasses can override if they have a different mechanism.
   */
  protected T getEntityIncludeDeleted(String id) {
    // Default implementation - subclasses should override
    throw new UnsupportedOperationException("Include deleted not implemented");
  }

  // ===================================================================
  // PHASE 3: Pagination & List Operations
  // ===================================================================

  /**
   * Pagination test is disabled in BaseEntityIT to avoid conflicts with parallel tests.
   * Comprehensive pagination testing is done in PaginationIT which runs in isolation.
   * This test just verifies basic list functionality works.
   */
  @Test
  void get_entityListWithPagination_200(TestNamespace ns) {
    // Create a few entities
    for (int i = 0; i < 3; i++) {
      K createRequest = createRequest(ns.prefix("list" + i), ns);
      createEntity(createRequest);
    }

    Awaitility.await("Wait for entities to be listable")
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(1))
        .atMost(Duration.ofSeconds(15))
        .untilAsserted(
            () -> {
              org.openmetadata.sdk.models.ListParams params =
                  new org.openmetadata.sdk.models.ListParams();
              params.setLimit(10);
              org.openmetadata.sdk.models.ListResponse<T> response = listEntities(params);

              assertNotNull(response, "List response should not be null");
              assertNotNull(response.getData(), "List data should not be null");
              assertTrue(response.getData().size() > 0, "Should have entities in list");
            });
  }

  @Test
  void get_entityListWithInvalidLimit_4xx(TestNamespace ns) {

    // Test with invalid limit (negative)
    org.openmetadata.sdk.models.ListParams params = new org.openmetadata.sdk.models.ListParams();
    params.setLimit(-1);

    assertThrows(
        Exception.class, () -> listEntities(params), "Listing with negative limit should fail");

    // Test with limit > max allowed (usually 1000000)
    params.setLimit(2000000);
    assertThrows(
        Exception.class, () -> listEntities(params), "Listing with excessive limit should fail");
  }

  @Test
  void get_entityListWithInvalidPaginationCursors_4xx(TestNamespace ns) {

    // Test with invalid after cursor
    org.openmetadata.sdk.models.ListParams paramsAfter =
        new org.openmetadata.sdk.models.ListParams();
    paramsAfter.setAfter("invalid-cursor-string");

    assertThrows(
        Exception.class,
        () -> listEntities(paramsAfter),
        "Listing with invalid after cursor should fail");

    // Test with invalid before cursor
    org.openmetadata.sdk.models.ListParams paramsBefore =
        new org.openmetadata.sdk.models.ListParams();
    paramsBefore.setBefore("invalid-cursor-string");

    assertThrows(
        Exception.class,
        () -> listEntities(paramsBefore),
        "Listing with invalid before cursor should fail");
  }

  @Test
  void get_entityWithInvalidFields_4xx(TestNamespace ns) {
    if (!supportsFieldsQueryParam) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Test GET by ID with invalid field
    String invalidField = "invalidFieldName123";
    assertThrows(
        Exception.class,
        () -> getEntityWithFields(entity.getId().toString(), invalidField),
        "GET with invalid field should fail");

    // Test GET by name with invalid field
    assertThrows(
        Exception.class,
        () -> getEntityByNameWithFields(entity.getFullyQualifiedName(), invalidField),
        "GET by name with invalid field should fail");
  }

  // ===================================================================
  // HELPER METHODS FOR PHASE 3
  // ===================================================================

  /**
   * List entities using the SDK. Subclasses must implement this.
   */
  protected abstract org.openmetadata.sdk.models.ListResponse<T> listEntities(
      org.openmetadata.sdk.models.ListParams params);

  // ===================================================================
  // PHASE 3: TAGS OPERATIONS
  // ===================================================================

  @Test
  void test_entityWithInvalidTag(TestNamespace ns) {
    if (!supportsTags) {
      return;
    }

    // Create an entity first
    K validCreate = createRequest(ns.prefix("valid_entity"), ns);
    T entity = createEntity(validCreate);

    // Try to patch with invalid tag - this should fail
    TagLabel invalidTag = new TagLabel().withTagFQN("invalidTag");
    entity.setTags(List.of(invalidTag));
    String entityId = entity.getId().toString();
    assertThrows(
        Exception.class,
        () -> patchEntity(entityId, entity),
        "Patching entity with invalid tag should fail");
  }

  @Test
  void test_tagUpdateOptimization_PUT(TestNamespace ns) {
    if (!supportsTags) {
      return;
    }

    // Create test-specific classification and tags to avoid deadlocks with parallel tests
    OpenMetadataClient client = SdkClients.adminClient();
    String classificationName = ns.prefix("TagPutClassification");

    org.openmetadata.schema.api.classification.CreateClassification createClassification =
        new org.openmetadata.schema.api.classification.CreateClassification()
            .withName(classificationName)
            .withDescription("Classification for tag PUT test");
    client
        .getHttpClient()
        .execute(
            HttpMethod.PUT,
            "/v1/classifications",
            createClassification,
            org.openmetadata.schema.entity.classification.Classification.class);

    // Create test-specific tags
    String tag1Name = "PutTag1";
    String tag2Name = "PutTag2";
    String tag3Name = "PutTag3";

    for (String tagName : List.of(tag1Name, tag2Name, tag3Name)) {
      org.openmetadata.schema.api.classification.CreateTag createTag =
          new org.openmetadata.schema.api.classification.CreateTag()
              .withName(tagName)
              .withDescription("Tag for PUT test")
              .withClassification(classificationName);
      client
          .getHttpClient()
          .execute(
              HttpMethod.PUT,
              "/v1/tags",
              createTag,
              org.openmetadata.schema.entity.classification.Tag.class);
    }

    TagLabel tag1 = new TagLabel().withTagFQN(classificationName + "." + tag1Name);
    TagLabel tag2 = new TagLabel().withTagFQN(classificationName + "." + tag2Name);

    // Create entity first without tags, then patch to add tags
    K create = createRequest(ns.prefix("tag_put_test"), ns);
    T entity = createEntity(create);

    // Add tags via PATCH
    entity.setTags(List.of(tag1, tag2));
    T tagged = patchEntity(entity.getId().toString(), entity);

    // Verify initial tags
    T fetched = getEntityWithFields(tagged.getId().toString(), "tags");
    assertEquals(2, fetched.getTags().size());
    assertTagsContain(fetched.getTags(), List.of(tag1, tag2));

    // PATCH with one new tag - SDK PATCH merges tags, not replaces
    TagLabel tag3 = new TagLabel().withTagFQN(classificationName + "." + tag3Name);
    fetched.setTags(List.of(tag1, tag2, tag3));
    T updated = patchEntity(fetched.getId().toString(), fetched);

    // Verify all three tags are present (PATCH merges tags)
    T updatedFetched = getEntityWithFields(updated.getId().toString(), "tags");
    assertNotNull(updatedFetched.getTags());
    assertEquals(3, updatedFetched.getTags().size());
    assertTagsContain(updatedFetched.getTags(), List.of(tag1, tag2, tag3));
  }

  @Test
  void test_tagUpdateOptimization_PATCH(TestNamespace ns) {
    if (!supportsTags) {
      return;
    }

    // Create test-specific classification and tags to avoid deadlocks with parallel tests
    OpenMetadataClient client = SdkClients.adminClient();
    String classificationName = ns.prefix("TagPatchClassification");

    org.openmetadata.schema.api.classification.CreateClassification createClassification =
        new org.openmetadata.schema.api.classification.CreateClassification()
            .withName(classificationName)
            .withDescription("Classification for tag PATCH test");
    client
        .getHttpClient()
        .execute(
            HttpMethod.PUT,
            "/v1/classifications",
            createClassification,
            org.openmetadata.schema.entity.classification.Classification.class);

    // Create test-specific tags
    String tag1Name = "Tag1";
    String tag2Name = "Tag2";
    String tag3Name = "Tag3";
    String tag4Name = "Tag4";

    for (String tagName : List.of(tag1Name, tag2Name, tag3Name, tag4Name)) {
      org.openmetadata.schema.api.classification.CreateTag createTag =
          new org.openmetadata.schema.api.classification.CreateTag()
              .withName(tagName)
              .withDescription("Tag for PATCH test")
              .withClassification(classificationName);
      client
          .getHttpClient()
          .execute(
              HttpMethod.PUT,
              "/v1/tags",
              createTag,
              org.openmetadata.schema.entity.classification.Tag.class);
    }

    TagLabel tag1 = new TagLabel().withTagFQN(classificationName + "." + tag1Name);
    TagLabel tag2 = new TagLabel().withTagFQN(classificationName + "." + tag2Name);

    // Create entity first without tags
    K create = createRequest(ns.prefix("tag_patch_test"), ns);
    T entity = createEntity(create);

    // Add initial tags via PATCH
    entity.setTags(List.of(tag1, tag2));
    T tagged = patchEntity(entity.getId().toString(), entity);

    T fetched = getEntityWithFields(tagged.getId().toString(), "tags");
    assertEquals(2, fetched.getTags().size());
    assertTagsContain(fetched.getTags(), List.of(tag1, tag2));

    // PATCH with different tags - SDK PATCH merges, so we need all 4 tags
    TagLabel tag3 = new TagLabel().withTagFQN(classificationName + "." + tag3Name);
    TagLabel tag4 = new TagLabel().withTagFQN(classificationName + "." + tag4Name);

    fetched.setTags(List.of(tag1, tag2, tag3, tag4));
    T patched = patchEntity(fetched.getId().toString(), fetched);

    // Verify all four tags are present (PATCH merges tags)
    T patchedFetched = getEntityWithFields(patched.getId().toString(), "tags");
    assertEquals(4, patchedFetched.getTags().size());
    assertTagsContain(patchedFetched.getTags(), List.of(tag1, tag2, tag3, tag4));
  }

  @Test
  void test_tagUpdateOptimization_LargeScale(TestNamespace ns) {
    if (!supportsTags) {
      return;
    }

    // Create test-specific classification and tags to avoid deadlocks with parallel tests
    OpenMetadataClient client = SdkClients.adminClient();
    String classificationName = ns.prefix("TagLargeClassification");

    org.openmetadata.schema.api.classification.CreateClassification createClassification =
        new org.openmetadata.schema.api.classification.CreateClassification()
            .withName(classificationName)
            .withDescription("Classification for large scale tag test");
    client
        .getHttpClient()
        .execute(
            HttpMethod.PUT,
            "/v1/classifications",
            createClassification,
            org.openmetadata.schema.entity.classification.Classification.class);

    // Create test-specific tags
    String tag1Name = "LargeTag1";
    String tag2Name = "LargeTag2";
    String tag3Name = "LargeTag3";
    String tag4Name = "LargeTag4";

    for (String tagName : List.of(tag1Name, tag2Name, tag3Name, tag4Name)) {
      org.openmetadata.schema.api.classification.CreateTag createTag =
          new org.openmetadata.schema.api.classification.CreateTag()
              .withName(tagName)
              .withDescription("Tag for large scale test")
              .withClassification(classificationName);
      client
          .getHttpClient()
          .execute(
              HttpMethod.PUT,
              "/v1/tags",
              createTag,
              org.openmetadata.schema.entity.classification.Tag.class);
    }

    // Create entity first without tags
    K create = createRequest(ns.prefix("tag_large_test"), ns);
    T entity = createEntity(create);

    // Add initial tags via PATCH
    List<TagLabel> initialTags = new ArrayList<>();
    initialTags.add(
        new TagLabel()
            .withTagFQN(classificationName + "." + tag1Name)
            .withLabelType(TagLabel.LabelType.MANUAL)
            .withState(TagLabel.State.CONFIRMED));
    initialTags.add(
        new TagLabel()
            .withTagFQN(classificationName + "." + tag2Name)
            .withLabelType(TagLabel.LabelType.MANUAL)
            .withState(TagLabel.State.CONFIRMED));

    entity.setTags(initialTags);
    T tagged = patchEntity(entity.getId().toString(), entity);

    // Verify we have 2 unique tags
    T fetched = getEntityWithFields(tagged.getId().toString(), "tags");
    assertEquals(2, fetched.getTags().size());

    // Add more unique tags via PATCH
    List<TagLabel> additionalTags = new ArrayList<>();
    additionalTags.add(
        new TagLabel()
            .withTagFQN(classificationName + "." + tag3Name)
            .withLabelType(TagLabel.LabelType.MANUAL)
            .withState(TagLabel.State.CONFIRMED));
    additionalTags.add(
        new TagLabel()
            .withTagFQN(classificationName + "." + tag4Name)
            .withLabelType(TagLabel.LabelType.MANUAL)
            .withState(TagLabel.State.CONFIRMED));

    List<TagLabel> allTags = new ArrayList<>(initialTags);
    allTags.addAll(additionalTags);

    fetched.setTags(allTags);
    T updated = patchEntity(fetched.getId().toString(), fetched);

    T updatedFetched = getEntityWithFields(updated.getId().toString(), "tags");
    assertEquals(4, updatedFetched.getTags().size());
    assertTagsContain(updatedFetched.getTags(), initialTags);
    assertTagsContain(updatedFetched.getTags(), additionalTags);
  }

  private void assertTagsContain(List<TagLabel> tags, List<TagLabel> expectedTags) {
    for (TagLabel expected : expectedTags) {
      assertTrue(
          tags.stream().anyMatch(tag -> tag.getTagFQN().equals(expected.getTagFQN())),
          "Tags should contain: " + expected.getTagFQN());
    }
  }

  private void assertTagsDoNotContain(List<TagLabel> tags, List<TagLabel> unexpectedTags) {
    for (TagLabel unexpected : unexpectedTags) {
      assertFalse(
          tags.stream().anyMatch(tag -> tag.getTagFQN().equals(unexpected.getTagFQN())),
          "Tags should not contain: " + unexpected.getTagFQN());
    }
  }

  // ===================================================================
  // VERSION HISTORY TESTS
  // ===================================================================

  @Test
  void get_entityVersionHistory_200(TestNamespace ns) {
    if (!supportsVersionHistory || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    assertEquals(0.1, created.getVersion(), 0.001);

    created.setDescription("Version 1 update - " + System.currentTimeMillis());
    T v2 = patchEntity(created.getId().toString(), created);
    assertEquals(0.2, v2.getVersion(), 0.001);

    v2.setDescription("Version 2 update - " + System.currentTimeMillis());
    T v3 = patchEntity(v2.getId().toString(), v2);
    assertTrue(v3.getVersion() >= 0.2, "Version should be at least 0.2");

    org.openmetadata.schema.type.EntityHistory history = getVersionHistory(created.getId());
    assertNotNull(history, "Version history should not be null");
    assertNotNull(history.getVersions(), "Versions list should not be null");
    assertTrue(history.getVersions().size() >= 2, "Should have at least 2 versions");
  }

  @Test
  void get_entityVersionHistory_paginated_200(TestNamespace ns) {
    if (!supportsVersionHistory || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    String entityId = created.getId().toString();
    String patchPath = getResourcePath() + entityId;
    RequestOptions patchOptions =
        RequestOptions.builder().header("Content-Type", "application/json-patch+json").build();

    OpenMetadataClient admin = SdkClients.adminClient();
    OpenMetadataClient user1 = SdkClients.user1Client();

    for (int i = 1; i <= 10; i++) {
      OpenMetadataClient patchClient = i % 2 == 0 ? user1 : admin;
      String op = i == 1 ? "add" : "replace";
      String value = (i % 2 == 0 ? "User1" : "Admin") + " patch " + i;
      try {
        patchClient
            .getHttpClient()
            .executeForString(
                HttpMethod.PATCH,
                patchPath,
                String.format(
                    "[{\"op\":\"%s\",\"path\":\"/description\",\"value\":\"%s\"}]", op, value),
                patchOptions);
      } catch (Exception e) {
        Assumptions.assumeTrue(
            false, "Skipping: user1 cannot patch this entity type - " + e.getMessage());
      }
    }

    org.openmetadata.schema.type.EntityHistory allVersions = getVersionHistory(created.getId());
    int totalVersions = allVersions.getVersions().size();
    assertTrue(totalVersions >= 11, "Should have at least 11 versions after repeated patches");

    org.openmetadata.schema.type.EntityHistory page1 =
        getVersionHistoryPaginated(created.getId(), 1, 0);
    assertNotNull(page1.getPaging(), "Paging metadata should be present");
    assertEquals(1, (int) page1.getPaging().getLimit());
    assertEquals(0, (int) page1.getPaging().getOffset());
    assertEquals(totalVersions, (int) page1.getPaging().getTotal());
    assertEquals(1, page1.getVersions().size());

    org.openmetadata.schema.type.EntityHistory page2 =
        getVersionHistoryPaginated(created.getId(), 1, 1);
    assertNotNull(page2.getPaging());
    assertEquals(1, (int) page2.getPaging().getLimit());
    assertEquals(1, (int) page2.getPaging().getOffset());
    assertEquals(totalVersions, (int) page2.getPaging().getTotal());
    assertEquals(1, page2.getVersions().size());

    double page1Version = versionOfHistoryEntry(page1.getVersions().get(0));
    double page2Version = versionOfHistoryEntry(page2.getVersions().get(0));
    assertTrue(page1Version > 1.0, "First page should contain a version after the 1.0 boundary");
    assertTrue(page1Version > page2Version, "Paginated results should remain newest-first");

    org.openmetadata.schema.type.EntityHistory unpaginated = getVersionHistory(created.getId());
    assertNotNull(unpaginated.getVersions());
    assertEquals(totalVersions, unpaginated.getVersions().size());
  }

  @Test
  void get_specificVersion_200(TestNamespace ns) {
    if (!supportsVersionHistory || !supportsGetByVersion || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    String originalDesc = created.getDescription();

    created.setDescription("Updated description for version test");
    T updated = patchEntity(created.getId().toString(), created);

    T version01 = getVersion(created.getId(), 0.1);
    assertNotNull(version01, "Version 0.1 should be retrievable");
    assertEquals(0.1, version01.getVersion(), 0.001);
    if (originalDesc != null) {
      assertEquals(originalDesc, version01.getDescription());
    }
  }

  @Test
  void get_entityVersionHistory_fieldChanged_200(TestNamespace ns) {
    if (!supportsVersionHistory || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("Description change for fieldChanged test");
    patchEntity(created.getId().toString(), created);

    org.openmetadata.schema.type.EntityHistory filtered =
        getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "description");
    assertNotNull(filtered, "Filtered version history should not be null");
    assertNotNull(filtered.getPaging(), "Paging metadata should be present");
    assertTrue(
        filtered.getVersions().size() >= 1,
        "Should have at least 1 version with description change");
    assertTrue(filtered.getPaging().getTotal() >= 1, "Total should reflect filtered count");

    org.openmetadata.schema.type.EntityHistory filteredWithDefaultLimit =
        getVersionHistoryWithFieldChanged(created.getId(), 0, 0, "description");
    assertNotNull(filteredWithDefaultLimit);
    assertNotNull(filteredWithDefaultLimit.getPaging());
    assertEquals(
        100,
        (int) filteredWithDefaultLimit.getPaging().getLimit(),
        "Filtered history should use a bounded default page size when limit is omitted");
    assertEquals(
        filtered.getPaging().getTotal(),
        filteredWithDefaultLimit.getPaging().getTotal(),
        "Default pagination should preserve the same filtered total");

    org.openmetadata.schema.type.EntityHistory substringNoMatch =
        getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "script");
    assertNotNull(substringNoMatch);
    assertEquals(
        0,
        substringNoMatch.getVersions().size(),
        "Substring matches should not be treated as field-name matches");
    assertEquals(0, (int) substringNoMatch.getPaging().getTotal());

    org.openmetadata.schema.type.EntityHistory nestedPathNoMatch =
        getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "nested.description");
    assertNotNull(nestedPathNoMatch);
    assertEquals(
        0,
        nestedPathNoMatch.getVersions().size(),
        "Exact field filters should not match nested paths for a top-level description change");
    assertEquals(0, (int) nestedPathNoMatch.getPaging().getTotal());

    org.openmetadata.schema.type.EntityHistory noMatch =
        getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "nonExistentField_xyz_12345");
    assertNotNull(noMatch);
    assertEquals(0, noMatch.getVersions().size(), "No versions should match a bogus field name");
    assertEquals(0, (int) noMatch.getPaging().getTotal());
  }

  @Test
  void get_entityVersionHistory_fieldChanged_ignoresWrongExtensionPrefix(TestNamespace ns) {
    if (!supportsVersionHistory || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("Description change for extension prefix test");
    patchEntity(created.getId().toString(), created);

    org.openmetadata.schema.type.EntityHistory baseline =
        getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "description");
    assertNotNull(baseline);
    assertNotNull(baseline.getPaging());

    String rogueExtension = "rogue.version.999.0";
    try {
      Entity.getCollectionDAO()
          .entityExtensionDAO()
          .insertVersionExtension(
              created.getId(),
              rogueExtension,
              getEntityType(),
              JsonUtils.pojoToJson(created),
              999.0,
              JsonUtils.pojoToJson(List.of("description")));

      org.openmetadata.schema.type.EntityHistory filtered =
          getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "description");

      assertNotNull(filtered);
      assertNotNull(filtered.getPaging());
      assertEquals(
          baseline.getPaging().getTotal(),
          filtered.getPaging().getTotal(),
          "Filtered history should ignore extension rows outside the entity version prefix");
      assertEquals(
          baseline.getVersions().size(),
          filtered.getVersions().size(),
          "Filtered history should not include rows from a rogue extension prefix");
    } finally {
      Entity.getCollectionDAO().entityExtensionDAO().delete(created.getId(), rogueExtension);
    }
  }

  @Test
  void get_entityVersionHistory_paginated_ordersNullVersionNumRowsByVersion(TestNamespace ns) {
    if (!supportsVersionHistory) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    List<Double> insertedVersions = List.of(9.0, 10.0, 11.0);

    try {
      for (Double version : insertedVersions) {
        ObjectNode historicalVersion = (ObjectNode) JsonUtils.pojoToJsonNode(created);
        historicalVersion.put("version", version);
        Entity.getCollectionDAO()
            .entityExtensionDAO()
            .insert(
                created.getId(),
                EntityUtil.getVersionExtension(getEntityType(), version),
                getEntityType(),
                JsonUtils.pojoToJson(historicalVersion));
      }

      org.openmetadata.schema.type.EntityHistory paginatedHistory =
          getVersionHistoryPaginated(created.getId(), 3, 1);

      assertNotNull(paginatedHistory);
      assertNotNull(paginatedHistory.getPaging());
      assertEquals(4, (int) paginatedHistory.getPaging().getTotal());
      assertEquals(3, paginatedHistory.getVersions().size());
      assertEquals(11.0, versionOfHistoryEntry(paginatedHistory.getVersions().get(0)), 0.001);
      assertEquals(10.0, versionOfHistoryEntry(paginatedHistory.getVersions().get(1)), 0.001);
      assertEquals(9.0, versionOfHistoryEntry(paginatedHistory.getVersions().get(2)), 0.001);
    } finally {
      for (Double version : insertedVersions) {
        Entity.getCollectionDAO()
            .entityExtensionDAO()
            .delete(created.getId(), EntityUtil.getVersionExtension(getEntityType(), version));
      }
    }
  }

  @Test
  void get_entityVersionHistory_fieldChanged_matchesNullChangedFieldKeysRows(TestNamespace ns) {
    if (!supportsVersionHistory || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("Description change for null changedFieldKeys test");
    patchEntity(created.getId().toString(), created);

    org.openmetadata.schema.type.EntityHistory baseline =
        getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "description");
    assertNotNull(baseline);
    assertNotNull(baseline.getPaging());

    double legacyVersion = 77.0;
    String legacyExtension = EntityUtil.getVersionExtension(getEntityType(), legacyVersion);

    try {
      ObjectNode historicalVersion = (ObjectNode) JsonUtils.pojoToJsonNode(created);
      historicalVersion.put("version", legacyVersion);
      ObjectNode changeDescription = historicalVersion.putObject("changeDescription");
      ArrayNode fieldsUpdated = changeDescription.putArray("fieldsUpdated");
      fieldsUpdated.addObject().put("name", "nested.description");

      Entity.getCollectionDAO()
          .entityExtensionDAO()
          .insert(
              created.getId(),
              legacyExtension,
              getEntityType(),
              JsonUtils.pojoToJson(historicalVersion));

      org.openmetadata.schema.type.EntityHistory filtered =
          getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "description");

      assertNotNull(filtered);
      assertNotNull(filtered.getPaging());
      assertEquals(
          baseline.getPaging().getTotal(),
          (int) filtered.getPaging().getTotal(),
          "Exact field filters should not match nested legacy rows without changedFieldKeys");
      assertFalse(
          filtered.getVersions().stream()
              .anyMatch(
                  version -> Math.abs(versionOfHistoryEntry(version) - legacyVersion) < 0.001),
          "Top-level description filters should not include nested legacy rows");

      org.openmetadata.schema.type.EntityHistory nestedFiltered =
          getVersionHistoryWithFieldChanged(created.getId(), 100, 0, "nested.description");

      assertNotNull(nestedFiltered);
      assertNotNull(nestedFiltered.getPaging());
      assertEquals(
          1,
          (int) nestedFiltered.getPaging().getTotal(),
          "Exact nested field filters should still match legacy rows without changedFieldKeys");
      assertTrue(
          nestedFiltered.getVersions().stream()
              .anyMatch(
                  version -> Math.abs(versionOfHistoryEntry(version) - legacyVersion) < 0.001),
          "Nested field filters should include the legacy row without changedFieldKeys");
    } finally {
      Entity.getCollectionDAO().entityExtensionDAO().delete(created.getId(), legacyExtension);
    }
  }

  private double versionOfHistoryEntry(Object versionEntry) {
    JsonNode node =
        versionEntry instanceof String versionJson
            ? JsonUtils.readValue(versionJson, JsonNode.class)
            : JsonUtils.pojoToJsonNode(versionEntry);
    assertTrue(node.hasNonNull("version"), "Version history entry should include version");
    return node.get("version").asDouble();
  }

  protected org.openmetadata.schema.type.EntityHistory getVersionHistory(UUID id) {
    throw new UnsupportedOperationException(
        "Version history not implemented - override in subclass");
  }

  protected org.openmetadata.schema.type.EntityHistory getVersionHistoryPaginated(
      UUID id, int limit, int offset) {
    throw new UnsupportedOperationException(
        "Paginated version history not implemented - override in subclass");
  }

  protected org.openmetadata.schema.type.EntityHistory getVersionHistoryWithFieldChanged(
      UUID id, int limit, int offset, String fieldChanged) {
    throw new UnsupportedOperationException(
        "fieldChanged version history not implemented - override in subclass");
  }

  protected T getVersion(UUID id, Double version) {
    throw new UnsupportedOperationException("Get version not implemented - override in subclass");
  }

  // ===================================================================
  // ADMIN DELETE TESTS
  // ===================================================================

  @Test
  void delete_entityAsAdmin_hardDelete_200(TestNamespace ns) {

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    String entityId = created.getId().toString();

    hardDeleteEntity(entityId);

    assertThrows(
        Exception.class,
        () -> getEntity(entityId),
        "Hard deleted entity should not be retrievable");

    if (supportsSoftDelete) {
      assertThrows(
          Exception.class,
          () -> getEntityIncludeDeleted(entityId),
          "Hard deleted entity should not be retrievable even with include=deleted");
    }
  }

  // ===================================================================
  // DESCRIPTION VALIDATION TESTS
  // ===================================================================

  @Test
  void put_entityEmptyDescriptionUpdate_200(TestNamespace ns) {
    if (!supportsEmptyDescription || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("");
    T updated = patchEntity(created.getId().toString(), created);

    assertEquals("", updated.getDescription(), "Description should be empty string");
  }

  /**
   * Test: Entity creation fails if description is required but missing
   * Equivalent to: post_entityWithMissingDescription_400 in EntityResourceTest
   */
  @Test
  void post_entityWithMissingDescription_400(TestNamespace ns) {
    if (supportsEmptyDescription) {
      return; // Skip if entity allows empty/null description
    }

    // Try to create entity with null description - should fail
    K createRequest = createRequest(ns.prefix("noDesc"), ns);
    // Note: Implementation depends on how createRequest handles description
    // Some entities may require description in createRequest
    assertThrows(
        Exception.class,
        () -> createEntity(createRequest),
        "Creating entity without description should fail when description is required");
  }

  /**
   * Test: PUT with null description updates correctly
   * Equivalent to: put_entityNullDescriptionUpdate_200 in EntityResourceTest
   */
  @Test
  void put_entityNullDescriptionUpdate_200(TestNamespace ns) {
    if (!supportsEmptyDescription || !supportsPatch) return;

    // Create entity with null description
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Update to a new description
    String newDesc = "Updated description from null";
    entity.setDescription(newDesc);
    T updated = patchEntity(entity.getId().toString(), entity);

    assertEquals(newDesc, updated.getDescription(), "Description should be updated");
  }

  /**
   * Test: PUT with non-empty description updates correctly
   * Equivalent to: put_entityNonEmptyDescriptionUpdate_200 in EntityResourceTest
   */
  @Test
  void put_entityNonEmptyDescriptionUpdate_200(TestNamespace ns) {
    if (!supportsPatch) return;

    // Create entity with initial description
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Set initial description if needed
    String initialDesc = "Initial description";
    entity.setDescription(initialDesc);
    T withDesc = patchEntity(entity.getId().toString(), entity);

    // Update to a different description
    String updatedDesc = "Updated description";
    withDesc.setDescription(updatedDesc);
    T updated = patchEntity(withDesc.getId().toString(), withDesc);

    assertEquals(updatedDesc, updated.getDescription(), "Description should be updated");
  }

  // ===================================================================
  // OWNER VALIDATION TESTS (Additional)
  // ===================================================================

  /**
   * Test: Creating entity with invalid owner type (no type specified)
   * Equivalent to: post_entityWithInvalidOwnerType_4xx in EntityResourceTest
   */
  @Test
  void post_entityWithInvalidOwnerType_4xx(TestNamespace ns) {
    if (!supportsOwners) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Try to set owner with no type specified - should fail
    org.openmetadata.schema.type.EntityReference invalidOwner =
        new org.openmetadata.schema.type.EntityReference().withId(testUser1().getId());
    // Type is not set

    entity.setOwners(List.of(invalidOwner));
    String entityId = entity.getId().toString();

    assertThrows(
        Exception.class,
        () -> patchEntity(entityId, entity),
        "Setting owner without type should fail");
  }

  /**
   * Test: Creating entity with non-existent owner
   * Equivalent to: post_entityWithNonExistentOwner_4xx in EntityResourceTest
   */
  @Test
  void post_entityWithNonExistentOwner_4xx(TestNamespace ns) {
    if (!supportsOwners) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Try to set non-existent owner
    org.openmetadata.schema.type.EntityReference nonExistentOwner =
        new org.openmetadata.schema.type.EntityReference()
            .withId(UUID.randomUUID())
            .withType("user");

    entity.setOwners(List.of(nonExistentOwner));
    String entityId = entity.getId().toString();

    assertThrows(
        Exception.class,
        () -> patchEntity(entityId, entity),
        "Setting non-existent owner should fail");
  }

  /**
   * Test: PATCH to set owner when entity has no owner
   * Equivalent to: patch_entityUpdateOwnerFromNull_200 in EntityResourceTest
   */
  @Test
  void patch_entityUpdateOwnerFromNull_200(TestNamespace ns) {
    if (!supportsOwners || !supportsPatch) return;

    // Create entity without owner
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Verify no owner initially
    T fetched = getEntityWithFields(entity.getId().toString(), "owners");
    assertTrue(
        fetched.getOwners() == null || fetched.getOwners().isEmpty(),
        "Entity should not have owner initially");

    // Set multiple owners
    org.openmetadata.schema.type.EntityReference owner1 =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser1().getId())
            .withType("user")
            .withName(testUser1().getName());

    org.openmetadata.schema.type.EntityReference owner2 =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser2().getId())
            .withType("user")
            .withName(testUser2().getName());

    fetched.setOwners(List.of(owner1, owner2));
    T updated = patchEntity(fetched.getId().toString(), fetched);

    // Verify owners were set
    T verify = getEntityWithFields(updated.getId().toString(), "owners");
    assertNotNull(verify.getOwners(), "Entity should have owners");
    assertEquals(2, verify.getOwners().size(), "Entity should have 2 owners");
  }

  // ===================================================================
  // FOLLOWER TESTS
  // ===================================================================

  @Test
  void put_addDeleteFollower_200(TestNamespace ns) {
    if (!supportsFollowers || !hasFollowerMethods()) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    addFollower(entity.getId(), testUser1().getId());

    T fetched = getEntityWithFields(entity.getId().toString(), "followers");
    assertNotNull(fetched.getFollowers(), "Entity should have followers");
    assertTrue(
        fetched.getFollowers().stream().anyMatch(f -> f.getId().equals(testUser1().getId())),
        "testUser1 should be a follower");

    addFollower(entity.getId(), testUser2().getId());

    T fetched2 = getEntityWithFields(entity.getId().toString(), "followers");
    assertEquals(2, fetched2.getFollowers().size(), "Entity should have 2 followers");

    deleteFollower(entity.getId(), testUser1().getId());

    T fetched3 = getEntityWithFields(entity.getId().toString(), "followers");
    assertEquals(1, fetched3.getFollowers().size(), "Entity should have 1 follower");
    assertFalse(
        fetched3.getFollowers().stream().anyMatch(f -> f.getId().equals(testUser1().getId())),
        "testUser1 should not be a follower anymore");
  }

  @Test
  void put_addFollowerDeleteEntity_200(TestNamespace ns) {
    if (!supportsFollowers || !supportsSoftDelete || !hasFollowerMethods()) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    addFollower(entity.getId(), testUser1().getId());

    T fetched = getEntityWithFields(entity.getId().toString(), "followers");
    assertNotNull(fetched.getFollowers());
    assertEquals(1, fetched.getFollowers().size());

    deleteEntity(entity.getId().toString());

    T deletedEntity = getEntityIncludeDeleted(entity.getId().toString());
    assertNotNull(deletedEntity);
    assertTrue(deletedEntity.getDeleted());
  }

  @Test
  void put_addDeleteInvalidFollower_4xx(TestNamespace ns) {
    if (!supportsFollowers || !hasFollowerMethods()) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    UUID nonExistentUserId = UUID.randomUUID();
    UUID entityId = entity.getId();

    assertThrows(
        Exception.class,
        () -> addFollower(entityId, nonExistentUserId),
        "Adding non-existent user as follower should fail");
  }

  // ===================================================================
  // FOLLOWER HELPER METHODS
  // ===================================================================

  /**
   * Check if follower methods are implemented. Subclasses should override and return true.
   */
  protected boolean hasFollowerMethods() {
    return false; // Default to false, subclasses enable when they implement
    // addFollower/deleteFollower
  }

  /**
   * Add a follower to an entity. Subclasses can override for entity-specific behavior.
   */
  protected void addFollower(UUID entityId, UUID userId) {
    throw new UnsupportedOperationException("addFollower not implemented - override in subclass");
  }

  /**
   * Delete a follower from an entity. Subclasses can override for entity-specific behavior.
   */
  protected void deleteFollower(UUID entityId, UUID userId) {
    throw new UnsupportedOperationException(
        "deleteFollower not implemented - override in subclass");
  }

  // ===================================================================
  // DELETE TESTS (Additional)
  // ===================================================================

  /**
   * Test: Delete entity by name
   * Equivalent to: post_delete_as_name_entity_as_admin_200 in EntityResourceTest
   */
  @Test
  void delete_entityByName_200(TestNamespace ns) {
    if (!supportsDeleteByName()) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Delete by name
    deleteEntityByName(entity.getFullyQualifiedName());

    // Verify entity is deleted
    String entityId = entity.getId().toString();
    assertThrows(
        Exception.class, () -> getEntity(entityId), "Deleted entity should not be retrievable");
  }

  /**
   * Check if delete by name is supported. Subclasses can override.
   */
  protected boolean supportsDeleteByName() {
    return false; // Default to false, subclasses enable
  }

  /**
   * Delete entity by name. Subclasses can override for entity-specific behavior.
   */
  protected void deleteEntityByName(String fqn) {
    throw new UnsupportedOperationException(
        "deleteEntityByName not implemented - override in subclass");
  }

  /**
   * Test: Delete non-existent entity returns 404
   * Equivalent to: delete_nonExistentEntity_404 in EntityResourceTest
   */
  @Test
  void delete_nonExistentEntity_404(TestNamespace ns) {

    // Try to delete non-existent entity
    String nonExistentId = UUID.randomUUID().toString();

    assertThrows(
        Exception.class,
        () -> deleteEntity(nonExistentId),
        "Deleting non-existent entity should fail with 404");
  }

  // ===================================================================
  // PATCH ATTRIBUTE TESTS (Additional)
  // ===================================================================

  /**
   * Test: Cannot undelete via PATCH (deleted attribute is disallowed)
   * Equivalent to: patch_deleted_attribute_disallowed_400 in EntityResourceTest
   */
  @Test
  void patch_deleted_attribute_disallowed_400(TestNamespace ns) {
    if (!supportsSoftDelete || !supportsPatch) return;

    // Create and soft delete entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    deleteEntity(entity.getId().toString());

    // Try to undelete via PATCH by setting deleted=false
    T deletedEntity = getEntityIncludeDeleted(entity.getId().toString());
    deletedEntity.setDeleted(false);

    String entityId = deletedEntity.getId().toString();
    // This should either fail or the deleted flag should remain true
    // depending on implementation
    try {
      T patched = patchEntity(entityId, deletedEntity);
      // If patch succeeds, deleted should still be true (ignored)
      assertTrue(patched.getDeleted(), "Deleted flag should not be changeable via PATCH");
    } catch (Exception e) {
      // Expected - PATCH on deleted entity may be disallowed
    }
  }

  // ===================================================================
  // PLACEHOLDER FOR REMAINING COMMON TESTS (Phase 4-5)
  // ===================================================================

  // Phase 4: ETag/Concurrency Tests
  // TODO: patch_etag_in_get_response
  // TODO: patch_with_valid_etag
  // TODO: patch_with_stale_etag
  // TODO: patch_concurrent_updates_with_etag
  // TODO: patch_concurrentUpdates_dataLossTest
  // TODO: patch_entityUpdatesOutsideASession

  // Phase 5: DataProducts/Domain Tests
  // TODO: patch_dataProducts_200_ok
  // TODO: patch_dataProducts_multipleOperations_200
  // Note: patchWrongDataProducts is covered by patch_invalidDataProducts_4xx
  // Note: patchWrongDomainId is covered by patch_entityWithInvalidDomain_4xx

  // ===================================================================
  // DOMAIN TESTS
  // ===================================================================

  @Test
  void patch_entityDomain_200(TestNamespace ns) {
    if (!supportsDomains || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    org.openmetadata.schema.type.EntityReference domainRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testDomain().getId())
            .withType("domain")
            .withName(testDomain().getName())
            .withFullyQualifiedName(testDomain().getFullyQualifiedName());

    entity.setDomains(List.of(domainRef));
    T updated = patchEntity(entity.getId().toString(), entity);

    T fetched = getEntityWithFields(updated.getId().toString(), "domains");
    assertNotNull(fetched.getDomains(), "Entity should have domains");
    assertEquals(1, fetched.getDomains().size(), "Entity should have 1 domain");
  }

  @Test
  void patch_entityWithInvalidDomain_4xx(TestNamespace ns) {
    if (!supportsDomains || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    org.openmetadata.schema.type.EntityReference invalidDomainRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(UUID.randomUUID())
            .withType("domain");

    entity.setDomains(List.of(invalidDomainRef));
    String entityId = entity.getId().toString();

    assertThrows(
        Exception.class,
        () -> patchEntity(entityId, entity),
        "Setting non-existent domain should fail");
  }

  // ===================================================================
  // DATA CONTRACT TESTS
  // ===================================================================

  @Test
  void get_entityDataContract_200(TestNamespace ns) {
    if (!supportsDataContract) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    org.openmetadata.schema.api.data.CreateDataContract contractRequest =
        new org.openmetadata.schema.api.data.CreateDataContract()
            .withName(ns.prefix("contract"))
            .withEntity(entity.getEntityReference())
            .withDescription("Data contract for test entity");
    org.openmetadata.schema.entity.data.DataContract contract =
        SdkClients.adminClient().dataContracts().create(contractRequest);

    T fetched = getEntityWithFields(entity.getId().toString(), "dataContract");
    assertNotNull(fetched.getDataContract(), "Entity should have a dataContract");
    assertEquals(
        contract.getId(),
        fetched.getDataContract().getId(),
        "DataContract reference should match created contract");
  }

  @Test
  void get_entityWithoutDataContract_200(TestNamespace ns) {
    if (!supportsDataContract) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    T fetched = getEntityWithFields(entity.getId().toString(), "dataContract");
    assertNull(
        fetched.getDataContract(), "Entity without a contract should have null dataContract");
  }

  // ===================================================================
  // ADMIN DELETE TESTS
  // ===================================================================

  @Test
  void delete_entityAsAdmin_200(TestNamespace ns) {

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    deleteEntity(entity.getId().toString());

    String entityId = entity.getId().toString();
    assertThrows(Exception.class, () -> getEntity(entityId), "Deleted entity should not be found");
  }

  @Test
  void delete_entityWithOwner_200(TestNamespace ns) {
    if (!supportsOwners || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    org.openmetadata.schema.type.EntityReference ownerRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser1().getId())
            .withType("user")
            .withName(testUser1().getName());

    entity.setOwners(List.of(ownerRef));
    T updated = patchEntity(entity.getId().toString(), entity);

    deleteEntity(updated.getId().toString());

    String entityId = updated.getId().toString();
    assertThrows(Exception.class, () -> getEntity(entityId), "Deleted entity should not be found");
  }

  // ===================================================================
  // DISPLAYNAME TESTS
  // ===================================================================

  @Test
  void patch_entityDisplayName_200(TestNamespace ns) {
    if (!supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    String newDisplayName = "Updated Display Name - " + System.currentTimeMillis();
    entity.setDisplayName(newDisplayName);
    T updated = patchEntity(entity.getId().toString(), entity);

    assertEquals(newDisplayName, updated.getDisplayName(), "DisplayName should be updated");
  }

  // ===================================================================
  // ATTRIBUTES PATCH TEST
  // ===================================================================

  @Test
  void patch_entityAttributes_200(TestNamespace ns) {
    if (!supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    Double initialVersion = entity.getVersion();

    String newDescription = "Updated description - " + System.currentTimeMillis();
    entity.setDescription(newDescription);
    T updated = patchEntity(entity.getId().toString(), entity);

    assertTrue(updated.getVersion() > initialVersion, "Version should increment after update");
    assertEquals(newDescription, updated.getDescription(), "Description should be updated");

    String newDisplayName = "Updated DisplayName - " + System.currentTimeMillis();
    updated.setDisplayName(newDisplayName);
    T updated2 = patchEntity(updated.getId().toString(), updated);

    assertEquals(newDisplayName, updated2.getDisplayName(), "DisplayName should be updated");
  }

  // ===================================================================
  // VALID OWNER TESTS
  // ===================================================================

  @Test
  void patch_validEntityOwner_200(TestNamespace ns) {
    if (!supportsOwners || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    org.openmetadata.schema.type.EntityReference userOwner =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser1().getId())
            .withType("user")
            .withName(testUser1().getName());

    entity.setOwners(List.of(userOwner));
    T withUserOwner = patchEntity(entity.getId().toString(), entity);

    T fetched = getEntityWithFields(withUserOwner.getId().toString(), "owners");
    assertNotNull(fetched.getOwners(), "Entity should have owners");
    assertEquals(1, fetched.getOwners().size(), "Entity should have 1 owner");
    assertEquals("user", fetched.getOwners().get(0).getType(), "Owner should be a user");

    org.openmetadata.schema.type.EntityReference teamOwner =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testGroupTeam().getId())
            .withType("team")
            .withName(testGroupTeam().getName());

    fetched.setOwners(List.of(teamOwner));
    T withTeamOwner = patchEntity(fetched.getId().toString(), fetched);

    T fetched2 = getEntityWithFields(withTeamOwner.getId().toString(), "owners");
    assertNotNull(fetched2.getOwners(), "Entity should have owners");
    assertEquals(1, fetched2.getOwners().size(), "Entity should have 1 owner");
    assertEquals("team", fetched2.getOwners().get(0).getType(), "Owner should be a team");
  }

  // ===================================================================
  // DELETED VERSION TESTS
  // ===================================================================

  @Test
  void get_deletedEntityVersion_200(TestNamespace ns) {
    if (!supportsSoftDelete || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    Double createdVersion = entity.getVersion();

    entity.setDescription("Updated before delete");
    T updated = patchEntity(entity.getId().toString(), entity);

    deleteEntity(updated.getId().toString());

    T deletedEntity = getEntityIncludeDeleted(updated.getId().toString());
    assertTrue(deletedEntity.getDeleted(), "Entity should be marked as deleted");

    T version01 = getVersion(entity.getId(), createdVersion);
    assertNotNull(version01, "Historical version should still be accessible");
    assertEquals(createdVersion, version01.getVersion(), 0.001);
  }

  // ===================================================================
  // SESSION CONSOLIDATION TESTS
  // ===================================================================

  @Test
  void patch_multipleUpdatesInSession_consolidation(TestNamespace ns) {
    if (!supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    entity.setDescription("First update");
    T update1 = patchEntity(entity.getId().toString(), entity);

    update1.setDescription("Second update");
    T update2 = patchEntity(update1.getId().toString(), update1);

    update2.setDescription("Third update");
    T update3 = patchEntity(update2.getId().toString(), update2);

    assertTrue(update3.getVersion() >= 0.1, "Version should be at least 0.1");
  }

  // ===================================================================
  // SYSTEM ENTITY TESTS
  // ===================================================================

  protected boolean isSystemEntity(T entity) {
    return false;
  }

  // ===================================================================
  // SEARCH TESTS (Elasticsearch) - TODO: Add when search index access is stable
  // ===================================================================

  protected String getSearchIndexName() {
    // Convert camelCase to snake_case for search index name
    String entityType = getEntityType();
    String snakeCase = entityType.replaceAll("([a-z])([A-Z])", "$1_$2").toLowerCase();
    return snakeCase + "_search_index";
  }

  protected void setDescription(K createRequest, String description) {
    try {
      createRequest
          .getClass()
          .getMethod("setDescription", String.class)
          .invoke(createRequest, description);
    } catch (Exception e) {
      fail(
          "Cannot set description on "
              + createRequest.getClass().getSimpleName()
              + ". Override setDescription() in your test class.");
    }
  }

  protected <V> void setFieldViaReflection(
      K createRequest, String setterName, Class<V> paramType, V value) {
    try {
      createRequest.getClass().getMethod(setterName, paramType).invoke(createRequest, value);
    } catch (Exception e) {
      fail(
          "Cannot call "
              + setterName
              + " on "
              + createRequest.getClass().getSimpleName()
              + ": "
              + e.getMessage());
    }
  }

  // ===================================================================
  // PUT OWNER UPDATE TESTS
  // ===================================================================

  protected boolean hasPutMethod() {
    return false;
  }

  @Test
  void put_entityUpdateOwner_200(TestNamespace ns) {
    if (!supportsOwners || !hasPutMethod()) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    org.openmetadata.schema.type.EntityReference ownerRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser1().getId())
            .withType("user")
            .withName(testUser1().getName());

    entity.setOwners(List.of(ownerRef));
    T updated = putEntity(entity);

    assertNotNull(updated.getOwners(), "Entity should have owners");
    assertEquals(1, updated.getOwners().size(), "Entity should have 1 owner");
  }

  protected T putEntity(T entity) {
    throw new UnsupportedOperationException("putEntity not implemented - override in subclass");
  }

  // ===================================================================
  // RELATIONSHIP FIELD CONSOLIDATION TESTS
  // ===================================================================

  @Test
  void patch_relationshipFields_consolidation_200(TestNamespace ns) {
    if (!supportsPatch || !supportsOwners) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    if (supportsDomains) {
      org.openmetadata.schema.type.EntityReference domainRef =
          new org.openmetadata.schema.type.EntityReference()
              .withId(testDomain().getId())
              .withType("domain")
              .withName(testDomain().getName())
              .withFullyQualifiedName(testDomain().getFullyQualifiedName());

      entity.setDomains(List.of(domainRef));
      T updated = patchEntity(entity.getId().toString(), entity);

      T fetched = getEntityWithFields(updated.getId().toString(), "domains");
      assertNotNull(fetched.getDomains(), "Entity should have domains after first patch");
    }

    org.openmetadata.schema.type.EntityReference ownerRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser1().getId())
            .withType("user")
            .withName(testUser1().getName());

    T current = getEntity(entity.getId().toString());
    current.setOwners(List.of(ownerRef));
    T updated2 = patchEntity(current.getId().toString(), current);

    T fetched2 = getEntityWithFields(updated2.getId().toString(), "owners");
    assertNotNull(fetched2.getOwners(), "Entity should have owners after second patch");
  }

  // ===================================================================
  // DATAPRODUCT TESTS
  // ===================================================================

  @Test
  void patch_dataProducts_200(TestNamespace ns) {
    if (!supportsDataProducts || !supportsDomains || !supportsPatch) return;
    // DataProduct tests require creating a DataProduct first, skipping for now
  }

  @Test
  void patch_invalidDataProducts_4xx(TestNamespace ns) {
    if (!supportsDataProducts || !supportsPatch) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    org.openmetadata.schema.type.EntityReference invalidDataProductRef =
        new org.openmetadata.schema.type.EntityReference().withId(UUID.randomUUID());

    entity.setDataProducts(List.of(invalidDataProductRef));
    String entityId = entity.getId().toString();

    assertThrows(
        Exception.class,
        () -> patchEntity(entityId, entity),
        "Setting non-existent dataProduct should fail");
  }

  // ===================================================================
  // PLACEHOLDER FOR REMAINING TESTS
  // ===================================================================

  // ===================================================================
  // AUTHORIZATION TESTS
  // ===================================================================

  protected boolean supportsLifeCycle = false;
  protected boolean supportsCertification = false;

  /**
   * Test: Owner can update their own entity
   * Equivalent to: put_entityCreate_as_owner_200 in EntityResourceTest
   */
  @Test
  void put_entityCreate_as_owner_200(TestNamespace ns) {
    if (!supportsOwners || !hasPutMethod()) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Set testUser1 as owner
    org.openmetadata.schema.type.EntityReference ownerRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser1().getId())
            .withType("user")
            .withName(testUser1().getName());

    entity.setOwners(List.of(ownerRef));
    T withOwner = patchEntity(entity.getId().toString(), entity);

    // Verify owner was set
    T fetched = getEntityWithFields(withOwner.getId().toString(), "owners");
    assertNotNull(fetched.getOwners(), "Entity should have owners");
    assertEquals(1, fetched.getOwners().size(), "Entity should have 1 owner");

    // TODO: Owner-based authorization tests need rework for fluent API pattern
    // The actual owner update test is skipped - would need entity-specific client usage
  }

  // TODO: Authorization tests need rework for fluent API pattern
  // Test: Non-owner cannot update entity they don't own
  // Equivalent to: put_entityUpdate_as_non_owner_4xx in EntityResourceTest
  // Commented out until authorization test pattern is redesigned

  // ===================================================================
  // LIFECYCLE TESTS
  // ===================================================================

  /**
   * Test: Add and update lifecycle information on entity
   * Equivalent to: postPutPatch_entityLifeCycle in EntityResourceTest
   */
  @Test
  void postPutPatch_entityLifeCycle(TestNamespace ns) {
    if (!supportsLifeCycle || !supportsPatch) return;

    // Create entity without lifecycle
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Add lifecycle using PATCH
    org.openmetadata.schema.type.AccessDetails accessed =
        new org.openmetadata.schema.type.AccessDetails()
            .withTimestamp(System.currentTimeMillis() / 1000)
            .withAccessedBy(testUser2Ref());

    org.openmetadata.schema.type.LifeCycle lifeCycle =
        new org.openmetadata.schema.type.LifeCycle().withAccessed(accessed);

    entity.setLifeCycle(lifeCycle);
    T updated = patchEntity(entity.getId().toString(), entity);

    // Verify lifecycle was set
    T fetched = getEntityWithFields(updated.getId().toString(), "lifeCycle");
    assertNotNull(fetched.getLifeCycle(), "Entity should have lifecycle");
    assertNotNull(fetched.getLifeCycle().getAccessed(), "Lifecycle should have accessed info");

    // Update lifecycle with created info
    org.openmetadata.schema.type.AccessDetails created =
        new org.openmetadata.schema.type.AccessDetails()
            .withTimestamp(System.currentTimeMillis() / 1000 - 1000)
            .withAccessedBy(testUser1Ref());

    fetched.getLifeCycle().setCreated(created);
    T updated2 = patchEntity(fetched.getId().toString(), fetched);

    T fetched2 = getEntityWithFields(updated2.getId().toString(), "lifeCycle");
    assertNotNull(fetched2.getLifeCycle().getCreated(), "Lifecycle should have created info");
  }

  /**
   * Test: Verify lifecycle updates do NOT cause version pollution
   * This test verifies the fix for https://github.com/open-metadata/OpenMetadata/issues/21326
   *
   * The bug was: Every time usage ingestion runs, it updates the lifecycle "accessed" timestamp,
   * which caused the entity version to increment. Over time, this led to entities with
   * extremely high version numbers (e.g., version 598.4), making the version history UI
   * slow and potentially causing crashes.
   *
   * The fix: Lifecycle-only changes should NOT increment the entity version.
   */
  @Test
  void patch_entityLifeCycle_noVersionPollution(TestNamespace ns) {
    if (!supportsLifeCycle || !supportsPatch) return;

    // Create entity without lifecycle
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    Double initialVersion = entity.getVersion();

    // Add initial lifecycle with accessed timestamp
    org.openmetadata.schema.type.AccessDetails accessed =
        new org.openmetadata.schema.type.AccessDetails()
            .withTimestamp(1695059900L)
            .withAccessedBy(testUser2Ref());

    org.openmetadata.schema.type.LifeCycle lifeCycle =
        new org.openmetadata.schema.type.LifeCycle().withAccessed(accessed);

    entity.setLifeCycle(lifeCycle);
    T updated = patchEntity(entity.getId().toString(), entity);
    Double versionAfterFirstLifeCycleUpdate = updated.getVersion();

    // Verify lifecycle was set but version did not change
    assertEquals(
        initialVersion,
        versionAfterFirstLifeCycleUpdate,
        "Lifecycle-only changes should NOT increment version");

    // Simulate usage run updating accessed time with a newer timestamp
    // This is what happens when usage ingestion runs repeatedly
    T fetched = getEntityWithFields(updated.getId().toString(), "lifeCycle");
    org.openmetadata.schema.type.AccessDetails accessedNewer =
        new org.openmetadata.schema.type.AccessDetails()
            .withTimestamp(1695060000L)
            .withAccessedBy(testUser2Ref());

    org.openmetadata.schema.type.LifeCycle lifeCycleNewer =
        new org.openmetadata.schema.type.LifeCycle().withAccessed(accessedNewer);

    fetched.setLifeCycle(lifeCycleNewer);
    T updated2 = patchEntity(fetched.getId().toString(), fetched);
    Double versionAfterSecondLifeCycleUpdate = updated2.getVersion();

    // Verify version did NOT increment for lifecycle-only change
    assertEquals(
        versionAfterFirstLifeCycleUpdate,
        versionAfterSecondLifeCycleUpdate,
        "Lifecycle-only changes should NOT increment version");

    // Simulate another usage run with even newer timestamp
    T fetched2 = getEntityWithFields(updated2.getId().toString(), "lifeCycle");
    org.openmetadata.schema.type.AccessDetails accessedEvenNewer =
        new org.openmetadata.schema.type.AccessDetails()
            .withTimestamp(1695060100L)
            .withAccessedBy(testUser2Ref());

    org.openmetadata.schema.type.LifeCycle lifeCycleEvenNewer =
        new org.openmetadata.schema.type.LifeCycle().withAccessed(accessedEvenNewer);

    fetched2.setLifeCycle(lifeCycleEvenNewer);
    T updated3 = patchEntity(fetched2.getId().toString(), fetched2);
    Double versionAfterThirdLifeCycleUpdate = updated3.getVersion();

    // Verify version still did NOT increment
    assertEquals(
        versionAfterSecondLifeCycleUpdate,
        versionAfterThirdLifeCycleUpdate,
        "Lifecycle-only changes should NOT increment version");

    // Verify the lifecycle data was actually updated even though version didn't change
    T finalEntity = getEntityWithFields(updated3.getId().toString(), "lifeCycle");
    assertNotNull(finalEntity.getLifeCycle(), "Lifecycle should still be present");
    assertEquals(
        1695060100L,
        finalEntity.getLifeCycle().getAccessed().getTimestamp(),
        "Lifecycle accessed timestamp should be updated to latest value");
  }

  /**
   * Test: When lifecycle AND other fields change together, version SHOULD increment
   * This ensures that the fix for lifecycle version pollution doesn't break
   * normal versioning when real changes occur alongside lifecycle updates.
   */
  @Test
  void patch_entityLifeCycleWithOtherChanges_versionIncrements(TestNamespace ns) {
    if (!supportsLifeCycle || !supportsPatch) return;

    // Create entity without lifecycle
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    Double initialVersion = entity.getVersion();

    // Add lifecycle AND change description at the same time
    org.openmetadata.schema.type.AccessDetails accessed =
        new org.openmetadata.schema.type.AccessDetails()
            .withTimestamp(1695059900L)
            .withAccessedBy(testUser2Ref());

    org.openmetadata.schema.type.LifeCycle lifeCycle =
        new org.openmetadata.schema.type.LifeCycle().withAccessed(accessed);

    entity.setLifeCycle(lifeCycle);
    entity.setDescription("Updated description for version test");
    T updated = patchEntity(entity.getId().toString(), entity);

    // Version SHOULD increment because description changed (not because of lifecycle)
    assertTrue(
        updated.getVersion() > initialVersion,
        "Version should increment when description changes alongside lifecycle. "
            + "Initial: "
            + initialVersion
            + ", After: "
            + updated.getVersion());

    // Now update ONLY lifecycle (no description change) - version should NOT increment
    T fetched = getEntityWithFields(updated.getId().toString(), "lifeCycle");
    Double versionAfterDescriptionChange = fetched.getVersion();

    org.openmetadata.schema.type.AccessDetails accessedNewer =
        new org.openmetadata.schema.type.AccessDetails()
            .withTimestamp(1695060000L)
            .withAccessedBy(testUser2Ref());

    fetched.setLifeCycle(new org.openmetadata.schema.type.LifeCycle().withAccessed(accessedNewer));
    T updated2 = patchEntity(fetched.getId().toString(), fetched);

    // Version should NOT increment since only lifecycle changed
    assertEquals(
        versionAfterDescriptionChange,
        updated2.getVersion(),
        "Version should NOT increment when only lifecycle changes");
  }

  /**
   * Helper method to get entity with lifecycle field.
   * Subclasses should override getEntityWithFields to include lifecycle.
   */
  protected T getEntityWithLifeCycle(String id) {
    return getEntityWithFields(id, "lifeCycle");
  }

  // ===================================================================
  // CERTIFICATION TESTS
  // ===================================================================

  /**
   * Test: Add certification to entity
   * Equivalent to: postPutPatch_entityCertification in EntityResourceTest
   */
  @Test
  void postPutPatch_entityCertification(TestNamespace ns) {
    if (!supportsCertification || !supportsPatch) return;

    // Create entity without certification
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Note: Full certification test requires:
    // 1. Creating a certification tag
    // 2. Configuring certification settings
    // 3. Applying certification to entity
    // This is a simplified test that verifies the certification field is patchable

    // Verify entity has no certification initially
    T fetched = getEntity(entity.getId().toString());
    assertNull(fetched.getCertification(), "Entity should not have certification initially");
  }

  // ===================================================================
  // CUSTOM EXTENSION TESTS
  // ===================================================================

  /**
   * Test: Add custom extension attributes to entity
   * Equivalent to: put_addEntityCustomAttributes in EntityResourceTest
   */
  @Test
  void put_addEntityCustomAttributes(TestNamespace ns) {
    if (!supportsCustomExtension) return;

    // Create entity without extension
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Note: Full custom attributes test requires:
    // 1. Adding custom property to entity type via Type API
    // 2. Creating entity with extension
    // 3. Updating extension via PUT and PATCH
    // This is a placeholder that verifies the entity can be created

    assertNotNull(entity.getId(), "Entity should be created");
  }

  // ===================================================================
  // SESSION TIMEOUT TESTS
  // ===================================================================

  /**
   * Test: Updates outside a session should create new change events
   * Equivalent to: patch_entityUpdatesOutsideASession in EntityResourceTest
   */
  @Test
  void patch_entityUpdatesOutsideASession(TestNamespace ns) {
    if (!supportsPatch) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    Double v1 = entity.getVersion();
    assertEquals(0.1, v1, 0.001, "Initial version should be 0.1");

    // First update within session
    String desc1 = "First update - " + System.currentTimeMillis();
    entity.setDescription(desc1);
    T updated1 = patchEntity(entity.getId().toString(), entity);
    Double v2 = updated1.getVersion();
    assertTrue(v2 > v1, "Version should increment after first update");

    // Second update - still in session, changes may be consolidated
    String desc2 = "Second update - " + System.currentTimeMillis();
    updated1.setDescription(desc2);
    T updated2 = patchEntity(updated1.getId().toString(), updated1);
    Double v3 = updated2.getVersion();
    assertTrue(v3 >= v2, "Version should be >= previous version");

    // Verify final description
    T fetched = getEntity(updated2.getId().toString());
    assertEquals(desc2, fetched.getDescription(), "Description should be updated");
  }

  // ===================================================================
  // DATA PRODUCT TESTS (Additional)
  // ===================================================================

  /**
   * Test: Add data products to entity with multiple operations
   * Equivalent to: patch_dataProducts_multipleOperations_200 in EntityResourceTest
   */
  @Test
  void patch_dataProducts_multipleOperations_200(TestNamespace ns) {
    if (!supportsDataProducts || !supportsDomains || !supportsPatch) return;

    // Note: This test requires creating DataProducts via DataProductService
    // which needs domain to be set first. This is a placeholder.
  }

  // ===================================================================
  // ASYNC DELETE TESTS (Placeholder - requires WebSocket)
  // ===================================================================

  protected boolean supportsAsyncDelete = false;

  /**
   * Test: Async delete of non-existent entity returns 404
   * Equivalent to: delete_async_nonExistentEntity_404 in EntityResourceTest
   */
  @Test
  void delete_async_nonExistentEntity_404(TestNamespace ns) {
    if (!supportsAsyncDelete) return;

    // Note: Async delete requires WebSocket connection to receive delete messages
    // This is a placeholder test
  }

  /**
   * Test: Async delete as non-admin should fail
   * Equivalent to: delete_async_entity_as_non_admin_401 in EntityResourceTest
   */
  @Test
  void delete_async_entity_as_non_admin_401(TestNamespace ns) {
    if (!supportsAsyncDelete) return;

    // Note: Async delete requires WebSocket connection
    // This is a placeholder test
  }

  /**
   * Test: Async delete with recursive hard delete
   * Equivalent to: delete_async_with_recursive_hardDelete in EntityResourceTest
   */
  @Test
  void delete_async_with_recursive_hardDelete(TestNamespace ns) {
    if (!supportsAsyncDelete) return;

    // Note: Async delete requires WebSocket connection
    // This is a placeholder test
  }

  /**
   * Test: Async soft delete
   * Equivalent to: delete_async_soft_delete in EntityResourceTest
   */
  @Test
  void delete_async_soft_delete(TestNamespace ns) {
    if (!supportsAsyncDelete || !supportsSoftDelete) return;

    // Note: Async delete requires WebSocket connection
    // This is a placeholder test
  }

  // ===================================================================
  // RECOGNIZER FEEDBACK TESTS (Placeholder - requires specific setup)
  // ===================================================================

  protected boolean supportsRecognizerFeedback = false;

  /**
   * Test: Recognizer feedback for auto-applied tags
   * Equivalent to: test_recognizerFeedback_autoAppliedTags in EntityResourceTest
   */
  @Test
  void test_recognizerFeedback_autoAppliedTags(TestNamespace ns) {
    if (!supportsRecognizerFeedback || !supportsTags) return;

    // Note: Recognizer feedback tests require specific recognizer setup
    // This is a placeholder test
  }

  /**
   * Test: Recognizer feedback exception list
   * Equivalent to: test_recognizerFeedback_exceptionList in EntityResourceTest
   */
  @Test
  void test_recognizerFeedback_exceptionList(TestNamespace ns) {
    if (!supportsRecognizerFeedback || !supportsTags) return;

    // Note: Recognizer feedback tests require specific recognizer setup
    // This is a placeholder test
  }

  // ===================================================================
  // FIELD FETCHER EFFICIENCY TESTS
  // ===================================================================

  /**
   * Test: Verify field fetchers work correctly
   * Equivalent to: test_fieldFetchers in EntityResourceTest
   */
  @Test
  void test_fieldFetchers(TestNamespace ns) {

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Verify entity can be fetched with different field combinations
    T basic = getEntity(entity.getId().toString());
    assertNotNull(basic, "Basic fetch should work");
    assertNotNull(basic.getId(), "ID should be present");

    // Fetch with specific fields
    if (supportsOwners) {
      T withOwners = getEntityWithFields(entity.getId().toString(), "owners");
      assertNotNull(withOwners, "Fetch with owners field should work");
    }

    if (supportsTags) {
      T withTags = getEntityWithFields(entity.getId().toString(), "tags");
      assertNotNull(withTags, "Fetch with tags field should work");
    }
  }

  /**
   * Test: Entity Status functionality
   * Verifies that entityStatus field works correctly for entities that support it
   */
  @Test
  void test_entityStatus(TestNamespace ns) {
    // Only test if entity has entityStatus field (check if getEntityStatus method exists)
    if ("glossaryTerm".equals(getEntityType())) {
      log.info(
          "Skipping entityStatus test for GlossaryTerm - has different entityStatus implementation");
      return;
    }

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    try {
      // Check if entity supports entityStatus
      org.openmetadata.schema.type.EntityStatus currentStatus = entity.getEntityStatus();

      // If entityStatus is null, the entity doesn't support this field - skip the test
      if (currentStatus == null) {
        log.info(
            "Entity {} does not support entityStatus field - skipping test",
            entity.getClass().getSimpleName());
        return;
      }

      // Default status should be UNPROCESSED
      assertEquals(
          org.openmetadata.schema.type.EntityStatus.UNPROCESSED,
          currentStatus,
          "Default entity status should be UNPROCESSED");

      // Test updating entityStatus via PATCH
      if (supportsPatch) {
        // Update to DRAFT
        entity.setEntityStatus(org.openmetadata.schema.type.EntityStatus.DRAFT);
        T updatedEntity = patchEntity(entity.getId().toString(), entity);
        assertEquals(
            org.openmetadata.schema.type.EntityStatus.DRAFT,
            updatedEntity.getEntityStatus(),
            "Entity status should be updated to DRAFT");

        // Update to IN_REVIEW
        updatedEntity.setEntityStatus(org.openmetadata.schema.type.EntityStatus.IN_REVIEW);
        T reviewEntity = patchEntity(updatedEntity.getId().toString(), updatedEntity);
        assertEquals(
            org.openmetadata.schema.type.EntityStatus.IN_REVIEW,
            reviewEntity.getEntityStatus(),
            "Entity status should be updated to IN_REVIEW");

        // Update to APPROVED
        reviewEntity.setEntityStatus(org.openmetadata.schema.type.EntityStatus.APPROVED);
        T approvedEntity = patchEntity(reviewEntity.getId().toString(), reviewEntity);
        assertEquals(
            org.openmetadata.schema.type.EntityStatus.APPROVED,
            approvedEntity.getEntityStatus(),
            "Entity status should be updated to APPROVED");

        // Update to DEPRECATED
        approvedEntity.setEntityStatus(org.openmetadata.schema.type.EntityStatus.DEPRECATED);
        T deprecatedEntity = patchEntity(approvedEntity.getId().toString(), approvedEntity);
        assertEquals(
            org.openmetadata.schema.type.EntityStatus.DEPRECATED,
            deprecatedEntity.getEntityStatus(),
            "Entity status should be updated to DEPRECATED");

        // Update to REJECTED
        deprecatedEntity.setEntityStatus(org.openmetadata.schema.type.EntityStatus.REJECTED);
        T rejectedEntity = patchEntity(deprecatedEntity.getId().toString(), deprecatedEntity);
        assertEquals(
            org.openmetadata.schema.type.EntityStatus.REJECTED,
            rejectedEntity.getEntityStatus(),
            "Entity status should be updated to REJECTED");

        // Verify entity can be retrieved with correct status
        T fetchedEntity = getEntity(rejectedEntity.getId().toString());
        assertEquals(
            org.openmetadata.schema.type.EntityStatus.REJECTED,
            fetchedEntity.getEntityStatus(),
            "Fetched entity should maintain the REJECTED status");

        // Verify version increments with status changes (if entity supports versioning)
        assertTrue(
            fetchedEntity.getVersion() > entity.getVersion(),
            "Version should increment when entityStatus is updated");
      }

    } catch (NoSuchMethodError | UnsupportedOperationException e) {
      log.info(
          "Entity "
              + entity.getClass().getSimpleName()
              + " does not support entityStatus: "
              + e.getMessage());
    }
  }

  /**
   * Test: Bulk loading efficiency
   * Test: Bulk entity creation and fetching works correctly.
   * Basic functionality test - comprehensive pagination is in PaginationIT.
   */
  @Test
  void test_bulkLoadingEfficiency(TestNamespace ns) {
    // Create multiple entities
    int count = 3;
    List<UUID> createdIds = new ArrayList<>();
    for (int i = 0; i < count; i++) {
      K createRequest = createRequest(ns.prefix("bulk" + i), ns);
      T entity = createEntity(createRequest);
      createdIds.add(entity.getId());
    }

    // Verify all entities can be fetched individually
    for (UUID id : createdIds) {
      T fetched = getEntity(id.toString());
      assertNotNull(fetched, "Entity should be fetchable");
    }

    // Basic list test - just verify list works
    org.openmetadata.sdk.models.ListParams params = new org.openmetadata.sdk.models.ListParams();
    params.setLimit(10);
    org.openmetadata.sdk.models.ListResponse<T> response = listEntities(params);
    assertNotNull(response, "List response should not be null");
    assertTrue(response.getData().size() > 0, "Should have entities");
  }

  // ===================================================================
  // ETAG/VERSION-BASED CONCURRENCY TESTS
  // ===================================================================

  protected boolean supportsEtag = true;

  /**
   * Test: Verify ETag/version is present in GET response
   * Equivalent to: patch_etag_in_get_response in EntityResourceTest
   *
   * In SDK, we use entity version as the optimistic lock mechanism.
   */
  @Test
  void patch_etag_in_get_response(TestNamespace ns) {
    if (!supportsPatch || !supportsEtag) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Get entity and verify version is present
    T fetched = getEntity(entity.getId().toString());
    assertNotNull(fetched.getVersion(), "Version (ETag equivalent) should be present");
    assertEquals(0.1, fetched.getVersion(), 0.001, "Initial version should be 0.1");

    // Update entity
    fetched.setDescription("Updated for ETag test");
    T updated = patchEntity(fetched.getId().toString(), fetched);

    // Verify version changed
    assertNotNull(updated.getVersion(), "Version should be present after update");
    assertTrue(updated.getVersion() > 0.1, "Version should increment after update");
  }

  /**
   * Test: PATCH with valid version succeeds
   * Equivalent to: patch_with_valid_etag in EntityResourceTest
   */
  @Test
  void patch_with_valid_etag(TestNamespace ns) {
    if (!supportsPatch || !supportsEtag) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    Double originalVersion = entity.getVersion();

    // Get fresh copy (simulates getting entity with current version/ETag)
    T fetched = getEntity(entity.getId().toString());
    assertEquals(originalVersion, fetched.getVersion(), 0.001, "Versions should match");

    // Update with current version
    fetched.setDescription("Updated with valid ETag/version");
    T updated = patchEntity(fetched.getId().toString(), fetched);

    // Verify update succeeded
    assertEquals("Updated with valid ETag/version", updated.getDescription());
    assertTrue(updated.getVersion() > originalVersion, "Version should increment");
  }

  /**
   * Test: PATCH with stale version behavior
   * Equivalent to: patch_with_stale_etag in EntityResourceTest
   *
   * Note: SDK JSON Patch merges changes, so stale version updates may succeed
   * if they don't conflict. This test verifies the version tracking works.
   */
  @Test
  void patch_with_stale_etag(TestNamespace ns) {
    if (!supportsPatch || !supportsEtag) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Get entity and save version
    T firstCopy = getEntity(entity.getId().toString());
    Double firstVersion = firstCopy.getVersion();

    // First update changes the version
    firstCopy.setDescription("First update");
    T afterFirst = patchEntity(firstCopy.getId().toString(), firstCopy);
    assertTrue(
        afterFirst.getVersion() > firstVersion, "Version should increment after first update");

    // Get fresh copy with new version
    T secondCopy = getEntity(entity.getId().toString());

    // Update with the current (not stale) version
    secondCopy.setDescription("Second update with current version");
    T afterSecond = patchEntity(secondCopy.getId().toString(), secondCopy);

    // Verify update worked
    assertEquals("Second update with current version", afterSecond.getDescription());
    assertTrue(
        afterSecond.getVersion() >= afterFirst.getVersion(), "Version should be >= previous");
  }

  /**
   * Test: Concurrent updates with version-based optimistic locking
   * Equivalent to: patch_concurrent_updates_with_etag in EntityResourceTest
   */
  @Test
  void patch_concurrent_updates_with_etag(TestNamespace ns) throws Exception {
    if (!supportsPatch || !supportsEtag) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Get entity
    T fetched = getEntity(entity.getId().toString());
    Double originalVersion = fetched.getVersion();

    // Concurrent update 1: description
    String desc1 = "Concurrent update 1 - " + System.currentTimeMillis();
    fetched.setDescription(desc1);
    T updated1 = patchEntity(fetched.getId().toString(), fetched);
    assertTrue(updated1.getVersion() > originalVersion, "Version should increment after update 1");

    // Get fresh copy for update 2
    T fresh = getEntity(entity.getId().toString());

    // Concurrent update 2: display name
    String displayName2 = "Concurrent Display - " + System.currentTimeMillis();
    fresh.setDisplayName(displayName2);
    T updated2 = patchEntity(fresh.getId().toString(), fresh);

    // Both updates should succeed (SDK merges non-conflicting changes)
    assertTrue(
        updated2.getVersion() >= updated1.getVersion(), "Version should be >= after update 2");

    // Verify final state
    T finalEntity = getEntity(entity.getId().toString());
    assertNotNull(finalEntity.getDescription(), "Description should be present");
    assertNotNull(finalEntity.getDisplayName(), "DisplayName should be present");
  }

  /**
   * Test: Concurrent updates should not cause data loss
   * Equivalent to: patch_concurrentUpdates_dataLossTest in EntityResourceTest
   */
  @Test
  void patch_concurrentUpdates_dataLossTest(TestNamespace ns) throws Exception {
    if (!supportsPatch || !supportsEtag) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Set initial description
    T fetched = getEntity(entity.getId().toString());
    fetched.setDescription("Initial description");
    T withDesc = patchEntity(fetched.getId().toString(), fetched);

    // Update 1: Modify description
    String newDesc = "Updated description - " + System.currentTimeMillis();
    withDesc.setDescription(newDesc);
    T updated1 = patchEntity(withDesc.getId().toString(), withDesc);
    assertEquals(newDesc, updated1.getDescription(), "Description should be updated");

    // Update 2: Modify display name (should not lose description)
    T fresh = getEntity(entity.getId().toString());
    String newDisplayName = "New Display Name - " + System.currentTimeMillis();
    fresh.setDisplayName(newDisplayName);
    T updated2 = patchEntity(fresh.getId().toString(), fresh);

    // Verify no data loss
    T finalEntity = getEntity(entity.getId().toString());
    assertEquals(newDisplayName, finalEntity.getDisplayName(), "DisplayName should be updated");
    assertEquals(newDesc, finalEntity.getDescription(), "Description should NOT be lost");
  }

  // ===================================================================
  // SEARCH INDEX TESTS
  // ===================================================================

  /**
   * Test: Entity with null description shows INCOMPLETE in search
   * Equivalent to: get_entityWithNullDescriptionFromSearch in EntityResourceTest
   */
  @Test
  void get_entityWithNullDescriptionFromSearch(TestNamespace ns) {
    if (!supportsSearchIndex || !supportsEmptyDescription) return;

    // Create entity without description
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Note: Full search test requires Elasticsearch integration
    // This verifies the entity was created with null/empty description
    T fetched = getEntity(entity.getId().toString());
    assertNotNull(fetched, "Entity should exist");
  }

  /**
   * Test: Entity with empty description shows INCOMPLETE in search
   * Equivalent to: get_entityWithEmptyDescriptionFromSearch in EntityResourceTest
   */
  @Test
  void get_entityWithEmptyDescriptionFromSearch(TestNamespace ns) {
    if (!supportsSearchIndex || !supportsEmptyDescription || !supportsPatch) return;

    // Create entity with empty description
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    entity.setDescription("");
    T updated = patchEntity(entity.getId().toString(), entity);

    // Verify empty description was set
    T fetched = getEntity(updated.getId().toString());
    assertEquals("", fetched.getDescription(), "Description should be empty");
  }

  // ===================================================================
  // SDK CRUD OPERATIONS TESTS
  // ===================================================================

  /**
   * Test: SDK CRUD operations (Create, Retrieve, Update, Delete)
   * Equivalent to: test_sdkCRUDOperations in EntityResourceTest
   */
  @Test
  void test_sdkCRUDOperations(TestNamespace ns) {

    // CREATE
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    assertNotNull(created, "Created entity should not be null");
    assertNotNull(created.getId(), "Entity ID should not be null");
    assertEquals(0.1, created.getVersion(), 0.001, "Initial version should be 0.1");

    // RETRIEVE by ID
    T retrievedById = getEntity(created.getId().toString());
    assertNotNull(retrievedById, "Retrieved entity should not be null");
    assertEquals(created.getId(), retrievedById.getId(), "IDs should match");
    assertEquals(created.getName(), retrievedById.getName(), "Names should match");

    // RETRIEVE by Name
    T retrievedByName = getEntityByName(created.getFullyQualifiedName());
    assertNotNull(retrievedByName, "Retrieved by name should not be null");
    assertEquals(created.getId(), retrievedByName.getId(), "IDs should match");

    // UPDATE
    String newDescription = "Updated via SDK CRUD test - " + System.currentTimeMillis();
    retrievedById.setDescription(newDescription);
    T updated = patchEntity(retrievedById.getId().toString(), retrievedById);
    assertEquals(newDescription, updated.getDescription(), "Description should be updated");
    assertTrue(updated.getVersion() > 0.1, "Version should increment");

    // DELETE
    deleteEntity(created.getId().toString());
    String entityId = created.getId().toString();
    assertThrows(
        Exception.class, () -> getEntity(entityId), "Deleted entity should not be retrievable");
  }

  /**
   * Test: SDK delete with options (soft delete, hard delete)
   * Equivalent to: test_sdkDeleteWithOptions in EntityResourceTest
   */
  @Test
  void test_sdkDeleteWithOptions(TestNamespace ns) {
    if (!supportsSoftDelete) return;

    // Create entity for soft delete test
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Soft delete
    deleteEntity(entity.getId().toString());

    // Should be able to retrieve with include=deleted
    T softDeleted = getEntityIncludeDeleted(entity.getId().toString());
    assertNotNull(softDeleted, "Soft deleted entity should be retrievable with include=deleted");
    assertTrue(softDeleted.getDeleted(), "Entity should be marked as deleted");

    // Hard delete
    hardDeleteEntity(entity.getId().toString());

    // Should not be retrievable even with include=deleted
    String entityId = entity.getId().toString();
    assertThrows(
        Exception.class,
        () -> getEntityIncludeDeleted(entityId),
        "Hard deleted entity should not be retrievable");
  }

  /**
   * Test: SDK entity with tags
   * Equivalent to: test_sdkEntityWithTags in EntityResourceTest
   */
  @Test
  void test_sdkEntityWithTags(TestNamespace ns) {
    if (!supportsTags) return;

    // Create test-specific classification and tags to avoid deadlocks with parallel tests
    OpenMetadataClient client = SdkClients.adminClient();
    String classificationName = ns.prefix("SdkTagsClassification");

    org.openmetadata.schema.api.classification.CreateClassification createClassification =
        new org.openmetadata.schema.api.classification.CreateClassification()
            .withName(classificationName)
            .withDescription("Classification for SDK tags test");
    client
        .getHttpClient()
        .execute(
            HttpMethod.PUT,
            "/v1/classifications",
            createClassification,
            org.openmetadata.schema.entity.classification.Classification.class);

    // Create test-specific tags
    String tag1Name = "SdkTag1";
    String tag2Name = "SdkTag2";

    for (String tagName : List.of(tag1Name, tag2Name)) {
      org.openmetadata.schema.api.classification.CreateTag createTag =
          new org.openmetadata.schema.api.classification.CreateTag()
              .withName(tagName)
              .withDescription("Tag for SDK test")
              .withClassification(classificationName);
      client
          .getHttpClient()
          .execute(
              HttpMethod.PUT,
              "/v1/tags",
              createTag,
              org.openmetadata.schema.entity.classification.Tag.class);
    }

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Add tags
    TagLabel tag1 = new TagLabel().withTagFQN(classificationName + "." + tag1Name);
    TagLabel tag2 = new TagLabel().withTagFQN(classificationName + "." + tag2Name);
    entity.setTags(List.of(tag1, tag2));
    T withTags = patchEntity(entity.getId().toString(), entity);

    // Verify tags
    T fetched = getEntityWithFields(withTags.getId().toString(), "tags");
    assertNotNull(fetched.getTags(), "Entity should have tags");
    assertEquals(2, fetched.getTags().size(), "Entity should have 2 tags");
  }

  /**
   * Test: SDK entity with owners
   * Equivalent to: test_sdkEntityWithOwners in EntityResourceTest
   */
  @Test
  void test_sdkEntityWithOwners(TestNamespace ns) {
    if (!supportsOwners || !supportsPatch) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Add owner
    org.openmetadata.schema.type.EntityReference ownerRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testUser1().getId())
            .withType("user")
            .withName(testUser1().getName());

    entity.setOwners(List.of(ownerRef));
    T withOwner = patchEntity(entity.getId().toString(), entity);

    // Verify owner
    T fetched = getEntityWithFields(withOwner.getId().toString(), "owners");
    assertNotNull(fetched.getOwners(), "Entity should have owners");
    assertEquals(1, fetched.getOwners().size(), "Entity should have 1 owner");
    assertEquals(testUser1().getId(), fetched.getOwners().get(0).getId(), "Owner ID should match");
  }

  /**
   * Test: SDK entity with domain and data products
   * Equivalent to: test_sdkEntityWithDomainAndDataProducts in EntityResourceTest
   */
  @Test
  void test_sdkEntityWithDomainAndDataProducts(TestNamespace ns) {
    if (!supportsDomains || !supportsPatch || !supportsPatchDomains) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Add domain
    org.openmetadata.schema.type.EntityReference domainRef =
        new org.openmetadata.schema.type.EntityReference()
            .withId(testDomain().getId())
            .withType("domain")
            .withName(testDomain().getName())
            .withFullyQualifiedName(testDomain().getFullyQualifiedName());

    entity.setDomains(List.of(domainRef));
    T withDomain = patchEntity(entity.getId().toString(), entity);

    // Verify domain
    T fetched = getEntityWithFields(withDomain.getId().toString(), "domains");
    assertNotNull(fetched.getDomains(), "Entity should have domains");
    assertEquals(1, fetched.getDomains().size(), "Entity should have 1 domain");
    assertEquals(
        testDomain().getId(), fetched.getDomains().get(0).getId(), "Domain ID should match");
  }

  // ===================================================================
  // SDK FLUENT API TESTS
  // ===================================================================

  /**
   * Test: SDK list fluent API works correctly.
   * Basic functionality test - comprehensive pagination is in PaginationIT.
   */
  @Test
  void testListFluentAPI(TestNamespace ns) {
    // Create a few entities
    for (int i = 0; i < 3; i++) {
      K createRequest = createRequest(ns.prefix("list" + i), ns);
      createEntity(createRequest);
    }

    // Basic list test - just verify list API works
    org.openmetadata.sdk.models.ListParams params = new org.openmetadata.sdk.models.ListParams();
    params.setLimit(10);
    org.openmetadata.sdk.models.ListResponse<T> response = listEntities(params);

    assertNotNull(response, "List response should not be null");
    assertNotNull(response.getData(), "Data should not be null");
    assertTrue(response.getData().size() > 0, "Should have entities");
    assertNotNull(response.getPaging(), "Paging info should be present");
  }

  /**
   * Test: SDK auto-pagination fluent API
   * Basic test - comprehensive pagination is in PaginationIT
   */
  @Test
  void testAutoPaginationFluentAPI(TestNamespace ns) {
    // Create a few entities
    for (int i = 0; i < 3; i++) {
      K createRequest = createRequest(ns.prefix("page" + i), ns);
      createEntity(createRequest);
    }

    // Basic pagination test - verify pagination works
    org.openmetadata.sdk.models.ListParams params = new org.openmetadata.sdk.models.ListParams();
    params.setLimit(2);

    org.openmetadata.sdk.models.ListResponse<T> page = listEntities(params);
    assertNotNull(page, "Page should not be null");
    assertNotNull(page.getPaging(), "Paging info should not be null");
    assertTrue(page.getData().size() <= 2, "Should respect limit");
  }

  /**
   * Test: SDK bulk fluent API
   * Equivalent to: testBulkFluentAPI in EntityResourceTest
   */
  @Test
  void testBulkFluentAPI(TestNamespace ns) {

    // Create multiple entities
    List<T> createdEntities = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      K createRequest = createRequest(ns.prefix("bulk_api_" + i), ns);
      T entity = createEntity(createRequest);
      createdEntities.add(entity);
    }

    // Verify all entities exist
    for (T created : createdEntities) {
      T fetched = getEntity(created.getId().toString());
      assertNotNull(fetched, "Bulk created entity should exist");
      assertEquals(created.getName(), fetched.getName(), "Names should match");
    }

    // Bulk update descriptions
    for (T entity : createdEntities) {
      T fetched = getEntity(entity.getId().toString());
      fetched.setDescription("Bulk updated - " + entity.getName());
      patchEntity(fetched.getId().toString(), fetched);
    }

    // Verify updates. Retry to absorb the cache write-through / pub-sub fan-out under parallel
    // load — the PATCH is synchronous server-side but concurrent test traffic can briefly stall
    // the fresh read of a just-updated row. 60s matches other eventual-consistency windows in
    // this test suite; NotificationTemplate showed the previous 10s budget hit 12s of stall.
    for (T entity : createdEntities) {
      String entityId = entity.getId().toString();
      Awaitility.await("Description should be bulk updated")
          .atMost(Duration.ofSeconds(60))
          .pollInterval(Duration.ofMillis(500))
          .untilAsserted(
              () -> {
                T refetched = getEntity(entityId);
                assertTrue(
                    refetched.getDescription() != null
                        && refetched.getDescription().startsWith("Bulk updated"),
                    "Description should be bulk updated");
              });
    }
  }

  // ===================================================================
  // VERSION HISTORY TESTS (Additional)
  // ===================================================================

  /**
   * Test: Get deleted entity version from version history
   * Equivalent to: get_deletedVersion in EntityResourceTest
   */
  @Test
  void get_deletedVersion(TestNamespace ns) {
    if (!supportsSoftDelete || !supportsPatch) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    Double previousVersion = entity.getVersion();

    // Update to create version history
    entity.setDescription("Updated before delete");
    T updated = patchEntity(entity.getId().toString(), entity);
    assertTrue(updated.getVersion() > previousVersion, "Version should increment");

    // Soft delete the entity
    deleteEntity(updated.getId().toString());

    // Get entity with include=deleted to verify it exists
    T deleted = getEntityIncludeDeleted(updated.getId().toString());
    assertNotNull(deleted, "Deleted entity should be retrievable");
    assertTrue(deleted.getDeleted(), "Entity should be marked as deleted");

    // Version should have incremented after delete
    assertTrue(
        deleted.getVersion() > updated.getVersion(), "Version should increment after delete");
  }

  // ===================================================================
  // SEARCH FLUENT API TESTS
  // ===================================================================

  /**
   * Test: Search fluent API
   * Equivalent to: testSearchFluentAPI in EntityResourceTest
   */
  @Test
  void testSearchFluentAPI(TestNamespace ns) {
    if (!supportsSearchIndex) return;

    // Create entity to ensure there's something to search
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    assertNotNull(entity, "Entity should be created for search test");

    try {
      // Test search fluent API
      org.openmetadata.sdk.api.Search.SearchResults results =
          org.openmetadata.sdk.api.Search.query("*")
              .in(getSearchIndex())
              .sortBy("name", org.openmetadata.sdk.api.Search.SortOrder.ASC)
              .limit(10)
              .execute();

      assertNotNull(results, "Search results should not be null");

      // Test suggest API
      org.openmetadata.sdk.api.Search.SuggestionResults suggestions =
          org.openmetadata.sdk.api.Search.suggest("test").in(getSearchIndex()).limit(5).execute();

      assertNotNull(suggestions, "Suggestions should not be null");

      // Test aggregation API
      org.openmetadata.sdk.api.Search.AggregationResults aggregations =
          org.openmetadata.sdk.api.Search.aggregate()
              .query("*")
              .in(getSearchIndex())
              .aggregateBy("tags.tagFQN")
              .execute();

      assertNotNull(aggregations, "Aggregations should not be null");

    } catch (Exception e) {
      // Search may fail if Elasticsearch is not properly configured
      // This is acceptable in some test environments
    }
  }

  protected String getSearchIndex() {
    return getSearchIndexName();
  }

  // ===================================================================
  // LINEAGE FLUENT API TESTS
  // ===================================================================

  protected boolean supportsLineage = false;

  /**
   * Test: Lineage fluent API
   * Equivalent to: testLineageFluentAPI in EntityResourceTest
   */
  @Test
  void testLineageFluentAPI(TestNamespace ns) {
    if (!supportsLineage) return;

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    assertNotNull(entity, "Entity should be created for lineage test");

    try {
      // Test lineage retrieval
      org.openmetadata.sdk.api.Lineage.LineageGraph lineage =
          org.openmetadata.sdk.api.Lineage.of(getEntityType(), entity.getId().toString())
              .upstream(1)
              .downstream(1)
              .includeDeleted(false)
              .fetch();

      assertNotNull(lineage, "Lineage graph should not be null");

    } catch (Exception e) {
      // Lineage retrieval may fail for newly created entities without edges
      // This is acceptable
    }
  }

  // ===================================================================
  // RECOGNIZER FEEDBACK TESTS (Additional)
  // ===================================================================

  /**
   * Test: Recognizer feedback for multiple entities
   * Equivalent to: test_recognizerFeedback_multipleEntities in EntityResourceTest
   */
  @Test
  void test_recognizerFeedback_multipleEntities(TestNamespace ns) {
    if (!supportsRecognizerFeedback || !supportsTags) return;
  }

  /**
   * Test: Invalid recognizer feedback
   * Equivalent to: test_recognizerFeedback_invalidFeedback in EntityResourceTest
   */
  @Test
  void test_recognizerFeedback_invalidFeedback(TestNamespace ns) {
    if (!supportsRecognizerFeedback || !supportsTags) return;
  }

  // ===================================================================
  // CONVERSATION CLEANUP TESTS
  // ===================================================================

  protected boolean supportsConversations = false;

  /**
   * Test: Cleanup conversations when entity is deleted
   * Equivalent to: test_cleanupConversations in EntityResourceTest
   */
  @Test
  void test_cleanupConversations(TestNamespace ns) {
    if (!supportsConversations) return;
  }

  // ===================================================================
  // SYSTEM ENTITY TESTS
  // ===================================================================

  /**
   * Test: System entities cannot be deleted
   * Equivalent to: delete_systemEntity in EntityResourceTest
   */
  @Test
  void delete_systemEntity(TestNamespace ns) {
    // System entities are pre-created and cannot be deleted
    // This test verifies the behavior when applicable
  }

  // ===================================================================
  // BULK API TESTS
  // Comprehensive coverage: CRUD, permissions, idempotency, versioning,
  // partial failure, async, empty, large batch, mixed create+update
  // ===================================================================

  /**
   * Test: Bulk create entities
   *
   * <p>Creates multiple entities using bulk API and validates all are created successfully.
   */
  @Test
  void test_bulkCreateOrUpdate(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 5; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_entity_" + i), ns));
    }

    BulkOperationResult result = executeBulkCreate(createRequests);

    assertNotNull(result);
    assertEquals(5, result.getNumberOfRowsProcessed());
    assertEquals(5, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
    assertEquals(ApiStatus.SUCCESS, result.getStatus());

    assertNotNull(result.getSuccessRequest());
    assertEquals(5, result.getSuccessRequest().size());

    for (BulkResponse bulkResponse : result.getSuccessRequest()) {
      String fqn = (String) bulkResponse.getRequest();
      assertNotNull(fqn);
      T retrieved = getEntityByName(fqn);
      assertNotNull(retrieved, "Entity should be retrievable by FQN: " + fqn);
    }
  }

  /**
   * Test: Bulk update existing entities
   *
   * <p>Creates entities via bulk, then re-sends same requests with updated descriptions.
   * Verifies description changes and version increments.
   */
  @Test
  void test_bulkUpdate_existingEntities(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_upd_", 3);

    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(3, createResult.getNumberOfRowsPassed());

    // Capture initial versions
    List<String> fqns = new ArrayList<>();
    List<Double> initialVersions = new ArrayList<>();
    for (BulkResponse resp : createResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      fqns.add(fqn);
      T entity = getEntityByName(fqn);
      initialVersions.add(entity.getVersion());
    }

    // Reuse same request objects with updated descriptions
    for (K req : createRequests) {
      setDescription(req, "Updated via bulk");
    }

    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(3, updateResult.getNumberOfRowsPassed());
    assertEquals(0, updateResult.getNumberOfRowsFailed());

    for (int i = 0; i < fqns.size(); i++) {
      T entity = getEntityByName(fqns.get(i));
      assertEquals("Updated via bulk", entity.getDescription());
      assertTrue(
          entity.getVersion() > initialVersions.get(i), "Version should increment after update");
    }
  }

  /**
   * Test: Bulk idempotency - calling bulk create twice should succeed (second call updates)
   */
  @Test
  void test_bulkCreateOrUpdate_idempotent(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_idem_", 3);

    BulkOperationResult first = executeBulkCreate(createRequests);
    assertEquals(3, first.getNumberOfRowsPassed());

    // Same request again should succeed (updates existing)
    BulkOperationResult second = executeBulkCreate(createRequests);
    assertEquals(3, second.getNumberOfRowsPassed());
    assertEquals(0, second.getNumberOfRowsFailed());
  }

  /**
   * Test: Bulk create with partial failure
   *
   * <p>Mix of valid and invalid entities verifies partial success handling.
   */
  @Test
  void test_bulkCreateOrUpdate_partialFailure(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_partial_" + i), ns));
    }

    K invalidRequest = createInvalidRequestForBulk(ns);
    if (invalidRequest != null) {
      createRequests.add(invalidRequest);
    }

    BulkOperationResult result = executeBulkCreate(createRequests);

    assertNotNull(result);
    assertTrue(result.getNumberOfRowsProcessed() >= 3);
    assertTrue(result.getNumberOfRowsPassed() >= 3);

    if (invalidRequest != null) {
      assertTrue(result.getNumberOfRowsFailed() >= 1);
      assertEquals(ApiStatus.PARTIAL_SUCCESS, result.getStatus());
    }
  }

  /**
   * Test: Async bulk create
   */
  @Test
  void test_bulkCreateOrUpdate_async(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 5; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_async_" + i), ns));
    }

    BulkOperationResult result = executeBulkCreateAsync(createRequests);

    assertNotNull(result);
    assertEquals(5, result.getNumberOfRowsProcessed());
    assertEquals(ApiStatus.SUCCESS, result.getStatus());
  }

  /**
   * Test: Bulk create with empty list returns empty result
   */
  @Test
  void test_bulkCreateOrUpdate_emptyList(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    BulkOperationResult result = executeBulkCreate(new ArrayList<>());

    assertNotNull(result);
    assertEquals(0, result.getNumberOfRowsProcessed());
    assertEquals(0, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  /**
   * Test: Bulk create large batch (50 entities)
   */
  @Test
  void test_bulkCreateOrUpdate_largeBatch(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 50; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_large_" + i), ns));
    }

    BulkOperationResult result = executeBulkCreate(createRequests);

    assertEquals(50, result.getNumberOfRowsProcessed());
    assertEquals(50, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  /**
   * Test: Mixed create + update in one bulk call
   *
   * <p>Creates some entities first, then sends a bulk request with a mix of new and existing.
   */
  @Test
  void test_bulkCreateOrUpdate_mixedCreateAndUpdate(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    // Create 4 requests sharing the same parent
    List<K> allRequests = createBulkRequests(ns, "bulk_mix_", 4);

    // Pre-create first 2
    List<K> preCreate = new ArrayList<>(allRequests.subList(0, 2));
    BulkOperationResult preResult = executeBulkCreate(preCreate);
    assertEquals(2, preResult.getNumberOfRowsPassed());

    // Update descriptions on the existing 2
    for (K req : preCreate) {
      setDescription(req, "Bulk updated");
    }

    // Mixed request: 2 existing (updates) + 2 new (creates)
    BulkOperationResult result = executeBulkCreate(allRequests);

    assertEquals(4, result.getNumberOfRowsProcessed());
    assertEquals(4, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());

    for (BulkResponse resp : result.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      T entity = getEntityByName(fqn);
      assertNotNull(entity);
    }
  }

  /**
   * Test: Verify all entities exist in DB after bulk create
   */
  @Test
  void test_bulkCreateOrUpdate_verifyEntitiesExist(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_verify_" + i), ns));
    }

    BulkOperationResult result = executeBulkCreate(createRequests);

    assertEquals(10, result.getNumberOfRowsPassed());
    assertNotNull(result.getSuccessRequest());
    assertEquals(10, result.getSuccessRequest().size());

    for (BulkResponse successResponse : result.getSuccessRequest()) {
      assertNotNull(successResponse.getRequest());
      assertEquals(200, successResponse.getStatus());
      T entity = getEntityByName((String) successResponse.getRequest());
      assertNotNull(entity, "Entity should exist: " + successResponse.getRequest());
      assertNotNull(entity.getId());
    }
  }

  /**
   * Test: Bulk update with tags.
   *
   * <p>Creates entities, then bulk-updates them with tags and verifies tags are applied.
   */
  @Test
  void test_bulkUpdate_tags(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsTags) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_tag_", 3);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(3, createResult.getNumberOfRowsPassed());

    SharedEntities shared = SharedEntities.get();
    for (K req : createRequests) {
      setFieldViaReflection(req, "setTags", List.class, List.of(shared.PII_SENSITIVE_TAG_LABEL));
    }

    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(3, updateResult.getNumberOfRowsPassed());
    assertEquals(0, updateResult.getNumberOfRowsFailed());

    for (BulkResponse resp : updateResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      T entity = getEntityByNameWithFields(fqn, "tags");
      assertNotNull(entity.getTags(), "Entity should have tags: " + fqn);
      assertFalse(entity.getTags().isEmpty(), "Tags should not be empty: " + fqn);
      assertTrue(
          entity.getTags().stream()
              .anyMatch(t -> t.getTagFQN().equals(shared.PII_SENSITIVE_TAG_LABEL.getTagFQN())),
          "Should contain the assigned tag: " + fqn);
    }
  }

  /**
   * Test: Bulk update with owners.
   *
   * <p>Creates entities, then bulk-updates them with an owner and verifies.
   */
  @Test
  void test_bulkUpdate_owners(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsOwners) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_own_", 3);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(3, createResult.getNumberOfRowsPassed());

    SharedEntities shared = SharedEntities.get();
    for (K req : createRequests) {
      setFieldViaReflection(req, "setOwners", List.class, List.of(shared.USER1_REF));
    }

    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(3, updateResult.getNumberOfRowsPassed());
    assertEquals(0, updateResult.getNumberOfRowsFailed());

    for (BulkResponse resp : updateResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      T entity = getEntityByNameWithFields(fqn, "owners");
      assertNotNull(entity.getOwners(), "Entity should have owners: " + fqn);
      assertFalse(entity.getOwners().isEmpty(), "Owners should not be empty: " + fqn);
      assertEquals(shared.USER1.getId(), entity.getOwners().get(0).getId());
    }
  }

  /**
   * Test: Bulk update with domain.
   *
   * <p>Creates entities, then bulk-updates them with a domain and verifies.
   */
  @Test
  void test_bulkUpdate_domain(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsDomains) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_dom_", 3);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(3, createResult.getNumberOfRowsPassed());

    SharedEntities shared = SharedEntities.get();
    for (K req : createRequests) {
      setFieldViaReflection(
          req, "setDomains", List.class, List.of(shared.DOMAIN.getFullyQualifiedName()));
    }

    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(3, updateResult.getNumberOfRowsPassed());
    assertEquals(0, updateResult.getNumberOfRowsFailed());

    for (BulkResponse resp : updateResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      T entity = getEntityByNameWithFields(fqn, "domains");
      assertNotNull(entity.getDomains(), "Entity should have domains: " + fqn);
      assertFalse(entity.getDomains().isEmpty(), "Domains should not be empty: " + fqn);
      assertEquals(shared.DOMAIN.getId(), entity.getDomains().get(0).getId());
    }
  }

  /**
   * Test: Bulk update with tags, owners, and domain all at once.
   */
  @Test
  void test_bulkUpdate_multipleFields(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsTags || !supportsOwners || !supportsDomains) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_multi_", 3);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(3, createResult.getNumberOfRowsPassed());

    SharedEntities shared = SharedEntities.get();
    for (K req : createRequests) {
      setDescription(req, "Multi-field bulk update");
      setFieldViaReflection(req, "setTags", List.class, List.of(shared.PII_SENSITIVE_TAG_LABEL));
      setFieldViaReflection(req, "setOwners", List.class, List.of(shared.USER1_REF));
      setFieldViaReflection(
          req, "setDomains", List.class, List.of(shared.DOMAIN.getFullyQualifiedName()));
    }

    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(3, updateResult.getNumberOfRowsPassed());
    assertEquals(0, updateResult.getNumberOfRowsFailed());

    for (BulkResponse resp : updateResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      T entity = getEntityByNameWithFields(fqn, "tags,owners,domains");

      assertEquals("Multi-field bulk update", entity.getDescription());

      assertNotNull(entity.getTags());
      assertFalse(entity.getTags().isEmpty());

      assertNotNull(entity.getOwners());
      assertFalse(entity.getOwners().isEmpty());
      assertEquals(shared.USER1.getId(), entity.getOwners().get(0).getId());

      assertNotNull(entity.getDomains());
      assertFalse(entity.getDomains().isEmpty());
      assertEquals(shared.DOMAIN.getId(), entity.getDomains().get(0).getId());
    }
  }

  /**
   * Test: Bulk update version history is preserved.
   *
   * <p>Creates entities, updates them, verifies version history has multiple entries.
   */
  @Test
  void test_bulkUpdate_versionHistory(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsVersionHistory) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_ver_", 2);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(2, createResult.getNumberOfRowsPassed());

    List<String> fqns = new ArrayList<>();
    List<Double> initialVersions = new ArrayList<>();
    for (BulkResponse resp : createResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      fqns.add(fqn);
      T entity = getEntityByName(fqn);
      initialVersions.add(entity.getVersion());
    }

    for (K req : createRequests) {
      setDescription(req, "Version history test update");
    }
    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(2, updateResult.getNumberOfRowsPassed());

    for (int i = 0; i < fqns.size(); i++) {
      T entity = getEntityByName(fqns.get(i));
      assertTrue(
          entity.getVersion() > initialVersions.get(i),
          "Version should increment after bulk update");
      assertNotNull(entity.getChangeDescription(), "Should have change description");
    }
  }

  /**
   * Test: Bulk re-submit with no changes does not increment version.
   */
  @Test
  void test_bulkUpdate_noChangeSameVersion(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_noop_", 2);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(2, createResult.getNumberOfRowsPassed());

    List<String> fqns = new ArrayList<>();
    List<Double> versions = new ArrayList<>();
    for (BulkResponse resp : createResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      fqns.add(fqn);
      T entity = getEntityByName(fqn);
      versions.add(entity.getVersion());
    }

    // Re-submit identical data
    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(2, updateResult.getNumberOfRowsPassed());

    for (int i = 0; i < fqns.size(); i++) {
      T entity = getEntityByName(fqns.get(i));
      assertEquals(
          versions.get(i),
          entity.getVersion(),
          "Version should NOT increment when no fields changed: " + fqns.get(i));
    }
  }

  /**
   * Test: Large batch bulk update (50 entities).
   */
  @Test
  void test_bulkUpdate_largeBatch(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_lg_", 50);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(50, createResult.getNumberOfRowsPassed());

    for (K req : createRequests) {
      setDescription(req, "Large batch updated");
    }

    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(50, updateResult.getNumberOfRowsPassed());
    assertEquals(0, updateResult.getNumberOfRowsFailed());

    // Spot-check a few
    List<BulkResponse> successes = updateResult.getSuccessRequest();
    for (int idx : List.of(0, 24, 49)) {
      String fqn = (String) successes.get(idx).getRequest();
      T entity = getEntityByName(fqn);
      assertEquals("Large batch updated", entity.getDescription());
    }
  }

  /**
   * Test: User-added tags are preserved when ingestion re-runs with description changes.
   *
   * <p>Simulates the real-world scenario: a user adds tags to entities, then the ingestion
   * connector runs again bringing description changes. The tags must NOT be overwritten
   * since PUT from bots doesn't overwrite user-provided fields like tags.
   */
  @Test
  void test_bulkUpdate_preservesUserTagsOnReIngestion(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsTags) return;

    // Step 1: Ingestion creates entities (no tags)
    List<K> createRequests = createBulkRequests(ns, "bulk_preserve_", 3);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(3, createResult.getNumberOfRowsPassed());

    // Step 2: User adds tags to entities via PATCH
    SharedEntities shared = SharedEntities.get();
    List<String> fqns = new ArrayList<>();
    for (BulkResponse resp : createResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      fqns.add(fqn);
      T entity = getEntityByNameWithFields(fqn, "tags");
      entity.setTags(List.of(shared.PII_SENSITIVE_TAG_LABEL));
      patchEntity(entity.getId().toString(), entity);
    }

    // Verify tags were applied
    for (String fqn : fqns) {
      T entity = getEntityByNameWithFields(fqn, "tags");
      assertNotNull(entity.getTags());
      assertFalse(entity.getTags().isEmpty(), "Tags should be present after PATCH: " + fqn);
    }

    // Step 3: Ingestion re-runs — bulk update with only description changes (no tags in request)
    for (K req : createRequests) {
      setDescription(req, "Re-ingested description");
    }
    BulkOperationResult reIngestionResult = executeBulkCreate(createRequests);
    assertEquals(3, reIngestionResult.getNumberOfRowsPassed());

    // Step 4: Verify tags are still present after re-ingestion
    for (String fqn : fqns) {
      T entity = getEntityByNameWithFields(fqn, "tags");
      assertEquals("Re-ingested description", entity.getDescription());
      assertNotNull(entity.getTags(), "Tags should still be present after re-ingestion: " + fqn);
      assertFalse(entity.getTags().isEmpty(), "Tags should NOT be cleared by re-ingestion: " + fqn);
      assertTrue(
          entity.getTags().stream()
              .anyMatch(t -> t.getTagFQN().equals(shared.PII_SENSITIVE_TAG_LABEL.getTagFQN())),
          "Original user-added tag should be preserved: " + fqn);
    }
  }

  /**
   * Test: User-added owners are preserved when ingestion re-runs with description changes.
   */
  @Test
  void test_bulkUpdate_preservesUserOwnersOnReIngestion(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsOwners) return;

    // Step 1: Ingestion creates entities (no owners)
    List<K> createRequests = createBulkRequests(ns, "bulk_own_pres_", 2);
    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(2, createResult.getNumberOfRowsPassed());

    // Step 2: User adds owners via PATCH
    SharedEntities shared = SharedEntities.get();
    List<String> fqns = new ArrayList<>();
    for (BulkResponse resp : createResult.getSuccessRequest()) {
      String fqn = (String) resp.getRequest();
      fqns.add(fqn);
      T entity = getEntityByNameWithFields(fqn, "owners");
      entity.setOwners(List.of(shared.USER1_REF));
      patchEntity(entity.getId().toString(), entity);
    }

    // Step 3: Re-ingestion with only description changes
    for (K req : createRequests) {
      setDescription(req, "Re-ingested with owners");
    }
    BulkOperationResult reIngestionResult = executeBulkCreate(createRequests);
    assertEquals(2, reIngestionResult.getNumberOfRowsPassed());

    // Step 4: Verify owners preserved
    for (String fqn : fqns) {
      T entity = getEntityByNameWithFields(fqn, "owners");
      assertEquals("Re-ingested with owners", entity.getDescription());
      assertNotNull(entity.getOwners(), "Owners should be preserved: " + fqn);
      assertFalse(entity.getOwners().isEmpty(), "Owners should NOT be cleared: " + fqn);
      assertEquals(shared.USER1.getId(), entity.getOwners().get(0).getId());
    }
  }

  /**
   * Test: Verify entities are indexed in search after bulk create.
   *
   * <p>Uses Awaitility to poll the search API until all bulk-created entities appear in the index.
   */
  @Test
  void test_bulkCreateOrUpdate_searchIndexed(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsSearchIndex) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 5; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_search_" + i), ns));
    }

    BulkOperationResult result = executeBulkCreate(createRequests);
    assertEquals(5, result.getNumberOfRowsPassed());

    List<String> expectedFqns = new ArrayList<>();
    for (BulkResponse resp : result.getSuccessRequest()) {
      expectedFqns.add((String) resp.getRequest());
    }

    String searchIndex = getSearchIndexName();
    OpenMetadataClient client = SdkClients.adminClient();

    Awaitility.await()
        .atMost(Duration.ofSeconds(180))
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(2))
        .ignoreExceptions()
        .untilAsserted(
            () -> {
              for (String fqn : expectedFqns) {
                String queryFilter =
                    String.format(
                        "{\"query\":{\"bool\":{\"must\":[{\"term\":{\"fullyQualifiedName\":\"%s\"}}]}}}",
                        fqn);
                String searchResponse =
                    client
                        .search()
                        .query("*")
                        .index(searchIndex)
                        .queryFilter(queryFilter)
                        .size(1)
                        .execute();

                assertNotNull(searchResponse, "Search response should not be null");
                JsonNode root = MAPPER.readTree(searchResponse);
                assertTrue(root.has("hits"), "Search response should have hits");

                JsonNode hits = root.get("hits").get("hits");
                assertTrue(hits.size() > 0, "Entity should be found in search index: " + fqn);
                assertEquals(fqn, hits.get(0).get("_source").get("fullyQualifiedName").asText());
              }
            });
  }

  /**
   * Test: Verify search index is updated after bulk update.
   *
   * <p>Creates entities, updates descriptions via bulk, then verifies search returns updated data.
   */
  @Test
  void test_bulkUpdate_searchIndexUpdated(TestNamespace ns) {
    if (!supportsBulkAPI || !supportsSearchIndex) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_search_upd_", 3);

    BulkOperationResult createResult = executeBulkCreate(createRequests);
    assertEquals(3, createResult.getNumberOfRowsPassed());

    List<String> fqns = new ArrayList<>();
    for (BulkResponse resp : createResult.getSuccessRequest()) {
      fqns.add((String) resp.getRequest());
    }

    String updatedDesc = "SearchUpdated_" + System.currentTimeMillis();
    for (K req : createRequests) {
      setDescription(req, updatedDesc);
    }

    BulkOperationResult updateResult = executeBulkCreate(createRequests);
    assertEquals(3, updateResult.getNumberOfRowsPassed());

    String searchIndex = getSearchIndexName();
    OpenMetadataClient client = SdkClients.adminClient();

    Awaitility.await()
        .atMost(Duration.ofSeconds(180))
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(3))
        .ignoreExceptions()
        .untilAsserted(
            () -> {
              for (String fqn : fqns) {
                String queryFilter =
                    String.format(
                        "{\"query\":{\"bool\":{\"must\":[{\"term\":{\"fullyQualifiedName\":\"%s\"}}]}}}",
                        fqn);
                String searchResponse =
                    client
                        .search()
                        .query("*")
                        .index(searchIndex)
                        .queryFilter(queryFilter)
                        .size(1)
                        .execute();

                JsonNode root = MAPPER.readTree(searchResponse);
                JsonNode hits = root.get("hits").get("hits");
                assertTrue(hits.size() > 0, "Entity should be found in search index: " + fqn);
                assertEquals(
                    updatedDesc,
                    hits.get(0).get("_source").path("description").asText(),
                    "Search should return updated description for: " + fqn);
              }
            });
  }

  // ===================================================================
  // BULK API PERMISSION TESTS
  // Tests authorization enforcement for admin, bot, and restricted users
  // ===================================================================

  /**
   * Test: Admin can bulk create entities
   */
  @Test
  void test_bulkCreate_adminSuccess(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_admin_" + i), ns));
    }

    HttpResponse<String> response =
        callBulkEndpoint(createRequests, SdkClients.getAdminToken(), false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(3, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  /**
   * Test: Bot/ingestion user can bulk create entities
   */
  @Test
  void test_bulkCreate_botSuccess(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_bot_" + i), ns));
    }

    String botToken = getBotToken();
    HttpResponse<String> response = callBulkEndpoint(createRequests, botToken, false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(3, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  /**
   * Test: Unauthenticated request returns 401
   */
  @Test
  void test_bulkCreate_noAuth_returns401(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    createRequests.add(createRequest(ns.prefix("bulk_noauth"), ns));

    String url = SdkClients.getServerUrl() + getResourcePath() + "bulk";
    java.net.http.HttpRequest request =
        java.net.http.HttpRequest.newBuilder()
            .uri(java.net.URI.create(url))
            .header("Content-Type", "application/json")
            .PUT(
                java.net.http.HttpRequest.BodyPublishers.ofString(
                    JsonUtils.pojoToJson(createRequests)))
            .build();

    HttpResponse<String> response =
        java.net.http.HttpClient.newHttpClient()
            .send(request, HttpResponse.BodyHandlers.ofString());

    assertEquals(401, response.statusCode());
  }

  /**
   * Test: DataConsumer cannot bulk create entities
   */
  @Test
  void test_bulkCreate_dataConsumer_denied(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    createRequests.add(createRequest(ns.prefix("bulk_consumer_create"), ns));

    String consumerToken = getDataConsumerToken();
    HttpResponse<String> response = callBulkEndpoint(createRequests, consumerToken, false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(0, result.getNumberOfRowsPassed());
    assertEquals(1, result.getNumberOfRowsFailed());
    assertNotNull(result.getFailedRequest());
    assertTrue(
        result.getFailedRequest().get(0).getStatus() == 403
            || result.getFailedRequest().get(0).getStatus() == 400);
  }

  /**
   * Test: DataConsumer cannot bulk update existing entities
   */
  @Test
  void test_bulkUpdate_dataConsumer_denied(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    // Create entity as admin (reuse same request object for update)
    List<K> createRequests = new ArrayList<>();
    createRequests.add(createRequest(ns.prefix("bulk_consumer_upd"), ns));

    HttpResponse<String> createResponse =
        callBulkEndpoint(createRequests, SdkClients.getAdminToken(), false);
    assertEquals(200, createResponse.statusCode());
    BulkOperationResult createResult =
        JsonUtils.readValue(createResponse.body(), BulkOperationResult.class);
    assertEquals(1, createResult.getNumberOfRowsPassed());

    // Attempt update as DataConsumer using same request with changed description
    setDescription(createRequests.get(0), "Consumer tried to update");

    String consumerToken = getDataConsumerToken();
    HttpResponse<String> response = callBulkEndpoint(createRequests, consumerToken, false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(0, result.getNumberOfRowsPassed());
    assertEquals(1, result.getNumberOfRowsFailed());
  }

  /**
   * Test: Bot/ingestion user can bulk update existing entities
   */
  @Test
  void test_bulkUpdate_botSuccess(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    // Create entity as admin
    List<K> createRequests = new ArrayList<>();
    createRequests.add(createRequest(ns.prefix("bulk_bot_upd"), ns));

    HttpResponse<String> createResponse =
        callBulkEndpoint(createRequests, SdkClients.getAdminToken(), false);
    assertEquals(200, createResponse.statusCode());

    // Update as bot using same request with changed description
    setDescription(createRequests.get(0), "Updated by bot");

    String botToken = getBotToken();
    HttpResponse<String> response = callBulkEndpoint(createRequests, botToken, false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(1, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  /**
   * Test: Async bulk returns 202 Accepted
   */
  @Test
  void test_bulkAsync_returns202(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    createRequests.add(createRequest(ns.prefix("bulk_async_perm"), ns));

    HttpResponse<String> response =
        callBulkEndpoint(createRequests, SdkClients.getAdminToken(), true);

    assertEquals(202, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertNotNull(result.getNumberOfRowsProcessed());
  }

  /**
   * Test: Bulk permission denied returns per-entity failure status
   */
  @Test
  void test_bulkCreate_permissionDenied_returnsFailureStatus(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    createRequests.add(createRequest(ns.prefix("bulk_perm_status"), ns));

    String consumerToken = getDataConsumerToken();
    HttpResponse<String> response = callBulkEndpoint(createRequests, consumerToken, false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(1, result.getNumberOfRowsFailed());

    BulkResponse failedRequest = result.getFailedRequest().get(0);
    assertTrue(
        failedRequest.getStatus() == 403 || failedRequest.getStatus() == 400,
        "Permission denied should return 4xx status, got: " + failedRequest.getStatus());
    assertNotNull(failedRequest.getMessage(), "Failed request should have an error message");
  }

  /**
   * Test: Admin can bulk update with mixed permissions (multiple entities, all succeed)
   */
  @Test
  void test_bulkCreate_admin_multipleBatch(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    for (int i = 0; i < 20; i++) {
      createRequests.add(createRequest(ns.prefix("bulk_admin_batch_" + i), ns));
    }

    HttpResponse<String> response =
        callBulkEndpoint(createRequests, SdkClients.getAdminToken(), false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(20, result.getNumberOfRowsProcessed());
    assertEquals(20, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  // ===================================================================
  // BULK API EDGE CASE TESTS
  // ===================================================================

  /**
   * Test: Duplicate FQNs in a single batch request.
   *
   * <p>Sends the same entity name twice in one request. The API should handle gracefully —
   * either dedup or process both (second becomes an update).
   */
  @Test
  void test_bulkCreateOrUpdate_duplicateFqnsInBatch(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_dup_", 1);
    K original = createRequests.get(0);

    // Add the same request again (duplicate FQN)
    createRequests.add(original);

    BulkOperationResult result = executeBulkCreate(createRequests);

    assertEquals(2, result.getNumberOfRowsProcessed());
    // Both should succeed: first creates, second updates
    assertEquals(2, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  @Test
  void test_bulkCreateOrUpdate_tripleDuplicateFqnsInBatch(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    List<K> createRequests = createBulkRequests(ns, "bulk_trip_", 1);
    K original = createRequests.get(0);

    // Add the same request two more times (triple duplicate FQN)
    createRequests.add(original);
    createRequests.add(original);

    BulkOperationResult result = executeBulkCreate(createRequests);

    assertEquals(3, result.getNumberOfRowsProcessed());
    assertEquals(3, result.getNumberOfRowsPassed());
    assertEquals(0, result.getNumberOfRowsFailed());
  }

  /**
   * Test: Concurrent bulk requests do not corrupt data.
   *
   * <p>Sends two bulk requests in parallel and verifies all entities are created.
   */
  @Test
  void test_bulkCreateOrUpdate_concurrent(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> batch1 = createBulkRequests(ns, "bulk_conc_a_", 5);
    List<K> batch2 = createBulkRequests(ns, "bulk_conc_b_", 5);

    String adminToken = SdkClients.getAdminToken();

    java.util.concurrent.ExecutorService executor =
        java.util.concurrent.Executors.newFixedThreadPool(2);
    java.util.concurrent.Future<HttpResponse<String>> future1 =
        executor.submit(() -> callBulkEndpoint(batch1, adminToken, false));
    java.util.concurrent.Future<HttpResponse<String>> future2 =
        executor.submit(() -> callBulkEndpoint(batch2, adminToken, false));

    HttpResponse<String> resp1 = future1.get();
    HttpResponse<String> resp2 = future2.get();
    executor.shutdown();

    assertEquals(200, resp1.statusCode());
    assertEquals(200, resp2.statusCode());

    BulkOperationResult result1 = JsonUtils.readValue(resp1.body(), BulkOperationResult.class);
    BulkOperationResult result2 = JsonUtils.readValue(resp2.body(), BulkOperationResult.class);

    assertEquals(5, result1.getNumberOfRowsPassed());
    assertEquals(5, result2.getNumberOfRowsPassed());
  }

  /**
   * Test: Data consumer receives 200 with FAILURE when all entities fail auth.
   *
   * <p>Verifies that when ALL entities fail authorization, the response reports complete failure
   * rather than silently dropping the entities.
   */
  @Test
  void test_bulkCreate_allAuthFailed_reportsFailure(TestNamespace ns) throws Exception {
    if (!supportsBulkAPI) return;

    List<K> createRequests = new ArrayList<>();
    createRequests.add(createRequest(ns.prefix("bulk_auth_fail_0"), ns));
    createRequests.add(createRequest(ns.prefix("bulk_auth_fail_1"), ns));

    String consumerToken = getDataConsumerToken();
    HttpResponse<String> response = callBulkEndpoint(createRequests, consumerToken, false);

    assertEquals(200, response.statusCode());

    BulkOperationResult result = JsonUtils.readValue(response.body(), BulkOperationResult.class);
    assertEquals(ApiStatus.FAILURE, result.getStatus());
    assertEquals(2, result.getNumberOfRowsProcessed());
    assertEquals(2, result.getNumberOfRowsFailed());
    assertEquals(0, result.getNumberOfRowsPassed());
    assertNotNull(result.getFailedRequest());
    assertEquals(2, result.getFailedRequest().size());
  }

  /**
   * Test: Bulk request with empty names are rejected.
   */
  @Test
  void test_bulkCreateOrUpdate_invalidEntitiesRejected(TestNamespace ns) {
    if (!supportsBulkAPI) return;

    K invalidRequest = createInvalidRequestForBulk(ns);
    if (invalidRequest == null) return;

    List<K> createRequests = new ArrayList<>();
    createRequests.add(invalidRequest);
    createRequests.add(createRequest(ns.prefix("bulk_valid_alongside"), ns));

    BulkOperationResult result = executeBulkCreate(createRequests);

    assertEquals(2, result.getNumberOfRowsProcessed());
    assertTrue(result.getNumberOfRowsFailed() > 0, "Invalid entity should fail");
    assertTrue(result.getNumberOfRowsPassed() > 0, "Valid entity should succeed");
    assertEquals(
        ApiStatus.PARTIAL_SUCCESS,
        result.getStatus(),
        "Mix of valid+invalid should be PARTIAL_SUCCESS");
  }

  // ===================================================================
  // BULK API HOOK METHODS
  // Subclasses that support bulk API should override these methods.
  // ===================================================================

  protected BulkOperationResult executeBulkCreate(List<K> createRequests) {
    throw new UnsupportedOperationException(
        "Bulk API not implemented for " + getEntityType() + ". Override executeBulkCreate()");
  }

  protected BulkOperationResult executeBulkCreateAsync(List<K> createRequests) {
    throw new UnsupportedOperationException(
        "Async Bulk API not implemented for "
            + getEntityType()
            + ". Override executeBulkCreateAsync()");
  }

  protected K createInvalidRequestForBulk(TestNamespace ns) {
    return null;
  }

  /**
   * Create multiple bulk requests sharing the same parent container.
   * Default implementation calls createRequest individually.
   * Subclasses should override for entities where createRequest creates a new parent
   * each time (e.g., Tables need all requests under the same schema).
   */
  protected List<K> createBulkRequests(TestNamespace ns, String prefix, int count) {
    List<K> requests = new ArrayList<>();
    for (int i = 0; i < count; i++) {
      requests.add(createRequest(ns.prefix(prefix + i), ns));
    }
    return requests;
  }

  private String getBotToken() {
    return org.openmetadata.it.auth.JwtAuthProvider.tokenFor(
        "ingestion-bot@open-metadata.org",
        "ingestion-bot@open-metadata.org",
        new String[] {"bot"},
        3600);
  }

  private String getDataConsumerToken() {
    return org.openmetadata.it.auth.JwtAuthProvider.tokenFor(
        "data-consumer@open-metadata.org",
        "data-consumer@open-metadata.org",
        new String[] {"DataConsumer"},
        3600);
  }

  private HttpResponse<String> callBulkEndpoint(List<K> requests, String authToken, boolean async)
      throws Exception {
    String url = SdkClients.getServerUrl() + getResourcePath() + "bulk";
    if (async) {
      url += "?async=true";
    }

    java.net.http.HttpRequest httpRequest =
        java.net.http.HttpRequest.newBuilder()
            .uri(java.net.URI.create(url))
            .header("Authorization", "Bearer " + authToken)
            .header("Content-Type", "application/json")
            .PUT(java.net.http.HttpRequest.BodyPublishers.ofString(JsonUtils.pojoToJson(requests)))
            .build();

    return java.net.http.HttpClient.newHttpClient()
        .send(httpRequest, HttpResponse.BodyHandlers.ofString());
  }

  // ===================================================================
  // SDK-ONLY CRUD TESTS
  // Equivalent to: test_sdkOnlyCreateRetrieveUpdate, test_sdkOnlyRetrieveByName,
  //                test_sdkOnlyListEntities, test_sdkOnlyAsyncOperations in EntityResourceTest
  // ===================================================================

  /**
   * Test: SDK-only create, retrieve, and update operations
   * Equivalent to: test_sdkOnlyCreateRetrieveUpdate in EntityResourceTest
   */
  @Test
  void test_sdkOnlyCreateRetrieveUpdate(TestNamespace ns) {
    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    assertNotNull(created, "Created entity should not be null");
    assertNotNull(created.getId(), "Created entity should have ID");
    assertEquals(0.1, created.getVersion(), 0.001, "Initial version should be 0.1");

    // Retrieve by ID
    T retrieved = getEntity(created.getId().toString());
    assertNotNull(retrieved, "Retrieved entity should not be null");
    assertEquals(created.getId(), retrieved.getId(), "IDs should match");

    // Update via patch
    if (supportsPatch) {
      retrieved.setDescription("SDK-only updated description");
      T updated = patchEntity(retrieved.getId().toString(), retrieved);

      assertNotNull(updated, "Updated entity should not be null");
      assertEquals(
          "SDK-only updated description",
          updated.getDescription(),
          "Description should be updated");
      assertTrue(updated.getVersion() > 0.1, "Version should increment after update");
    }
  }

  /**
   * Test: SDK-only retrieve by name
   * Equivalent to: test_sdkOnlyRetrieveByName in EntityResourceTest
   */
  @Test
  void test_sdkOnlyRetrieveByName(TestNamespace ns) {
    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    // Retrieve by name
    T retrieved = getEntityByName(created.getFullyQualifiedName());
    assertNotNull(retrieved, "Entity should be retrievable by FQN");
    assertEquals(created.getId(), retrieved.getId(), "IDs should match");
    assertEquals(
        created.getFullyQualifiedName(), retrieved.getFullyQualifiedName(), "FQNs should match");
  }

  /**
   * Test: SDK-only list entities
   * Basic test - comprehensive pagination testing is in PaginationIT
   */
  @Test
  void test_sdkOnlyListEntities(TestNamespace ns) {
    // Create a few entities
    for (int i = 0; i < 3; i++) {
      K createRequest = createRequest(ns.prefix("sdk_list_" + i), ns);
      createEntity(createRequest);
    }

    // Basic list test
    org.openmetadata.sdk.models.ListParams params = new org.openmetadata.sdk.models.ListParams();
    params.setLimit(10);
    org.openmetadata.sdk.models.ListResponse<T> response = listEntities(params);

    assertNotNull(response, "List response should not be null");
    assertNotNull(response.getData(), "List data should not be null");
    assertTrue(response.getData().size() > 0, "Should have entities in list");
  }

  // ===================================================================
  // FIELD QUERY PARAM TESTS
  // Equivalent to: get_entityWithDifferentFieldsQueryParam in EntityResourceTest
  // ===================================================================

  /**
   * Test: Get entity with different fields query parameters
   * Equivalent to: get_entityWithDifferentFieldsQueryParam in EntityResourceTest
   */
  @Test
  void get_entityWithDifferentFieldsQueryParam(TestNamespace ns) {
    if (!supportsFieldsQueryParam) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Build fields string based on what entity supports
    List<String> fields = new ArrayList<>();
    if (supportsOwners) fields.add("owners");
    if (supportsTags) fields.add("tags");
    if (supportsDomains) fields.add("domains");
    if (supportsFollowers) fields.add("followers");

    String fieldsStr = String.join(",", fields);
    if (fieldsStr.isEmpty()) {
      return; // Skip if entity doesn't support any of these fields
    }

    // Get entity by ID with fields
    T retrieved = getEntityWithFields(entity.getId().toString(), fieldsStr);
    assertNotNull(retrieved, "Entity should be retrievable with fields");
    assertEquals(entity.getId(), retrieved.getId(), "IDs should match");

    // Get entity by name with fields - use a subset of fields
    String nameFields = (supportsOwners ? "owners" : "") + (supportsTags ? ",tags" : "");
    nameFields = nameFields.startsWith(",") ? nameFields.substring(1) : nameFields;
    if (!nameFields.isEmpty()) {
      T retrievedByName = getEntityByNameWithFields(entity.getFullyQualifiedName(), nameFields);
      assertNotNull(retrievedByName, "Entity should be retrievable by name with fields");
      assertEquals(entity.getId(), retrievedByName.getId(), "IDs should match");
    }
  }

  // ===================================================================
  // INVALID LIST TESTS
  // Equivalent to: testInvalidEntityList in EntityResourceTest
  // ===================================================================

  /**
   * Test: List entities with invalid parameters
   * Equivalent to: testInvalidEntityList in EntityResourceTest
   */
  @Test
  void testInvalidEntityList(TestNamespace ns) {
    // Test with invalid limit (negative)
    org.openmetadata.sdk.models.ListParams invalidParams =
        new org.openmetadata.sdk.models.ListParams();
    invalidParams.setLimit(-1);

    assertThrows(
        Exception.class,
        () -> listEntities(invalidParams),
        "Listing with negative limit should fail");
  }

  // ===================================================================
  // ASYNC OPERATIONS TESTS
  // Equivalent to: test_sdkOnlyAsyncOperations in EntityResourceTest
  // ===================================================================

  /**
   * Test: SDK-only async operations (soft delete and restore)
   * Equivalent to: test_sdkOnlyAsyncOperations in EntityResourceTest
   */
  @Test
  void test_sdkOnlyAsyncOperations(TestNamespace ns) {
    // Create entity
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);
    assertNotNull(created, "Created entity should not be null");

    // Async delete (soft delete)
    if (supportsSoftDelete) {
      deleteEntity(created.getId().toString());

      // Verify entity is soft deleted
      assertThrows(
          Exception.class,
          () -> getEntity(created.getId().toString()),
          "Soft deleted entity should not be directly retrievable");

      // Restore entity
      restoreEntity(created.getId().toString());

      // Verify entity is restored
      T restored = getEntity(created.getId().toString());
      assertNotNull(restored, "Restored entity should be retrievable");
      assertEquals(created.getId(), restored.getId(), "IDs should match after restore");
    }
  }

  // ===================================================================
  // SEARCH INDEX TESTS
  // Equivalent to: checkCreatedEntity, checkDeletedEntity, checkIndexCreated,
  //                deleteTagAndCheckRelationshipsInSearch, updateDescriptionAndCheckInSearch
  // ===================================================================

  /**
   * Test: Verify entity is indexed in search after creation
   * Equivalent to: checkCreatedEntity in EntityResourceTest
   */
  @Test
  void checkCreatedEntity(TestNamespace ns) throws Exception {
    Assumptions.assumeTrue(supportsSearchIndex);

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Poll until entity appears in search index (async indexing may take time)
    Awaitility.await("Wait for entity to appear in search index")
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(2))
        .atMost(Duration.ofSeconds(180))
        .ignoreExceptions()
        .untilAsserted(
            () -> {
              String searchResponse = searchForEntity(entity.getId().toString());
              assertNotNull(searchResponse, "Search response should not be null");
              assertTrue(
                  searchResponse.contains(entity.getId().toString()),
                  "Entity should be present in search index");
            });
  }

  /**
   * Test: Verify entity is removed from search after deletion
   * Equivalent to: checkDeletedEntity in EntityResourceTest
   */
  @Test
  void checkDeletedEntity(TestNamespace ns) throws Exception {
    Assumptions.assumeTrue(supportsSearchIndex);
    Assumptions.assumeTrue(supportsSoftDelete);

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Poll until entity appears in search index before delete
    Awaitility.await("Wait for entity to appear in search index")
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(1))
        .atMost(Duration.ofSeconds(180))
        .ignoreExceptions()
        .untilAsserted(
            () -> {
              String searchResponse = searchForEntity(entity.getId().toString());
              assertNotNull(searchResponse, "Search response should not be null");
              assertTrue(
                  searchResponse.contains(entity.getId().toString()),
                  "Entity should be present in search index before delete");
            });

    // Delete entity
    deleteEntity(entity.getId().toString());

    // Verify entity is no longer in search (or marked deleted)
    // After soft delete, entity may still be in index but marked as deleted
    // This is acceptable behavior - the key is the delete operation succeeded
    String searchAfterDelete = searchForEntity(entity.getId().toString());
    assertNotNull(searchAfterDelete, "Search should still work after delete");
  }

  /**
   * Test: Verify search index exists for entity type Equivalent to: checkIndexCreated in
   * EntityResourceTest
   */
  @Test
  void checkIndexCreated(TestNamespace ns) throws Exception {
    Assumptions.assumeTrue(supportsSearchIndex);

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Poll until entity appears in search index
    Awaitility.await("Wait for entity to appear in search index")
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(1))
        .atMost(Duration.ofSeconds(180))
        .ignoreExceptions()
        .untilAsserted(
            () -> {
              String searchResponse = searchForEntity(entity.getId().toString());
              assertNotNull(searchResponse, "Search response should not be null");
              assertTrue(
                  searchResponse.contains(entity.getId().toString()),
                  "Entity should be present in search index");
            });
  }

  /**
   * Test: Update description and verify change is reflected in search Equivalent to:
   * updateDescriptionAndCheckInSearch in EntityResourceTest
   */
  @Test
  void updateDescriptionAndCheckInSearch(TestNamespace ns) throws Exception {
    Assumptions.assumeTrue(supportsSearchIndex);
    Assumptions.assumeTrue(supportsPatch);

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // First wait for entity to appear in search index
    Awaitility.await("Wait for entity to appear in search index")
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(1))
        .atMost(Duration.ofSeconds(180))
        .ignoreExceptions()
        .untilAsserted(
            () -> {
              String searchResponse = searchForEntity(entity.getId().toString());
              assertNotNull(searchResponse, "Search response should not be null");
              assertTrue(
                  searchResponse.contains(entity.getId().toString()),
                  "Entity should be present in search index");
            });

    String newDescription = "Updated description for search test " + UUID.randomUUID();
    entity.setDescription(newDescription);
    T updated = patchEntity(entity.getId().toString(), entity);

    // Wait for updated entity to be reflected in search
    Awaitility.await("Wait for search to reflect update")
        .pollDelay(Duration.ofMillis(500))
        .pollInterval(Duration.ofSeconds(1))
        .atMost(Duration.ofSeconds(180))
        .ignoreExceptions()
        .untilAsserted(
            () -> {
              String searchResponse = searchForEntity(updated.getId().toString());
              assertNotNull(searchResponse, "Search response should not be null");
              assertTrue(
                  searchResponse.contains(newDescription),
                  "Updated description should be in search index");
            });
  }

  /**
   * Test: Delete a tag and verify entity can still be retrieved. Equivalent to:
   * deleteTagAndCheckRelationshipsInSearch in EntityResourceTest
   *
   * <p>This test verifies that when a tag is deleted, the entity that had that tag can still be
   * retrieved and the tag is removed from the entity.
   */
  @Test
  void deleteTagAndCheckRelationshipsInSearch(TestNamespace ns) throws Exception {
    Assumptions.assumeTrue(supportsTags);
    Assumptions.assumeTrue(supportsPatch);

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    OpenMetadataClient client = SdkClients.adminClient();
    String tagName = ns.prefix("searchRelTag");
    String classificationName = ns.prefix("searchRelClassification");

    org.openmetadata.schema.api.classification.CreateClassification createClassification =
        new org.openmetadata.schema.api.classification.CreateClassification()
            .withName(classificationName)
            .withDescription("Classification for tag deletion test");
    client
        .getHttpClient()
        .execute(
            HttpMethod.PUT,
            "/v1/classifications",
            createClassification,
            org.openmetadata.schema.entity.classification.Classification.class);

    org.openmetadata.schema.api.classification.CreateTag createTag =
        new org.openmetadata.schema.api.classification.CreateTag()
            .withName(tagName)
            .withDescription("Tag for deletion test")
            .withClassification(classificationName);
    org.openmetadata.schema.entity.classification.Tag tag =
        client
            .getHttpClient()
            .execute(
                HttpMethod.PUT,
                "/v1/tags",
                createTag,
                org.openmetadata.schema.entity.classification.Tag.class);

    String tagFqn = classificationName + "." + tagName;

    TagLabel tagLabel =
        new TagLabel()
            .withTagFQN(tagFqn)
            .withSource(TagLabel.TagSource.CLASSIFICATION)
            .withLabelType(TagLabel.LabelType.MANUAL);
    entity.setTags(List.of(tagLabel));
    T entityWithTag = patchEntity(entity.getId().toString(), entity);

    assertNotNull(entityWithTag.getTags(), "Entity should have tags");
    assertTrue(
        entityWithTag.getTags().stream().anyMatch(t -> t.getTagFQN().equals(tagFqn)),
        "Entity should have the test tag");

    client
        .getHttpClient()
        .executeForString(HttpMethod.DELETE, "/v1/tags/" + tag.getId() + "?hardDelete=true", null);

    try {
      client.getHttpClient().executeForString(HttpMethod.GET, "/v1/tags/" + tag.getId(), null);
      fail("Tag should have been deleted");
    } catch (Exception e) {
      // Expected
    }

    T refreshedEntity = getEntityWithFields(entityWithTag.getId().toString(), "tags");
    assertNotNull(refreshedEntity, "Entity should still be retrievable after tag deletion");

    try {
      client
          .getHttpClient()
          .executeForString(
              HttpMethod.DELETE,
              "/v1/classifications/name/" + classificationName + "?hardDelete=true",
              null);
    } catch (Exception e) {
      // Ignore cleanup errors
    }
  }

  /**
   * Search for a specific entity by ID.
   * Subclasses should override to use entity-specific search.
   */
  protected String searchForEntity(String entityId) throws Exception {
    OpenMetadataClient client = SdkClients.adminClient();
    return client.search().query("id:" + entityId).index(getSearchIndexName()).size(1).execute();
  }

  protected String searchEntities() throws Exception {
    OpenMetadataClient client = SdkClients.adminClient();
    return client.search().query("*").index(getSearchIndexName()).execute();
  }

  // ===================================================================
  // AUTHORIZATION TESTS
  // Equivalent to: post_delete_entity_as_bot, put_entityUpdate_as_non_owner_4xx,
  //                patch_entityDescriptionAndTestAuthorizer
  // ===================================================================

  /**
   * Test: Bot can create and delete entities
   * Equivalent to: post_delete_entity_as_bot in EntityResourceTest
   */
  @Test
  void post_delete_entity_as_bot(TestNamespace ns) {
    Assumptions.assumeTrue(supportsBotOperations());

    // Create entity as bot
    T entity = createEntityAsBot(createMinimalRequest(ns));
    assertNotNull(entity, "Bot should be able to create entity");

    // Delete entity as bot
    deleteEntityAsBot(entity.getId().toString());

    // Verify entity is deleted
    assertThrows(
        Exception.class,
        () -> getEntity(entity.getId().toString()),
        "Entity should be deleted by bot");
  }

  /**
   * Check if this entity type supports bot operations.
   * Override in subclasses that don't support bot operations.
   */
  protected boolean supportsBotOperations() {
    return !supportsOwners; // Bot can operate on entities that don't require owners
  }

  /**
   * Create entity as bot. Subclasses should override.
   */
  protected T createEntityAsBot(K createRequest) {
    return createEntity(createRequest); // Default uses admin
  }

  /**
   * Delete entity as bot. Subclasses should override.
   */
  protected void deleteEntityAsBot(String id) {
    deleteEntity(id); // Default uses admin
  }

  /**
   * Test: Non-owner cannot update entity
   * Equivalent to: put_entityUpdate_as_non_owner_4xx in EntityResourceTest
   */
  @Test
  void put_entityUpdate_as_non_owner_4xx(TestNamespace ns) {
    Assumptions.assumeTrue(supportsOwners);
    Assumptions.assumeTrue(supportsPatch);

    // Create entity with USER1 as owner
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Set USER1 as owner
    entity.setOwners(List.of(testUser1Ref()));
    T withOwner = patchEntity(entity.getId().toString(), entity);

    // Try to update as USER2 (non-owner) - should fail
    withOwner.setDescription("Updated by non-owner");
    String entityId = withOwner.getId().toString();

    // This test verifies the authorization pattern - actual implementation
    // depends on entity-specific client behavior
    assertNotNull(withOwner.getOwners(), "Entity should have owner set");
  }

  /**
   * Test: Patch entity description with different user roles
   * Equivalent to: patch_entityDescriptionAndTestAuthorizer in EntityResourceTest
   */
  @Test
  void patch_entityDescriptionAndTestAuthorizer(TestNamespace ns) {
    Assumptions.assumeTrue(supportsPatch);

    // Create entity
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Admin can update
    entity.setDescription("Updated by admin");
    T updated = patchEntity(entity.getId().toString(), entity);
    assertEquals("Updated by admin", updated.getDescription(), "Admin should update description");

    // Set owner and verify owner can update
    if (supportsOwners) {
      updated.setOwners(List.of(testUser1Ref()));
      T withOwner = patchEntity(updated.getId().toString(), updated);
      assertNotNull(withOwner.getOwners(), "Entity should have owner");
    }
  }

  // ===================================================================
  // DELETE OPERATION TESTS
  // Equivalent to: post_delete_entity_as_admin_200, post_delete_entityWithOwner_200,
  //                post_delete_as_name_entity_as_admin_200
  // ===================================================================

  /**
   * Test: Admin can delete entity
   * Equivalent to: post_delete_entity_as_admin_200 in EntityResourceTest
   */
  @Test
  void post_delete_entity_as_admin_200(TestNamespace ns) {
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Delete as admin
    deleteEntity(entity.getId().toString());

    // Verify entity is deleted
    assertThrows(
        Exception.class,
        () -> getEntity(entity.getId().toString()),
        "Deleted entity should not be retrievable");
  }

  /**
   * Test: Delete entity that has an owner
   * Equivalent to: post_delete_entityWithOwner_200 in EntityResourceTest
   */
  @Test
  void post_delete_entityWithOwner_200(TestNamespace ns) {
    Assumptions.assumeTrue(supportsOwners);

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Set owner
    entity.setOwners(List.of(testUser1Ref()));
    T withOwner = patchEntity(entity.getId().toString(), entity);

    // Delete entity with owner
    deleteEntity(withOwner.getId().toString());

    // Verify entity is deleted
    assertThrows(
        Exception.class,
        () -> getEntity(withOwner.getId().toString()),
        "Entity with owner should be deletable");
  }

  /**
   * Test: Delete entity by name as admin
   * Equivalent to: post_delete_as_name_entity_as_admin_200 in EntityResourceTest
   */
  @Test
  void post_delete_as_name_entity_as_admin_200(TestNamespace ns) {
    if (!supportsDeleteByName()) return;

    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);

    // Delete by name
    deleteEntityByName(entity.getFullyQualifiedName());

    // Verify entity is deleted
    assertThrows(
        Exception.class,
        () -> getEntity(entity.getId().toString()),
        "Entity deleted by name should not be retrievable");
  }

  // ===================================================================
  // PERFORMANCE / EFFICIENCY TESTS
  // Equivalent to: test_fieldFetchersEfficiency in EntityResourceTest
  // ===================================================================

  /**
   * Test: Entity fetching is efficient.
   * Basic functionality test - comprehensive pagination is in PaginationIT.
   */
  @Test
  void test_fieldFetchersEfficiency(TestNamespace ns) {
    // Create a few entities
    List<T> entities = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      K createRequest = createRequest(ns.prefix("efficiency_" + i), ns);
      entities.add(createEntity(createRequest));
    }

    // Fetch each entity by ID - verify field fetching works
    for (T entity : entities) {
      T fetched = getEntity(entity.getId().toString());
      assertNotNull(fetched, "Entity should be fetchable");
      assertEquals(entity.getName(), fetched.getName(), "Names should match");
    }

    // Basic list test
    org.openmetadata.sdk.models.ListParams params = new org.openmetadata.sdk.models.ListParams();
    params.setLimit(10);
    org.openmetadata.sdk.models.ListResponse<T> response = listEntities(params);
    assertNotNull(response, "List response should not be null");
    assertTrue(response.getData().size() > 0, "Should have entities");
  }

  // ===================================================================
  // CSV IMPORT/EXPORT TESTS
  // Equivalent to: testImportExport, testImportInvalidCsv in EntityResourceTest
  // Only runs for entities that support import/export (supportsImportExport = true)
  // ===================================================================

  // Additional feature flags for import/export functionality
  protected boolean supportsBatchImport = false; // Override in subclasses that support batching
  protected boolean supportsRecursiveImport =
      false; // Override in subclasses that support recursive import

  /**
   * Get the entity service for import/export operations.
   * Subclasses that support import/export should override this method.
   *
   * @return The EntityServiceBase for this entity type, or null if not supported
   */
  protected org.openmetadata.sdk.services.EntityServiceBase<T> getEntityService() {
    return null; // Override in subclasses that support import/export
  }

  /**
   * Get the container entity name for import/export.
   * For example, for tables this would return the schema FQN.
   *
   * @param ns Test namespace
   * @return Container entity name/FQN for import/export operations
   */
  protected String getImportExportContainerName(TestNamespace ns) {
    return null; // Override in subclasses that support import/export
  }

  // ===================================================================
  // ABSTRACT CSV HELPER METHODS - Override in subclasses
  // Each entity type has different CSV structure and fields
  // ===================================================================

  /**
   * Generate valid CSV data for testing import functionality.
   * Each entity type has different CSV structure (e.g., table columns vs schema properties).
   *
   * @param ns Test namespace for unique naming
   * @param entities List of entities to generate CSV data for
   * @return Valid CSV data string with headers
   */
  protected String generateValidCsvData(TestNamespace ns, List<T> entities) {
    return null; // Override in subclasses that support import/export
  }

  /**
   * Generate invalid CSV data for testing error handling.
   *
   * @param ns Test namespace for unique naming
   * @return Invalid CSV data string (e.g., invalid field values, wrong data types)
   */
  protected String generateInvalidCsvData(TestNamespace ns) {
    return null; // Override in subclasses that support import/export
  }

  /**
   * Generate malformed CSV data for testing format validation.
   *
   * @return Malformed CSV data string (e.g., missing quotes, wrong delimiters)
   */
  protected String generateMalformedCsvData() {
    return "name,description\nunclosed\"quote,missing quote"; // Default malformed CSV
  }

  /**
   * Generate CSV data with missing required fields.
   *
   * @param ns Test namespace for unique naming
   * @return CSV data missing required columns
   */
  protected String generateCsvWithMissingRequiredFields(TestNamespace ns) {
    return null; // Override in subclasses that support import/export
  }

  /**
   * Generate CSV data with extra unexpected columns.
   *
   * @param ns Test namespace for unique naming
   * @return CSV data with additional columns
   */
  protected String generateCsvWithExtraColumns(TestNamespace ns) {
    return null; // Override in subclasses that support import/export
  }

  /**
   * Generate large CSV data for batch testing.
   *
   * @param ns Test namespace for unique naming
   * @param rowCount Number of rows to generate
   * @return Large CSV data for batch processing tests
   */
  protected String generateLargeCsvData(TestNamespace ns, int rowCount) {
    return null; // Override in subclasses that support import/export
  }

  /**
   * Get required CSV headers for this entity type.
   *
   * @return List of required header names
   */
  protected List<String> getRequiredCsvHeaders() {
    return new ArrayList<>(); // Override in subclasses
  }

  /**
   * Get all CSV headers (both required and optional) for this entity type.
   *
   * @return List of all header names
   */
  protected List<String> getAllCsvHeaders() {
    return new ArrayList<>(); // Override in subclasses
  }

  /**
   * Get optional CSV headers for this entity type.
   *
   * @return List of optional header names
   */
  protected List<String> getOptionalCsvHeaders() {
    return new ArrayList<>(); // Override in subclasses
  }

  /**
   * Validate CSV import result against expected entities.
   *
   * @param result The import result to validate
   * @param expectedEntities List of entities that were expected to be imported
   */
  protected void validateCsvImportResult(CsvImportResult result, List<T> expectedEntities) {
    assertNotNull(result, "Import result should not be null");
    assertNotNull(result.getStatus(), "Import status should not be null");
    // Basic validation - subclasses can override for entity-specific validation
  }

  /**
   * Validate that CSV import actually persisted data correctly to the database.
   * This method should be called after CSV import to verify data persistence.
   *
   * @param originalEntities Entities that existed before import (for updates)
   * @param csvData The CSV data that was imported
   * @param result The import result
   */
  protected void validateCsvDataPersistence(
      List<T> originalEntities, String csvData, CsvImportResult result) {
    // Simple approach: Just fetch entities by name and validate expected changes
    assertNotNull(result, "Import result should not be null");
    assertEquals(ApiStatus.SUCCESS, result.getStatus(), "Import should succeed for validation");

    // Subclasses should override to:
    // 1. getByName() for each entity they expect to be changed
    // 2. Assert that expected field values are present
  }

  /**
   * Helper method to assert that two lists of entity references are equal.
   */
  protected void assertEntityReferencesEqual(
      List<org.openmetadata.schema.type.EntityReference> expected,
      List<org.openmetadata.schema.type.EntityReference> actual,
      String fieldName) {
    if (expected == null && actual == null) {
      return;
    }
    if (expected == null || actual == null) {
      fail(fieldName + " references do not match: one is null, other is not");
    }
    assertEquals(expected.size(), actual.size(), fieldName + " reference count should match");

    Set<String> expectedNames =
        expected.stream()
            .map(org.openmetadata.schema.type.EntityReference::getName)
            .collect(java.util.stream.Collectors.toSet());
    Set<String> actualNames =
        actual.stream()
            .map(org.openmetadata.schema.type.EntityReference::getName)
            .collect(java.util.stream.Collectors.toSet());

    assertEquals(expectedNames, actualNames, fieldName + " reference names should match");
  }

  /**
   * Helper method to assert that tag lists are equal.
   */
  protected void assertTagsEqual(
      List<org.openmetadata.schema.type.TagLabel> expected,
      List<org.openmetadata.schema.type.TagLabel> actual,
      String fieldName) {
    if (expected == null && actual == null) {
      return;
    }
    if (expected == null || actual == null) {
      fail(fieldName + " tags do not match: one is null, other is not");
    }

    Set<String> expectedTags =
        expected.stream()
            .map(org.openmetadata.schema.type.TagLabel::getTagFQN)
            .collect(java.util.stream.Collectors.toSet());
    Set<String> actualTags =
        actual.stream()
            .map(org.openmetadata.schema.type.TagLabel::getTagFQN)
            .collect(java.util.stream.Collectors.toSet());

    assertEquals(expectedTags, actualTags, fieldName + " tag FQNs should match");
  }

  /**
   * Create multiple test entities for CSV generation.
   *
   * @param ns Test namespace
   * @param count Number of entities to create
   * @return List of created entities
   */
  protected List<T> createTestEntities(TestNamespace ns, int count) {
    List<T> entities = new ArrayList<>();
    for (int i = 0; i < count; i++) {
      K createRequest = createRequest(ns.prefix("csvTest" + i), ns);
      T entity = createEntity(createRequest);
      entities.add(entity);
    }
    return entities;
  }

  /**
   * Perform CSV import with specified parameters.
   *
   * @param ns Test namespace
   * @param csvData CSV data to import
   * @param dryRun Whether to perform dry run
   * @return CsvImportResult
   */
  protected CsvImportResult performImportCsv(TestNamespace ns, String csvData, boolean dryRun) {
    return performImportCsv(ns, csvData, dryRun, false);
  }

  /**
   * Perform CSV import with recursive option.
   *
   * @param ns Test namespace
   * @param csvData CSV data to import
   * @param dryRun Whether to perform dry run
   * @param recursive Whether to perform recursive import
   * @return CsvImportResult
   */
  protected CsvImportResult performImportCsv(
      TestNamespace ns, String csvData, boolean dryRun, boolean recursive) {
    try {
      org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
      String containerName = getImportExportContainerName(ns);
      String result = service.importCsv(containerName, csvData, dryRun);
      return JsonUtils.readValue(result, CsvImportResult.class);
    } catch (Exception e) {
      throw new RuntimeException("CSV import failed: " + e.getMessage(), e);
    }
  }

  /**
   * Validate CSV structure and headers.
   *
   * @param csvData CSV data to validate
   */
  protected void validateCsvStructure(String csvData) {
    assertNotNull(csvData, "CSV data should not be null");
    assertFalse(csvData.trim().isEmpty(), "CSV data should not be empty");

    String[] lines = csvData.split("\n");
    assertTrue(lines.length >= 1, "CSV should have at least header row");

    String[] headers = lines[0].split(",");
    List<String> requiredHeaders = getRequiredCsvHeaders();

    for (String required : requiredHeaders) {
      boolean found =
          Arrays.stream(headers).anyMatch(header -> header.trim().equals(required.trim()));
      assertTrue(found, "CSV missing required header: " + required);
    }
  }

  /**
   * Test: Basic CSV export works.
   * Equivalent to: testImportExport (export part) in EntityResourceTest
   */
  @Test
  void test_exportCsv(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String containerName = getImportExportContainerName(ns);
    Assumptions.assumeTrue(containerName != null, "Container name not provided");

    // Create an entity first
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    assertNotNull(entity, "Entity should be created");

    // Export CSV
    try {
      String csv = service.exportCsv(containerName);
      assertNotNull(csv, "Export should return CSV data");
      assertFalse(csv.isEmpty(), "CSV should not be empty");
      // CSV should have at least a header row
      assertTrue(csv.contains(","), "CSV should have comma-separated values");
    } catch (org.openmetadata.sdk.exceptions.OpenMetadataException e) {
      // Some entities may not support export - that's OK if supportsImportExport is false
      fail("Export failed: " + e.getMessage());
    }
  }

  /**
   * Test: CSV import with dry run validates data.
   * Equivalent to: testImportInvalidCsv (validation part) in EntityResourceTest
   */
  @Test
  void test_importCsvDryRun(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String containerName = getImportExportContainerName(ns);
    Assumptions.assumeTrue(containerName != null, "Container name not provided");

    // Create an entity first
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    assertNotNull(entity, "Entity should be created");

    // Export to get valid CSV format
    try {
      String exportedCsv = service.exportCsv(containerName);
      assertNotNull(exportedCsv, "Export should return CSV data");

      // Import with dry run (should not actually modify data)
      String result = service.importCsv(containerName, exportedCsv, true);
      assertNotNull(result, "Import dry run should return a result");

      // Verify CsvImportResult values
      CsvImportResult importResult = JsonUtils.readValue(result, CsvImportResult.class);
      assertNotNull(importResult, "Should parse CsvImportResult from response");
      assertTrue(importResult.getDryRun(), "Should be a dry run");
      assertNotNull(importResult.getStatus(), "Status should not be null");
      assertTrue(
          importResult.getNumberOfRowsProcessed() >= 0, "Rows processed should be non-negative");
      assertTrue(importResult.getNumberOfRowsPassed() >= 0, "Rows passed should be non-negative");
      assertTrue(importResult.getNumberOfRowsFailed() >= 0, "Rows failed should be non-negative");
      assertEquals(
          importResult.getNumberOfRowsProcessed(),
          importResult.getNumberOfRowsPassed() + importResult.getNumberOfRowsFailed(),
          "Rows processed should equal passed + failed");
    } catch (org.openmetadata.sdk.exceptions.OpenMetadataException e) {
      fail("Import/export failed: " + e.getMessage());
    }
  }

  /**
   * Test: CSV import round-trip (export then import).
   * Equivalent to: testImportExport in EntityResourceTest
   */
  @Test
  void test_importExportRoundTrip(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String containerName = getImportExportContainerName(ns);
    Assumptions.assumeTrue(containerName != null, "Container name not provided");

    // Create an entity first
    K createRequest = createMinimalRequest(ns);
    T entity = createEntity(createRequest);
    assertNotNull(entity, "Entity should be created");

    try {
      // Export current state
      String exportedCsv = service.exportCsv(containerName);
      assertNotNull(exportedCsv, "Export should return CSV data");

      // Import the exported data (should succeed without changes)
      String result = service.importCsv(containerName, exportedCsv, false);
      assertNotNull(result, "Import should return a result");

      // Verify CsvImportResult values
      CsvImportResult importResult = JsonUtils.readValue(result, CsvImportResult.class);
      assertNotNull(importResult, "Should parse CsvImportResult from response");
      assertFalse(importResult.getDryRun(), "Should not be a dry run");
      assertEquals(
          ApiStatus.SUCCESS,
          importResult.getStatus(),
          "Import should succeed: " + importResult.getImportResultsCsv());
      assertTrue(
          importResult.getNumberOfRowsProcessed() >= 0, "Rows processed should be non-negative");
      assertEquals(0, importResult.getNumberOfRowsFailed(), "No rows should fail on round-trip");

      // Export again and verify consistency
      String reExportedCsv = service.exportCsv(containerName);
      assertNotNull(reExportedCsv, "Re-export should return CSV data");

      // The re-exported CSV should be similar to original
      // (may have minor differences like timestamps, but structure should match)
      String[] originalLines = exportedCsv.split("\n");
      String[] reExportedLines = reExportedCsv.split("\n");

      // At minimum, header should match
      assertEquals(
          originalLines[0], reExportedLines[0], "CSV headers should match after round-trip");
    } catch (org.openmetadata.sdk.exceptions.OpenMetadataException e) {
      fail("Import/export round-trip failed: " + e.getMessage());
    }
  }

  // ===================================================================
  // COMPREHENSIVE CSV IMPORT/EXPORT TESTS
  // Template-based tests that work with any entity CSV structure
  // ===================================================================

  /**
   * Test: Import CSV with dry run and valid data.
   * Verifies dry run validation works without making changes.
   */
  @Test
  void test_importCsv_dryRun_validData(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");
    Assumptions.assumeTrue(
        generateValidCsvData(ns, new ArrayList<>()) != null,
        "Entity does not provide CSV data generation");

    // Create test entities
    List<T> entities = createTestEntities(ns, 3);

    // Generate valid CSV data
    String csvData = generateValidCsvData(ns, entities);
    validateCsvStructure(csvData);

    // Perform dry run import
    CsvImportResult result = performImportCsv(ns, csvData, true);

    // Validate dry run result
    assertTrue(result.getDryRun(), "Should be a dry run");
    assertNotNull(result.getStatus(), "Status should not be null");
    assertTrue(result.getNumberOfRowsProcessed() >= 0, "Rows processed should be non-negative");
    validateCsvImportResult(result, entities);
  }

  /**
   * Test: Import CSV with actual data persistence.
   * Verifies actual import creates/updates entities.
   */
  @Test
  void test_importCsv_actual_validData(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");
    Assumptions.assumeTrue(
        generateValidCsvData(ns, new ArrayList<>()) != null,
        "Entity does not provide CSV data generation");

    // Create test entities
    List<T> entities = createTestEntities(ns, 2);

    // Generate valid CSV data
    String csvData = generateValidCsvData(ns, entities);
    validateCsvStructure(csvData);

    // Perform actual import
    CsvImportResult result = performImportCsv(ns, csvData, false);

    // Validate actual import result
    assertFalse(result.getDryRun(), "Should not be a dry run");
    assertEquals(
        ApiStatus.SUCCESS,
        result.getStatus(),
        "Import should succeed: " + result.getImportResultsCsv());
    assertTrue(result.getNumberOfRowsPassed() > 0, "Should have passed rows");
    assertEquals(0, result.getNumberOfRowsFailed(), "Should have no failed rows");
    validateCsvImportResult(result, entities);

    // CRITICAL: Validate data persistence - verify CSV changes were actually saved to database
    validateCsvDataPersistence(entities, csvData, result);
  }

  /**
   * Test: Export CSV basic functionality.
   * Verifies CSV export produces valid format.
   */
  @Test
  void test_exportCsv_basicFunctionality(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String containerName = getImportExportContainerName(ns);
    Assumptions.assumeTrue(containerName != null, "Container name not provided");

    // Create test entities
    createTestEntities(ns, 2);

    // Export CSV
    try {
      String csv = service.exportCsv(containerName);

      // Validate export result
      assertNotNull(csv, "Export should return CSV data");
      assertFalse(csv.trim().isEmpty(), "CSV should not be empty");
      assertTrue(csv.contains(","), "CSV should have comma-separated values");

      // Validate structure
      String[] lines = csv.split("\n");
      assertTrue(lines.length >= 1, "CSV should have at least header row");

      // Check if has content beyond headers
      if (lines.length > 1) {
        assertTrue(lines[1].trim().length() > 0, "CSV should have data rows");
      }
    } catch (Exception e) {
      fail("Export failed: " + e.getMessage());
    }
  }

  /**
   * Test: Import/Export round-trip consistency.
   * Verifies data integrity through export→import cycle.
   */
  @Test
  void test_importExport_roundTripConsistency(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String containerName = getImportExportContainerName(ns);
    Assumptions.assumeTrue(containerName != null, "Container name not provided");

    // Create test entities
    List<T> entities = createTestEntities(ns, 2);

    try {
      // Export current state
      String exportedCsv = service.exportCsv(containerName);
      assertNotNull(exportedCsv, "Export should return CSV data");

      // Import back the exported data
      CsvImportResult importResult = performImportCsv(ns, exportedCsv, false);
      assertEquals(ApiStatus.SUCCESS, importResult.getStatus(), "Round-trip import should succeed");

      // Export again to verify consistency
      String reExportedCsv = service.exportCsv(containerName);
      assertNotNull(reExportedCsv, "Re-export should return CSV data");

      // Compare headers (structure should be consistent)
      String[] originalLines = exportedCsv.split("\n");
      String[] reExportedLines = reExportedCsv.split("\n");
      assertEquals(
          originalLines[0], reExportedLines[0], "CSV headers should match after round-trip");

    } catch (Exception e) {
      fail("Round-trip test failed: " + e.getMessage());
    }
  }

  /**
   * Test: Import CSV with invalid format.
   * Verifies malformed CSV handling.
   */
  @Test
  void test_importCsv_invalidFormat(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    // Test with malformed CSV data
    String malformedCsv = generateMalformedCsvData();

    try {
      CsvImportResult result = performImportCsv(ns, malformedCsv, true);

      // Should either fail completely or have failed rows
      assertTrue(
          result.getStatus() == ApiStatus.FAILURE || result.getNumberOfRowsFailed() > 0,
          "Malformed CSV should cause failures");

    } catch (Exception e) {
      // Exception is acceptable for malformed CSV
      assertTrue(
          e.getMessage().contains("CSV") || e.getMessage().contains("format"),
          "Exception should be related to CSV format");
    }
  }

  /**
   * Test: Import CSV with invalid data values.
   * Verifies data validation during import.
   */
  @Test
  void test_importCsv_invalidData(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String invalidCsv = generateInvalidCsvData(ns);
    Assumptions.assumeTrue(invalidCsv != null, "Entity does not provide invalid CSV data");

    try {
      CsvImportResult result = performImportCsv(ns, invalidCsv, true);

      // Should have validation failures
      assertTrue(
          result.getStatus() == ApiStatus.PARTIAL_SUCCESS
              || result.getStatus() == ApiStatus.FAILURE,
          "Invalid data should cause validation failures");
      assertTrue(result.getNumberOfRowsFailed() > 0, "Should have failed rows with invalid data");

    } catch (Exception e) {
      // Exception is acceptable for invalid data
      assertFalse(e.getMessage().isEmpty(), "Exception should have meaningful message");
    }
  }

  /**
   * Test: Import CSV with missing required fields.
   * Verifies required field validation.
   */
  @Test
  void test_importCsv_missingRequiredFields(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String csvWithMissingFields = generateCsvWithMissingRequiredFields(ns);
    Assumptions.assumeTrue(
        csvWithMissingFields != null, "Entity does not provide CSV with missing fields");

    try {
      CsvImportResult result = performImportCsv(ns, csvWithMissingFields, true);

      // Should fail validation for missing required fields
      assertTrue(
          result.getStatus() == ApiStatus.FAILURE || result.getNumberOfRowsFailed() > 0,
          "Missing required fields should cause failures");

    } catch (Exception e) {
      // Exception is acceptable for missing required fields
      assertTrue(
          e.getMessage().contains("required") || e.getMessage().contains("missing"),
          "Exception should indicate missing required fields");
    }
  }

  /**
   * Test: Import CSV with extra columns.
   * Verifies graceful handling of unexpected columns.
   */
  @Test
  void test_importCsv_extraColumns(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    String csvWithExtraColumns = generateCsvWithExtraColumns(ns);
    Assumptions.assumeTrue(
        csvWithExtraColumns != null, "Entity does not provide CSV with extra columns");

    try {
      CsvImportResult result = performImportCsv(ns, csvWithExtraColumns, true);

      // Should handle extra columns gracefully
      assertTrue(
          result.getStatus() == ApiStatus.SUCCESS
              || result.getStatus() == ApiStatus.PARTIAL_SUCCESS,
          "Extra columns should be handled gracefully");
      assertTrue(
          result.getNumberOfRowsProcessed() > 0, "Should process rows despite extra columns");

    } catch (Exception e) {
      // Minor exception might be acceptable, but shouldn't be fatal
      assertFalse(
          e.getMessage().contains("fatal") || e.getMessage().contains("critical"),
          "Extra columns should not cause critical errors");
    }
  }

  /**
   * Test: Import CSV with batch processing.
   * Verifies batch processing handles multiple entities correctly.
   */
  @Test
  void test_importCsv_batchProcessing(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");
    Assumptions.assumeTrue(supportsBatchImport, "Entity does not support batch import");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    // Generate large dataset (more than typical batch size)
    String largeCsv = generateLargeCsvData(ns, 50);
    Assumptions.assumeTrue(largeCsv != null, "Entity does not provide large CSV data");

    try {
      CsvImportResult result = performImportCsv(ns, largeCsv, false);

      // Should handle batch processing successfully
      assertEquals(ApiStatus.SUCCESS, result.getStatus(), "Batch processing should succeed");
      assertTrue(result.getNumberOfRowsProcessed() >= 50, "Should process all batch rows");
      assertTrue(result.getNumberOfRowsPassed() > 0, "Should have successfully processed rows");

    } catch (Exception e) {
      fail("Batch processing failed: " + e.getMessage());
    }
  }

  /**
   * Test: Import empty CSV data.
   * Verifies handling of empty CSV (headers only).
   */
  @Test
  void test_importCsv_emptyData(TestNamespace ns) {
    Assumptions.assumeTrue(supportsImportExport, "Entity does not support import/export");

    org.openmetadata.sdk.services.EntityServiceBase<T> service = getEntityService();
    Assumptions.assumeTrue(service != null, "Entity service not provided");

    // Create CSV with all headers only (no data rows)
    List<String> headers = getAllCsvHeaders();
    if (headers.isEmpty()) {
      headers = getRequiredCsvHeaders();
    }
    if (headers.isEmpty()) {
      headers = List.of("name"); // Default header
    }
    String emptyCsv = String.join(",", headers);

    try {
      CsvImportResult result = performImportCsv(ns, emptyCsv, true);

      // Should handle empty data gracefully
      assertNotNull(result.getStatus(), "Status should not be null");
      assertEquals(0, result.getNumberOfRowsFailed(), "Empty CSV should not have failed rows");
      assertTrue(result.getNumberOfRowsProcessed() >= 0, "Processed count should be non-negative");

    } catch (Exception e) {
      // Should not fail for empty CSV
      fail("Empty CSV should not cause failures: " + e.getMessage());
    }
  }

  private static final ObjectMapper MAPPER = new ObjectMapper();

  @Nested
  class ListEntityHistoryByTimestampPaginationTest {
    @Test
    void test_listEntityHistoryByTimestamp_pagination(TestNamespace ns) throws Exception {
      Assumptions.assumeTrue(
          supportsListHistoryByTimestamp,
          "Entity does not support listEntityHistoryByTimestamp endpoint");
      Assumptions.assumeTrue(supportsPatch, "Entity does not support patch operations");

      OpenMetadataClient client = SdkClients.adminClient();
      long startTs = System.currentTimeMillis();

      List<T> createdEntities = new ArrayList<>();
      for (int i = 0; i < 3; i++) {
        K createRequest = createRequest(ns.prefix("versions_test_" + i), ns);
        T entity = createEntity(createRequest);
        createdEntities.add(entity);

        entity.setDescription("Updated description v2 - " + System.currentTimeMillis());
        patchEntity(entity.getId().toString(), entity);

        entity.setDescription("Updated description v3 - " + System.currentTimeMillis());
        patchEntity(entity.getId().toString(), entity);
      }

      long endTs = System.currentTimeMillis();
      String basePath = getResourcePath() + "history";

      String response =
          client
              .getHttpClient()
              .executeForString(
                  HttpMethod.GET,
                  basePath + "?startTs=" + startTs + "&endTs=" + endTs + "&limit=2",
                  null);

      assertNotNull(response, "Response should not be null");
      JsonNode result = MAPPER.readTree(response);

      assertTrue(result.has("data"), "Response should have 'data' field");
      JsonNode data = result.get("data");
      assertTrue(data.isArray(), "Data should be an array");
      assertTrue(data.size() <= 2, "Data size should respect limit of 2");

      if (result.has("paging") && result.get("paging").has("after")) {
        String afterCursor = result.get("paging").get("after").asText();
        assertNotNull(afterCursor, "After cursor should be present for paginated results");

        String page2Response =
            client
                .getHttpClient()
                .executeForString(
                    HttpMethod.GET,
                    basePath
                        + "?startTs="
                        + startTs
                        + "&endTs="
                        + endTs
                        + "&limit=2&after="
                        + afterCursor,
                    null);

        JsonNode page2Result = MAPPER.readTree(page2Response);
        assertTrue(page2Result.has("data"), "Page 2 response should have 'data' field");
        JsonNode page2Data = page2Result.get("data");
        assertTrue(page2Data.isArray(), "Page 2 data should be an array");

        if (page2Result.has("paging") && page2Result.get("paging").has("before")) {
          String beforeCursor = page2Result.get("paging").get("before").asText();

          String backResponse =
              client
                  .getHttpClient()
                  .executeForString(
                      HttpMethod.GET,
                      basePath
                          + "?startTs="
                          + startTs
                          + "&endTs="
                          + endTs
                          + "&limit=2&before="
                          + beforeCursor,
                      null);

          JsonNode backResult = MAPPER.readTree(backResponse);
          assertTrue(backResult.has("data"), "Back navigation response should have 'data' field");
        }
      }
    }

    @Test
    void test_listEntityHistoryByTimestamp_withInvalidLimit(TestNamespace ns) throws Exception {
      Assumptions.assumeTrue(
          supportsListHistoryByTimestamp,
          "Entity does not support listEntityHistoryByTimestamp endpoint");

      OpenMetadataClient client = SdkClients.adminClient();
      long now = System.currentTimeMillis();
      String basePath = getResourcePath() + "history";

      assertThrows(
          Exception.class,
          () ->
              client
                  .getHttpClient()
                  .executeForString(
                      HttpMethod.GET,
                      basePath + "?startTs=" + (now - 1000) + "&endTs=" + now + "&limit=0",
                      null),
          "Limit of 0 should fail validation");

      assertThrows(
          Exception.class,
          () ->
              client
                  .getHttpClient()
                  .executeForString(
                      HttpMethod.GET,
                      basePath + "?startTs=" + (now - 1000) + "&endTs=" + now + "&limit=-1",
                      null),
          "Negative limit should fail validation");
    }

    @Test
    void test_listEntityHistoryByTimestamp_emptyTimeRange(TestNamespace ns) throws Exception {
      Assumptions.assumeTrue(
          supportsListHistoryByTimestamp,
          "Entity does not support listEntityHistoryByTimestamp endpoint");

      OpenMetadataClient client = SdkClients.adminClient();
      long futureStart = System.currentTimeMillis() + 86400000;
      long futureEnd = futureStart + 1000;
      String basePath = getResourcePath() + "history";

      String response =
          client
              .getHttpClient()
              .executeForString(
                  HttpMethod.GET,
                  basePath + "?startTs=" + futureStart + "&endTs=" + futureEnd + "&limit=10",
                  null);

      JsonNode result = MAPPER.readTree(response);
      assertTrue(result.has("data"), "Response should have 'data' field");
      JsonNode data = result.get("data");
      assertTrue(data.isArray(), "Data should be an array");
      assertEquals(0, data.size(), "Data should be empty for future time range");
    }

    @Test
    @Timeout(180)
    void test_listEntityHistoryByTimestamp_completePaginationCycle(TestNamespace ns)
        throws Exception {
      Assumptions.assumeTrue(
          supportsListHistoryByTimestamp,
          "Entity does not support listEntityHistoryByTimestamp endpoint");
      Assumptions.assumeTrue(supportsPatch, "Entity does not support patch operations");

      OpenMetadataClient client = SdkClients.adminClient();
      long startTs = System.currentTimeMillis();

      for (int i = 0; i < 5; i++) {
        K createRequest = createRequest(ns.prefix("pagination_cycle_" + i), ns);
        T entity = createEntity(createRequest);

        entity.setDescription("Updated v2 - " + System.currentTimeMillis());
        patchEntity(entity.getId().toString(), entity);
      }

      long endTs = System.currentTimeMillis();
      String basePath = getResourcePath() + "history";
      int limit = 3;

      List<String> allIds = new ArrayList<>();
      List<String> afterCursors = new ArrayList<>();
      String afterCursor = null;
      String lastPageBeforeCursor = null;
      int forwardPageCount = 0;

      do {
        String url = basePath + "?startTs=" + startTs + "&endTs=" + endTs + "&limit=" + limit;
        if (afterCursor != null) {
          url += "&after=" + afterCursor;
        }

        String response = client.getHttpClient().executeForString(HttpMethod.GET, url, null);
        JsonNode result = MAPPER.readTree(response);
        JsonNode data = result.get("data");

        for (JsonNode item : data) {
          if (item.has("id")) {
            allIds.add(item.get("id").asText());
          }
        }

        forwardPageCount++;
        afterCursor = null;
        if (result.has("paging") && !result.get("paging").isNull()) {
          JsonNode paging = result.get("paging");
          if (paging.has("after") && !paging.get("after").isNull()) {
            afterCursor = paging.get("after").asText();
            afterCursors.add(afterCursor);
          }
          if (paging.has("before") && !paging.get("before").isNull()) {
            lastPageBeforeCursor = paging.get("before").asText();
          }
        }
      } while (afterCursor != null);

      assertTrue(forwardPageCount > 1, "Should have paginated through multiple pages forward");
      assertFalse(allIds.isEmpty(), "Should have collected entity version IDs");

      if (!afterCursors.isEmpty() && lastPageBeforeCursor != null) {
        String beforeCursor = lastPageBeforeCursor;

        int backwardPageCount = 0;
        while (beforeCursor != null) {
          String backUrl =
              basePath
                  + "?startTs="
                  + startTs
                  + "&endTs="
                  + endTs
                  + "&limit="
                  + limit
                  + "&before="
                  + beforeCursor;
          String backResponse =
              client.getHttpClient().executeForString(HttpMethod.GET, backUrl, null);
          JsonNode backResult = MAPPER.readTree(backResponse);

          backwardPageCount++;
          beforeCursor = null;
          if (backResult.has("paging") && !backResult.get("paging").isNull()) {
            JsonNode paging = backResult.get("paging");
            if (paging.has("before") && !paging.get("before").isNull()) {
              beforeCursor = paging.get("before").asText();
            }
          }
        }

        assertTrue(
            backwardPageCount >= 1, "Should have been able to navigate backward at least once");

        String firstPageUrl =
            basePath + "?startTs=" + startTs + "&endTs=" + endTs + "&limit=" + limit;
        String firstPageResponse =
            client.getHttpClient().executeForString(HttpMethod.GET, firstPageUrl, null);
        JsonNode firstPageResult = MAPPER.readTree(firstPageResponse);

        boolean hasBeforeOnFirstPage = false;
        if (firstPageResult.has("paging") && !firstPageResult.get("paging").isNull()) {
          JsonNode paging = firstPageResult.get("paging");
          hasBeforeOnFirstPage = paging.has("before") && !paging.get("before").isNull();
        }
        assertFalse(hasBeforeOnFirstPage, "First page should not have 'before' cursor");
      }
    }
  }

  // ===================================================================
  // CHANGE SUMMARY TESTS
  // ===================================================================

  /**
   * Test: Retrieve changeSummary by entity ID after updating the entity.
   * The changeSummary API returns metadata about who changed each field,
   * the source of the change, and when it was changed.
   */
  @Test
  void get_changeSummaryById_200(TestNamespace ns) throws Exception {
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("Updated description for changeSummary test");
    T updated = patchEntity(created.getId().toString(), created);

    OpenMetadataClient client = SdkClients.adminClient();
    String response =
        client
            .getHttpClient()
            .executeForString(
                HttpMethod.GET,
                "/v1/changeSummary/" + getEntityType() + "/" + updated.getId(),
                null);
    assertNotNull(response, "ChangeSummary response should not be null");
    JsonNode result = MAPPER.readTree(response);
    assertTrue(result.has("changeSummary"), "Response must contain changeSummary field");
    assertTrue(result.has("totalEntries"), "Response must contain totalEntries field");
    assertTrue(
        result.get("totalEntries").asInt() > 0,
        "totalEntries should be > 0 after patching description");
    JsonNode changeSummaryNode = result.get("changeSummary");
    assertTrue(
        changeSummaryNode.isObject() && changeSummaryNode.size() > 0,
        "changeSummary should contain at least one entry after patching description");
  }

  /**
   * Test: Retrieve changeSummary by entity FQN after updating the entity.
   */
  @Test
  void get_changeSummaryByFqn_200(TestNamespace ns) throws Exception {
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("Updated description for changeSummary FQN test");
    T updated = patchEntity(created.getId().toString(), created);

    OpenMetadataClient client = SdkClients.adminClient();
    String fqn = updated.getFullyQualifiedName();
    String response =
        client
            .getHttpClient()
            .executeForString(
                HttpMethod.GET, "/v1/changeSummary/" + getEntityType() + "/name/" + fqn, null);
    assertNotNull(response, "ChangeSummary response should not be null");
    JsonNode result = MAPPER.readTree(response);
    assertTrue(result.has("changeSummary"), "Response must contain changeSummary field");
    assertTrue(result.has("totalEntries"), "Response must contain totalEntries field");
    assertTrue(
        result.get("totalEntries").asInt() > 0,
        "totalEntries should be > 0 after patching description");
    JsonNode changeSummaryNode = result.get("changeSummary");
    assertTrue(
        changeSummaryNode.isObject() && changeSummaryNode.size() > 0,
        "changeSummary should contain at least one entry after patching description");
  }

  /**
   * Test: Retrieve changeSummary with fieldPrefix filter.
   * Verifies that the filtering parameter works correctly.
   */
  @Test
  void get_changeSummaryWithFieldPrefix_200(TestNamespace ns) throws Exception {
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("Updated for fieldPrefix test");
    T updated = patchEntity(created.getId().toString(), created);

    OpenMetadataClient client = SdkClients.adminClient();
    String response =
        client
            .getHttpClient()
            .executeForString(
                HttpMethod.GET,
                "/v1/changeSummary/"
                    + getEntityType()
                    + "/"
                    + updated.getId()
                    + "?fieldPrefix=description",
                null);
    assertNotNull(response, "ChangeSummary filtered response should not be null");
    JsonNode result = MAPPER.readTree(response);
    assertTrue(result.has("changeSummary"), "Response must contain changeSummary field");
    assertTrue(result.has("totalEntries"), "Response must contain totalEntries field");

    JsonNode changeSummary = result.get("changeSummary");
    assertTrue(
        changeSummary.isObject() && changeSummary.size() > 0,
        "Filtered changeSummary should contain at least one entry matching 'description' prefix");
    changeSummary
        .fieldNames()
        .forEachRemaining(
            key ->
                assertTrue(
                    key.startsWith("description"),
                    "All keys should start with 'description', but found: " + key));
  }

  /**
   * Test: Retrieve changeSummary with pagination parameters.
   */
  @Test
  void get_changeSummaryWithPagination_200(TestNamespace ns) throws Exception {
    K createRequest = createMinimalRequest(ns);
    T created = createEntity(createRequest);

    created.setDescription("Updated for pagination test");
    patchEntity(created.getId().toString(), created);

    OpenMetadataClient client = SdkClients.adminClient();
    String response =
        client
            .getHttpClient()
            .executeForString(
                HttpMethod.GET,
                "/v1/changeSummary/"
                    + getEntityType()
                    + "/"
                    + created.getId()
                    + "?limit=1&offset=0",
                null);
    assertNotNull(response, "ChangeSummary paginated response should not be null");
    JsonNode result = MAPPER.readTree(response);
    assertTrue(result.has("changeSummary"), "Response must contain changeSummary field");
    assertTrue(result.has("totalEntries"), "Response must contain totalEntries field");
    assertTrue(result.has("offset"), "Paginated response must contain offset field");
    assertTrue(result.has("limit"), "Paginated response must contain limit field");
  }

  /**
   * Test: changeSummary returns 404 for non-existent entity.
   */
  @Test
  void get_changeSummaryNotFound_404(TestNamespace ns) {
    OpenMetadataClient client = SdkClients.adminClient();
    UUID randomId = UUID.randomUUID();
    Exception thrown =
        assertThrows(
            Exception.class,
            () ->
                client
                    .getHttpClient()
                    .executeForString(
                        HttpMethod.GET,
                        "/v1/changeSummary/" + getEntityType() + "/" + randomId,
                        null));
    assertTrue(
        thrown.getMessage().contains("404") || thrown.getMessage().contains("not found"),
        "Should get 404 for non-existent entity, got: " + thrown.getMessage());
  }
}