InfluxDbIO { public static Read read() { return new AutoValue_InfluxDbIO_Read.Builder() .setDisableCertificateValidation(false) .setRetentionPolicy(DEFAULT_RETENTION_POLICY) .build(); } private  InfluxDbIO(); static Write write(); static Read read(); static InfluxDB getConnection(
      DataSourceConfiguration configuration, boolean disableCertificateValidation);  }
@Test public void validateReadTest() { String influxHost = "http: String userName = "admin"; String password = "admin"; String influxDatabaseName = "testDataBase"; 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); doReturn(getDatabase(influxDatabaseName)).when(influxDb).query(new Query("SHOW DATABASES")); doReturn(getDatabase(influxDatabaseName)).when(influxDb).query(new Query("SHOW SHARDS")); doReturn(mockResultForNumberAndSizeOfBlocks()) .when(influxDb) .query(new Query("EXPLAIN SELECT * FROM cpu", influxDatabaseName)); doReturn(mockResult("cpu", 20)) .when(influxDb) .query(new Query("SELECT * FROM cpu", influxDatabaseName)); PCollection<Long> data = pipeline .apply( "Read data to InfluxDB", InfluxDbIO.read() .withDataSourceConfiguration( DataSourceConfiguration.create( StaticValueProvider.of(influxHost), StaticValueProvider.of(userName), StaticValueProvider.of(password))) .withDatabase(influxDatabaseName) .withQuery("SELECT * FROM cpu")) .apply(Count.globally()); PAssert.that(data).containsInAnyOrder(20L); PipelineResult result = pipeline.run(); Assert.assertEquals(State.DONE, result.waitUntilFinish()); }