### Question:
ListingConnector implements IndexingConnector, ItemRetriever,
    IncrementalChangeHandler { @Override public void init(IndexingConnectorContext context) throws Exception { checkState(Configuration.isInitialized(), "configuration not initialized"); exceptionHandler = TraverseExceptionHandlerFactory.createFromConfig(); indexingService = checkNotNull(context.getIndexingService()); for (String traverseName : traverserConfigKey.get()) { context.registerTraverser( new TraverserConfiguration.Builder(traverseName).itemRetriever(this).build()); } defaultAcl = DefaultAcl.fromConfiguration(indexingService); threadPoolExecutor = new ThreadPoolExecutor( DEFAULT_THREAD_NUM, DEFAULT_THREAD_NUM, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10 * DEFAULT_THREAD_NUM), new ThreadPoolExecutor.CallerRunsPolicy()); listeningExecutorService = MoreExecutors.listeningDecorator(threadPoolExecutor); repositoryContext = new RepositoryContext.Builder() .setEventBus( new AsyncEventBus( "EventBus-" + ListingConnector.class.getName(), listeningExecutorService)) .setDefaultAclMode(defaultAcl.getDefaultAclMode()) .build(); if (checkpointHandler == null) { checkpointHandler = LocalFileCheckpointHandler.fromConfiguration(); } repositoryContext.getEventBus().register(this); repository.init(repositoryContext); } ListingConnector(Repository repository);  ListingConnector(Repository repository, CheckpointHandler checkpointHandler); @Override String getDefaultId(); @Override void init(IndexingConnectorContext context); @Override void traverse(); @Subscribe void handleAsyncOperation(AsyncApiOperation asyncOp); @Override void saveCheckpoint(boolean isShutdown); @Override void destroy(); @Override void process(Item item); @Override void handleIncrementalChanges(); static final String CONFIG_TRAVERSER; static final String CHECKPOINT_FULL; static final String CHECKPOINT_INCREMENTAL; }### Answer:
@Test public void testEventBusIsInitialized() throws Exception { setDefaultConfig(); ListingConnector connector = new ListingConnector(mockRepository); connector.init(mockConnectorContext); verify(mockRepository).init(repositoryContextCaptor.capture()); RepositoryContext rc = repositoryContextCaptor.getAllValues().get(0); rc.getEventBus().unregister(connector); }
@Test public void testConfigurationIsUninitialized() throws Exception { ListingConnector connector = new ListingConnector(mockRepository); try { connector.init(mockConnectorContext); fail(); } catch (IllegalStateException ex) { verifyNoMoreInteractions(mockRepository); } }
@Test public void testDefaultAclModeIsInitialized() throws Exception { Properties config = new Properties(); config.put(DefaultAcl.DEFAULT_ACL_MODE, "override"); config.put(DefaultAcl.DEFAULT_ACL_READERS_USERS, "google:user1@example.com"); overrideDefaultConfig(config); SettableFuture<Operation> updateFuture = SettableFuture.create(); updateFuture.set(new Operation().setDone(true)); when(mockIndexingService.indexItem(any(), any())).thenReturn(updateFuture); ListingConnector connector = new ListingConnector(mockRepository); connector.init(mockConnectorContext); verify(mockRepository).init(repositoryContextCaptor.capture()); RepositoryContext rc = repositoryContextCaptor.getAllValues().get(0); assertEquals(DefaultAclMode.OVERRIDE, rc.getDefaultAclMode()); }
@Test public void testTraverseConfiguration() throws Exception { Properties properties = new Properties(); properties.put(ListingConnector.CONFIG_TRAVERSER, "checkpoint, test"); overrideDefaultConfig(properties); ListingConnector connector = new ListingConnector(mockRepository); connector.init(mockConnectorContext); verify(mockConnectorContext, times(2)).registerTraverser(any()); }
@Test public void testTraverseConfigurationDefault() throws Exception { overrideDefaultConfig(new Properties()); ListingConnector connector = new ListingConnector(mockRepository); connector.init(mockConnectorContext); verify(mockConnectorContext, times(1)).registerTraverser(any()); }