AbstractSink implements ISink { @Override public void processTuples() throws TexeraException { if (cursor == CLOSED) { throw new DataflowException(ErrorMessages.OPERATOR_NOT_OPENED); } Tuple nextTuple; while ((nextTuple = inputOperator.getNextTuple()) != null) { processOneTuple(nextTuple); cursor++; } }  @Override void open(); void setInputOperator(IOperator inputOperator); IOperator getInputOperator(); @Override void processTuples(); @Override void close(); Schema getOutputSchema(); Schema transformToOutputSchema(Schema... inputSchema);  }
@Test public void testProcessTuples() throws Exception { Tuple sampleTuple = Mockito.mock(Tuple.class); Mockito.when(childOperator.getNextTuple()).thenReturn(sampleTuple).thenReturn(null); sink.open(); sink.processTuples(); sink.close(); Mockito.verify(childOperator, Mockito.times(2)).getNextTuple(); }