@KuduClusterTestContext(kuduClusterBasedTest = false)
@Test
public void testAssignPartitions() throws Exception
{

  AbstractKuduInputOperator<UnitTestTablePojo,InputOperatorControlTuple> mockedInputOperator =
      PowerMockito.mock(AbstractKuduInputOperator.class);
  when(mockedInputOperator.getNumberOfPartitions()).thenReturn(5);
  PowerMockito.mockStatic(KryoCloneUtils.class);
  when(KryoCloneUtils.cloneObject(mockedInputOperator)).thenReturn(mockedInputOperator);
  KuduOneToManyPartitioner kuduOneToManyPartitioner = new KuduOneToManyPartitioner(mockedInputOperator);
  buildMockWiring(mockedInputOperator, KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE);
  kuduOneToManyPartitioner.setPrototypeKuduInputOperator(mockedInputOperator);
  Map<Integer,List<KuduPartitionScanAssignmentMeta>> assignedPartitions = kuduOneToManyPartitioner.assign(
      kuduOneToManyPartitioner.getListOfPartitionAssignments(
      new ArrayList<Partitioner.Partition<AbstractKuduInputOperator>>(),
      partitioningContext),partitioningContext);
  assertThat(assignedPartitions.size(), is(5));
  for (List<KuduPartitionScanAssignmentMeta> eachOperatorassignment: assignedPartitions.values()) {
    assertThat(eachOperatorassignment.size(), lessThanOrEqualTo(3));
  }
  for (List<KuduPartitionScanAssignmentMeta> eachOperatorassignment: assignedPartitions.values()) {
    assertThat(eachOperatorassignment.size(), greaterThanOrEqualTo(2));
  }
}