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()