### Question:
BaseStreamStep extends BaseStep { @Override public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException { Preconditions.checkArgument( first, BaseMessages.getString( PKG, "BaseStreamStep.ProcessRowsError" ) ); Preconditions.checkNotNull( source ); Preconditions.checkNotNull( window ); source.open(); bufferStream().forEach( result -> { if ( result.isSafeStop() ) { getTrans().safeStop(); } putRows( result.getRows() ); } ); super.setOutputDone(); source.close(); return false; } BaseStreamStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
                         TransMeta transMeta, Trans trans ); boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); @Override void setOutputDone(); @Override boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); @Override void stopRunning( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ); @Override void resumeRunning(); @Override void pauseRunning(); @Override Collection<StepStatus> subStatuses(); @VisibleForTesting StreamSource<List<Object>> getSource(); @VisibleForTesting void setSource( StreamSource<List<Object>> source );  }### Answer:
@Test public void testStop() throws KettleException { Result result = new Result(); result.setSafeStop( false ); result.setRows( Collections.emptyList() ); when( streamWindow.buffer( any() ) ).thenReturn( Arrays.asList( result ) ); baseStreamStep.processRow( meta, stepData ); assertFalse( baseStreamStep.isSafeStopped() ); verify( streamSource ).close(); }
@Test public void testSafeStop() throws KettleException { Result result = new Result(); result.setSafeStop( true ); when( streamWindow.buffer( any() ) ).thenReturn( Arrays.asList( result ) ); baseStreamStep.processRow( meta, stepData ); assertTrue( baseStreamStep.isSafeStopped() ); verify( streamSource, times( 2 ) ).close(); }