### Question:
BlockPropagationManager { @VisibleForTesting CompletableFuture<Block> importOrSavePendingBlock(final Block block) { synchronized (pendingBlocks) { if (!protocolContext.getBlockchain().contains(block.getHeader().getParentHash())) { if (pendingBlocks.registerPendingBlock(block)) { LOG.info( "Saving announced block {} ({}) for future import", block.getHeader().getNumber(), block.getHash()); } return CompletableFuture.completedFuture(block); } } if (!importingBlocks.add(block.getHash())) { return CompletableFuture.completedFuture(block); } if (protocolContext.getBlockchain().contains(block.getHash())) { importingBlocks.remove(block.getHash()); return CompletableFuture.completedFuture(block); } final BlockHeader parent = protocolContext .getBlockchain() .getBlockHeader(block.getHeader().getParentHash()) .orElseThrow( () -> new IllegalArgumentException( "Incapable of retrieving header from non-existent parent of " + block.getHeader().getNumber() + ".")); final ProtocolSpec protocolSpec = protocolSchedule.getByBlockNumber(block.getHeader().getNumber()); final BlockHeaderValidator blockHeaderValidator = protocolSpec.getBlockHeaderValidator(); final BadBlockManager badBlockManager = protocolSpec.getBadBlocksManager(); return ethContext .getScheduler() .scheduleSyncWorkerTask( () -> validateAndProcessPendingBlock( blockHeaderValidator, block, parent, badBlockManager)); } BlockPropagationManager(
      final SynchronizerConfiguration config,
      final ProtocolSchedule protocolSchedule,
      final ProtocolContext protocolContext,
      final EthContext ethContext,
      final SyncState syncState,
      final PendingBlocks pendingBlocks,
      final MetricsSystem metricsSystem,
      final BlockBroadcaster blockBroadcaster); void start();  }### Answer:
@SuppressWarnings("unchecked") @Test public void shouldNotImportBlocksThatAreAlreadyBeingImported() { final EthScheduler ethScheduler = mock(EthScheduler.class); when(ethScheduler.scheduleSyncWorkerTask(any(Supplier.class))) .thenReturn(new CompletableFuture<>()); final EthContext ethContext = new EthContext( new EthPeers("eth", TestClock.fixed(), metricsSystem), new EthMessages(), ethScheduler); final BlockPropagationManager blockPropagationManager = new BlockPropagationManager( syncConfig, protocolSchedule, protocolContext, ethContext, syncState, pendingBlocks, metricsSystem, blockBroadcaster); blockchainUtil.importFirstBlocks(2); final Block nextBlock = blockchainUtil.getBlock(2); blockPropagationManager.importOrSavePendingBlock(nextBlock); blockPropagationManager.importOrSavePendingBlock(nextBlock); verify(ethScheduler, times(1)).scheduleSyncWorkerTask(any(Supplier.class)); }
@SuppressWarnings("unchecked") @Test public void shouldDetectAndCacheInvalidBlocks() { final EthScheduler ethScheduler = mock(EthScheduler.class); when(ethScheduler.scheduleSyncWorkerTask(any(Supplier.class))) .thenAnswer( new Answer<Object>() { @Override public Object answer(final InvocationOnMock invocation) throws Throwable { return invocation.getArgument(0, Supplier.class).get(); } }); final EthContext ethContext = new EthContext( new EthPeers("eth", TestClock.fixed(), metricsSystem), new EthMessages(), ethScheduler); final BlockPropagationManager blockPropagationManager = new BlockPropagationManager( syncConfig, protocolSchedule, protocolContext, ethContext, syncState, pendingBlocks, metricsSystem, blockBroadcaster); blockchainUtil.importFirstBlocks(2); final Block firstBlock = blockchainUtil.getBlock(1); final BadBlockManager badBlocksManager = protocolSchedule.getByBlockNumber(1).getBadBlocksManager(); final Block badBlock = new BlockDataGenerator() .block( BlockDataGenerator.BlockOptions.create() .setBlockNumber(1) .setParentHash(firstBlock.getHash()) .setBlockHeaderFunctions(new MainnetBlockHeaderFunctions())); assertThat(badBlocksManager.getBadBlocks()).isEmpty(); blockPropagationManager.importOrSavePendingBlock(badBlock); assertThat(badBlocksManager.getBadBlocks().size()).isEqualTo(1); verify(ethScheduler, times(1)).scheduleSyncWorkerTask(any(Supplier.class)); }