Add a system test for 'Table.upload_from_file'.

google/cloud/bigquery/job.py

@@ -359,7 +359,12 @@ def begin(self, client=None):
  ``NoneType``
  :param client: the client to use. If not passed, falls back to the
  ``client`` stored on the current dataset.
+
+ :raises: :exc:`ValueError` if the job has already begin.
  """
+ if self.state is not None:
+ raise ValueError("Job already begun.")
+
  client = self._require_client(client)
  path = '/projects/%s/jobs' % (self.project,)
  api_response = client.connection.api_request(

google/cloud/bigquery/table.py

@@ -868,7 +868,8 @@ def upload_from_file(self,
 
  :rtype: :class:`google.cloud.bigquery.jobs.LoadTableFromStorageJob`
  :returns: the job instance used to load the data (e.g., for
- querying status).
+ querying status). Note that the job is already started:
+ do not call ``job.begin()``.
  :raises: :class:`ValueError` if ``size`` is not passed in and can not
  be determined, or if the ``file_obj`` can be detected to be
  a file opened in text mode.

system_tests/bigquery.py

@@ -288,6 +288,59 @@ def _has_rows(result):
  self.assertEqual(sorted(rows, key=by_age),
  sorted(ROWS, key=by_age))
 
+ def test_load_table_from_local_file_then_dump_table(self):
+ import csv
+ import tempfile
+ ROWS = [
+ ('Phred Phlyntstone', 32),
+ ('Bharney Rhubble', 33),
+ ('Wylma Phlyntstone', 29),
+ ('Bhettye Rhubble', 27),
+ ]
+ TABLE_NAME = 'test_table'
+
+ dataset = Config.CLIENT.dataset(DATASET_NAME)
+
+ retry_403(dataset.create)()
+ self.to_delete.append(dataset)
+
+ full_name = bigquery.SchemaField('full_name', 'STRING',
+ mode='REQUIRED')
+ age = bigquery.SchemaField('age', 'INTEGER', mode='REQUIRED')
+ table = dataset.table(TABLE_NAME, schema=[full_name, age])
+ table.create()
+ self.to_delete.insert(0, table)
+
+ with tempfile.NamedTemporaryFile(mode='w+') as csv_file:
+ writer = csv.writer(csv_file)
+ writer.writerow(('Full Name', 'Age'))
+ writer.writerows(ROWS)
+ csv_file.flush()
+
+ with open(csv_file.name, 'rb') as csv_read:
+ job = table.upload_from_file(
+ csv_read,
+ source_format='CSV',
+ skip_leading_rows=1,
+ create_disposition='CREATE_NEVER',
+ write_disposition='WRITE_EMPTY',
+ )
+
+ def _job_done(instance):
+ return instance.state.lower() == 'done'
+
+ # Retry until done.
+ retry = RetryInstanceState(_job_done, max_tries=8)
+ retry(job.reload)()
+
+ self.assertTrue(_job_done(job))
+ self.assertEqual(job.output_rows, len(ROWS))
+
+ rows, _, _ = table.fetch_data()
+ by_age = operator.itemgetter(1)
+ self.assertEqual(sorted(rows, key=by_age),
+ sorted(ROWS, key=by_age))
+
  def test_load_table_from_storage_then_dump_table(self):
  import csv
  import tempfile

unit_tests/bigquery/test_job.py

@@ -484,6 +484,16 @@ def test_from_api_repr_w_properties(self):
  self.assertTrue(dataset._client is client)
  self._verifyResourceProperties(dataset, RESOURCE)
 
+ def test_begin_w_already_running(self):
+ conn = _Connection()
+ client = _Client(project=self.PROJECT, connection=conn)
+ table = _Table()
+ job = self._makeOne(self.JOB_NAME, table, [self.SOURCE1], client)
+ job._properties['status'] = {'state': 'RUNNING'}
+
+ with self.assertRaises(ValueError):
+ job.begin()
+
  def test_begin_w_bound_client(self):
  PATH = 'projects/%s/jobs' % self.PROJECT
  RESOURCE = self._makeResource()