@Test
public void testImportBigQueryFromGcs() throws Exception {

  when(mockBigqueryTablesGet.execute()).thenReturn(fakeTable);
  final ArgumentCaptor<Job> jobCaptor = ArgumentCaptor.forClass(Job.class);
  doAnswer(
          new Answer<Job>() {
            @Override
            public Job answer(InvocationOnMock invocationOnMock) throws Throwable {
              verify(mockBigqueryJobs, times(1)).insert(eq(jobProjectId), jobCaptor.capture());
              return jobCaptor.getValue();
            }
          })
      .when(mockBigqueryJobsInsert)
      .execute();
  when(mockBigqueryJobsGet.execute()).thenReturn(jobHandle);
  // Run importBigQueryFromGcs method.
  helper.importFromGcs(
      jobProjectId,
      tableRef,
      fakeTableSchema,
      /* timePartitioning= */ null,
      kmsKeyName,
      BigQueryFileFormat.NEWLINE_DELIMITED_JSON,
      BigQueryConfiguration.OUTPUT_TABLE_CREATE_DISPOSITION.getDefault(),
      BigQueryConfiguration.OUTPUT_TABLE_WRITE_DISPOSITION.getDefault(),
      ImmutableList.of("test-import-path"),
      true);
  // Verify correct calls to BigQuery are made.
  verify(mockBigquery, times(2)).jobs();
  // Verify correct calls to BigQuery.Jobs are made.
  verify(mockBigqueryJobs, times(1))
      .get(eq(jobProjectId), eq(jobCaptor.getValue().getJobReference().getJobId()));
  Job job = jobCaptor.getValue();
  assertThat(job.getConfiguration().getLoad().getSourceUris()).contains("test-import-path");
  assertThat(job.getConfiguration().getLoad().getDestinationTable()).isEqualTo(tableRef);
  assertThat(job.getJobReference().getLocation()).isEqualTo("test_location");
  // Verify we poll for job in correct location
  verify(mockBigqueryJobsGet).setLocation(eq("test_location"));
  // Verify correct calls to BigQuery.Jobs.Get are made.
  verify(mockBigqueryJobsGet, times(1)).execute();
  // Verify correct calls to BigQuery.Jobs.Insert are made.
  verify(mockBigqueryJobsInsert, times(1)).execute();
  verify(mockBigquery).datasets();
}