Add unit tests for TargetInsertStatement bind input validation and bind-index ordering

Cover TargetInsertStatement.bind(): rejection when TTL or writetime
columns are enabled but no value is supplied, sequential bind indices
for target columns, trailing TTL and writetime indices and types,
constant-column skipping, explode-map binding, and exception wrapping
when converting an origin value fails.

Author: vasiliy-mikhailov

src/test/java/com/datastax/cdm/cql/statement/TargetInsertStatementTest.java

@@ -22,8 +22,10 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 
 import com.datastax.cdm.cql.CommonMocks;
+import com.datastax.cdm.properties.KnownProperties;
 import com.datastax.oss.driver.api.core.cql.*;
 import com.datastax.oss.driver.api.core.type.DataTypes;
 
@@ -343,4 +345,237 @@ public void bind_withNonEmptyMap_shouldCallSet() {
         verify(boundStatement, atLeastOnce()).set(anyInt(), any(), any(Class.class));
     }
 
+    // ---- bind(): input validation and bind-index ordering ----
+
+    /**
+     * When TTL columns are enabled but no TTL value is supplied, bind() rejects the input with a RuntimeException
+     * naming the missing TTL property, before any value is bound onto the statement.
+     */
+    @Test
+    public void bind_checkBindInputsThrowsWhenTtlMissing() {
+        when(writetimeTTLFeature.isEnabled()).thenReturn(true);
+        when(writetimeTTLFeature.hasTTLColumns()).thenReturn(true);
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        // If any value is set onto the statement, fail with a distinct exception so the assertion below
+        // cannot pass by accident.
+        doThrow(new IllegalStateException("TTL set called unexpectedly")).when(boundStatement).set(anyInt(), isNull(),
+                any(Class.class));
+
+        RuntimeException ex = assertThrows(RuntimeException.class,
+                () -> targetInsertStatement.bind(originRow, targetRow, null, null, null, null));
+        assertEquals(KnownProperties.ORIGIN_TTL_NAMES + " specified, but no TTL value was provided", ex.getMessage());
+    }
+
+    /**
+     * When writetime columns are enabled but no writetime value is supplied, bind() rejects the input with a
+     * RuntimeException naming the missing writetime property.
+     */
+    @Test
+    public void bind_checkBindInputsThrowsWhenWriteTimeMissing() {
+        when(writetimeTTLFeature.isEnabled()).thenReturn(true);
+        when(writetimeTTLFeature.hasWritetimeColumns()).thenReturn(true);
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        doThrow(new IllegalStateException("WriteTime set called unexpectedly")).when(boundStatement).set(anyInt(),
+                isNull(), eq(Long.class));
+
+        RuntimeException ex = assertThrows(RuntimeException.class,
+                () -> targetInsertStatement.bind(originRow, targetRow, null, null, null, null));
+        assertEquals(KnownProperties.ORIGIN_WRITETIME_NAMES + " specified, but no WriteTime value was provided",
+                ex.getMessage());
+    }
+
+    /**
+     * With both TTL and writetime enabled, bind() binds each target column at sequential indices and appends the TTL
+     * and writetime values at the two trailing indices, each with its expected type.
+     */
+    @Test
+    public void bind_checkBindInputsValidatesAndSequentialBindsWithTtlAndWritetime() {
+        when(writetimeTTLFeature.isEnabled()).thenReturn(true);
+        when(writetimeTTLFeature.hasTTLColumns()).thenReturn(true);
+        when(writetimeTTLFeature.hasWritetimeColumns()).thenReturn(true);
+
+        ArgumentCaptor<Integer> indexCaptor = ArgumentCaptor.forClass(Integer.class);
+        ArgumentCaptor<Object> valueCaptor = ArgumentCaptor.forClass(Object.class);
+        ArgumentCaptor<Class> classCaptor = ArgumentCaptor.forClass(Class.class);
+
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        Integer ttl = 7200;
+        Long writeTime = 42L;
+
+        BoundStatement result = targetInsertStatement.bind(originRow, targetRow, ttl, writeTime, null, null);
+        assertNotNull(result);
+
+        verify(boundStatement, times(targetColumnNames.size() + 2)).set(indexCaptor.capture(), valueCaptor.capture(),
+                classCaptor.capture());
+
+        List<Integer> allIndices = indexCaptor.getAllValues();
+        List<Object> allValues = valueCaptor.getAllValues();
+        List<Class> allClasses = classCaptor.getAllValues();
+
+        // Columns are bound at sequential indices.
+        for (int i = 0; i < targetColumnNames.size(); i++) {
+            assertEquals(i, allIndices.get(i).intValue());
+        }
+
+        // TTL is bound at the index immediately after the columns, with its value and Integer type.
+        int ttlIdx = targetColumnNames.size();
+        assertEquals(ttlIdx, allIndices.get(ttlIdx).intValue());
+        assertEquals(ttl, allValues.get(ttlIdx));
+        assertEquals(Integer.class, allClasses.get(ttlIdx));
+
+        // WriteTime is bound at the next index, with its value and Long type.
+        int wtIdx = targetColumnNames.size() + 1;
+        assertEquals(wtIdx, allIndices.get(wtIdx).intValue());
+        assertEquals(writeTime, allValues.get(wtIdx));
+        assertEquals(Long.class, allClasses.get(wtIdx));
+    }
+
+    /**
+     * Constant columns are not bound: bind() sets only the non-constant columns, at sequential indices starting from
+     * zero.
+     */
+    @Test
+    public void bind_withDebugDisabledConstantColumnsStillBindsCorrectly() {
+        commonSetup(false, true, false);
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        ArgumentCaptor<Integer> indexCaptor = ArgumentCaptor.forClass(Integer.class);
+
+        targetInsertStatement.bind(originRow, targetRow, null, null, null, null);
+
+        // Constant columns are skipped.
+        int expectedBindCount = targetColumnNames.size() - constantColumns.size();
+        verify(boundStatement, times(expectedBindCount)).set(indexCaptor.capture(), any(), any(Class.class));
+
+        List<Integer> capturedIndices = indexCaptor.getAllValues();
+        for (int i = 0; i < capturedIndices.size(); i++) {
+            assertEquals(i, capturedIndices.get(i).intValue());
+        }
+    }
+
+    /**
+     * With an explode-map column present, bind() sets every target column at sequential indices starting from zero.
+     */
+    @Test
+    public void bind_withExplodeMapSequentialIndices() {
+        commonSetup(true, false, false);
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        ArgumentCaptor<Integer> indexCaptor = ArgumentCaptor.forClass(Integer.class);
+
+        targetInsertStatement.bind(originRow, targetRow, null, null, getSampleData(explodeMapKeyType),
+                getSampleData(explodeMapValueType));
+
+        // Every column is bound at its sequential index.
+        verify(boundStatement, times(targetColumnNames.size())).set(indexCaptor.capture(), any(), any(Class.class));
+        List<Integer> capturedIndices = indexCaptor.getAllValues();
+        for (int i = 0; i < capturedIndices.size(); i++) {
+            assertEquals(i, capturedIndices.get(i).intValue());
+        }
+    }
+
+    /**
+     * bind() sets every target column exactly once, at sequential indices 0, 1, 2, ...
+     */
+    @Test
+    public void bind_verifiesSequentialBindIndicesForAllColumns() {
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        BoundStatement result = targetInsertStatement.bind(originRow, targetRow, null, null, null, null);
+
+        assertNotNull(result);
+
+        ArgumentCaptor<Integer> indexCaptor = ArgumentCaptor.forClass(Integer.class);
+        verify(boundStatement, times(targetColumnNames.size())).set(indexCaptor.capture(), any(), any(Class.class));
+
+        List<Integer> capturedIndices = indexCaptor.getAllValues();
+        assertEquals(targetColumnNames.size(), capturedIndices.size());
+        for (int i = 0; i < capturedIndices.size(); i++) {
+            assertEquals(i, capturedIndices.get(i).intValue());
+        }
+    }
+
+    /**
+     * The TTL value is bound at the index immediately following the last column, as an Integer.
+     */
+    @Test
+    public void bind_verifyTtlSetAtCorrectIndex() {
+        when(writetimeTTLFeature.isEnabled()).thenReturn(true);
+        when(writetimeTTLFeature.hasTTLColumns()).thenReturn(true);
+
+        ArgumentCaptor<Integer> indexCaptor = ArgumentCaptor.forClass(Integer.class);
+        ArgumentCaptor<Object> valueCaptor = ArgumentCaptor.forClass(Object.class);
+        ArgumentCaptor<Class> classCaptor = ArgumentCaptor.forClass(Class.class);
+
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        targetInsertStatement.bind(originRow, targetRow, 3600, null, null, null);
+
+        verify(boundStatement, times(targetColumnNames.size() + 1)).set(indexCaptor.capture(), valueCaptor.capture(),
+                classCaptor.capture());
+
+        List<Integer> indices = indexCaptor.getAllValues();
+        List<Object> values = valueCaptor.getAllValues();
+        List<Class> classes = classCaptor.getAllValues();
+
+        Integer ttlIndex = indices.get(indices.size() - 1);
+        Object ttlValue = values.get(values.size() - 1);
+        Class ttlClass = classes.get(classes.size() - 1);
+
+        assertEquals(targetColumnNames.size(), ttlIndex.intValue());
+        assertEquals(3600, ttlValue);
+        assertEquals(Integer.class, ttlClass);
+    }
+
+    /**
+     * The writetime value is bound at the trailing index, after all columns and the TTL, as a Long.
+     */
+    @Test
+    public void bind_verifyWritetimeSetAtCorrectIndex() {
+        when(writetimeTTLFeature.isEnabled()).thenReturn(true);
+        when(writetimeTTLFeature.hasTTLColumns()).thenReturn(true);
+        when(writetimeTTLFeature.hasWritetimeColumns()).thenReturn(true);
+
+        ArgumentCaptor<Integer> indexCaptor = ArgumentCaptor.forClass(Integer.class);
+        ArgumentCaptor<Object> valueCaptor = ArgumentCaptor.forClass(Object.class);
+        ArgumentCaptor<Class> classCaptor = ArgumentCaptor.forClass(Class.class);
+
+        targetInsertStatement = new TargetInsertStatement(propertyHelper, targetSession);
+
+        targetInsertStatement.bind(originRow, targetRow, 3600, 999L, null, null);
+
+        verify(boundStatement, times(targetColumnNames.size() + 2)).set(indexCaptor.capture(), valueCaptor.capture(),
+                classCaptor.capture());
+
+        List<Integer> indices = indexCaptor.getAllValues();
+        List<Object> values = valueCaptor.getAllValues();
+        List<Class> classes = classCaptor.getAllValues();
+
+        Integer wtIndex = indices.get(indices.size() - 1);
+        Object wtValue = values.get(values.size() - 1);
+        Class wtClass = classes.get(classes.size() - 1);
+
+        assertEquals(targetColumnNames.size() + 1, wtIndex.intValue());
+        assertEquals(999L, wtValue);
+        assertEquals(Long.class, wtClass);
+    }
+
+    /**
+     * When converting an origin value throws, bind() wraps the failure in a RuntimeException whose message identifies
+     * the value being bound.
+     */
+    @Test
+    public void bind_exceptionWithNonNullOriginValue() {
+        when(targetTable.getCorrespondingIndex(0)).thenReturn(0);
+        when(originTable.getAndConvertData(0, originRow)).thenThrow(new RuntimeException("binding error"));
+
+        RuntimeException ex = assertThrows(RuntimeException.class, () -> targetInsertStatement.bind(originRow,
+                targetRow, 3600, 123456789L, getSampleData(explodeMapKeyType), getSampleData(explodeMapValueType)));
+
+        assertTrue(ex.getMessage().contains("Error trying to bind value"));
+    }
+
 }