AddElementsHandler implements OperationHandler<AddElements> { @Override public Void doOperation(final AddElements operation, final Context context, final Store store) throws OperationException { addElements(operation, (HBaseStore) store); return null; }  @Override Void doOperation(final AddElements operation,
                            final Context context, final Store store);  }
@Test public void shouldAddElements() throws OperationException, StoreException, IOException { final AddElementsHandler handler = new AddElementsHandler(); final List<Element> elements = createElements(); final List<Element> elementsWithNull = new ArrayList<>(elements); elementsWithNull.add(null); final AddElements addElements = new AddElements.Builder() .input(elementsWithNull) .build(); final Context context = mock(Context.class); final HBaseStore store = mock(HBaseStore.class); final HTable table = mock(HTable.class); given(store.getTable()).willReturn(table); final HBaseProperties properties = HBaseProperties.loadStoreProperties(StreamUtil.storeProps(getClass())); final int writeBufferSize = 5; properties.setWriteBufferSize(writeBufferSize); given(store.getProperties()).willReturn(properties); given(store.getSchema()).willReturn(SCHEMA); handler.doOperation(addElements, context, store); final ArgumentCaptor<List<Put>> putsCaptor = (ArgumentCaptor) ArgumentCaptor.forClass(List.class); verify(table, times(2)).put(putsCaptor.capture()); verify(table, times(2)).flushCommits(); final List<List<Put>> allPuts = putsCaptor.getAllValues(); assertEquals(2, allPuts.size()); final List<Put> combinedPuts = new ArrayList<>(); combinedPuts.addAll(allPuts.get(0)); combinedPuts.addAll(allPuts.get(1)); final List<Element> expectedElements = new ArrayList<>(); for (final Element element : elements) { expectedElements.add(element); if (element instanceof Edge && !((Edge) element).getSource().equals(((Edge) element).getDestination())) { expectedElements.add(element); } } final Element[] expectedElementsArr = expectedElements.toArray(new Element[expectedElements.size()]); final List<Element> elementsAdded = CellUtil.getElements(combinedPuts, new ElementSerialisation(SCHEMA), false); assertEquals(expectedElements.size(), elementsAdded.size()); assertThat(elementsAdded, IsCollectionContaining.hasItems(expectedElementsArr)); }
@Test public void shouldDoNothingIfNoElementsProvided() throws OperationException, StoreException, IOException { final AddElementsHandler handler = new AddElementsHandler(); final AddElements addElements = new AddElements(); final Context context = mock(Context.class); final HBaseStore store = mock(HBaseStore.class); final Table table = mock(Table.class); given(store.getTable()).willReturn(table); final HBaseProperties properties = HBaseProperties.loadStoreProperties(StreamUtil.storeProps(getClass())); given(store.getProperties()).willReturn(properties); given(store.getSchema()).willReturn(SCHEMA); handler.doOperation(addElements, context, store); verify(table, never()).put(any(Put.class)); }
@Test public void shouldThrowNoExceptionsWhenValidateFlagSetToFalse() throws OperationException, StoreException { final AddElements addElements = new AddElements.Builder() .input(new Edge("Unknown group", "source", "dest", true)) .validate(false) .build(); final AddElementsHandler handler = new AddElementsHandler(); final Context context = mock(Context.class); final HBaseStore store = mock(HBaseStore.class); final Table table = mock(Table.class); given(store.getTable()).willReturn(table); final HBaseProperties properties = HBaseProperties.loadStoreProperties(StreamUtil.storeProps(getClass())); given(store.getProperties()).willReturn(properties); given(store.getSchema()).willReturn(SCHEMA); handler.doOperation(addElements, context, store); }