InfluxDbIO { public static Write write() { return new AutoValue_InfluxDbIO_Write.Builder() .setRetentionPolicy(DEFAULT_RETENTION_POLICY) .setDisableCertificateValidation(false) .setBatchSize(DEFAULT_BUFFER_LIMIT) .setConsistencyLevel(ConsistencyLevel.QUORUM) .build(); } private  InfluxDbIO(); static Write write(); static Read read(); static InfluxDB getConnection(
      DataSourceConfiguration configuration, boolean disableCertificateValidation);  }
@Test public void validateWriteTest() { InfluxDB influxDb = Mockito.mock(InfluxDB.class); PowerMockito.when( InfluxDBFactory.connect( anyString(), anyString(), anyString(), any(OkHttpClient.Builder.class))) .thenReturn(influxDb); PowerMockito.when(InfluxDBFactory.connect(anyString(), anyString(), anyString())) .thenReturn(influxDb); String influxHost = "http: String userName = "admin"; String password = "admin"; String influxDatabaseName = "testDataBase"; AtomicInteger countInvocation = new AtomicInteger(); Mockito.doAnswer(invocation -> countInvocation.getAndIncrement()) .when(influxDb) .write(any(List.class)); doReturn(getDatabase(influxDatabaseName)).when(influxDb).query(new Query("SHOW DATABASES")); final int numOfElementsToWrite = 1000; pipeline .apply("Generate data", Create.of(GenerateData.getMetric("test_m", numOfElementsToWrite))) .apply( "Write data to InfluxDB", InfluxDbIO.write() .withDataSourceConfiguration( DataSourceConfiguration.create( StaticValueProvider.of(influxHost), StaticValueProvider.of(userName), StaticValueProvider.of(password))) .withDatabase(influxDatabaseName)); PipelineResult result = pipeline.run(); Assert.assertEquals(State.DONE, result.waitUntilFinish()); Assert.assertTrue(countInvocation.get() > 0); }