Allow the sdk_location to point to a tarball as well as a directory

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=117060287

Author: robertwb

google/cloud/dataflow/utils/dependency.py

@@ -322,10 +322,13 @@ def stage_job_resources(options, file_copy=_dependency_file_copy,
       # used by tests running with the SDK built at head.
       if setup_options.sdk_location == 'default':
         module_path = os.path.abspath(__file__)
-        sdk_dir = os.path.join(os.path.dirname(module_path), '..')
+        sdk_path = os.path.join(
+            os.path.dirname(module_path), '..', names.DATAFLOW_SDK_TARBALL_FILE)
+      elif os.path.isdir(setup_options.sdk_location):
+        sdk_path = os.path.join(
+            setup_options.sdk_location, names.DATAFLOW_SDK_TARBALL_FILE)
       else:
-        sdk_dir = setup_options.sdk_location
-      sdk_path = os.path.join(sdk_dir, names.DATAFLOW_SDK_TARBALL_FILE)
+        sdk_path = setup_options.sdk_location
       if os.path.isfile(sdk_path):
         logging.info('Copying dataflow SDK "%s" to staging location.', sdk_path)
         file_copy(sdk_path, staged_path)

google/cloud/dataflow/utils/dependency_test.py

@@ -261,7 +261,7 @@ def test_sdk_location_local_not_present(self):
     self.assertEqual(
         'The file "%s" cannot be found. Its '
         'directory was specified by the --sdk_location command-line option.' %
-        os.path.join(sdk_location, names.DATAFLOW_SDK_TARBALL_FILE),
+        sdk_location,
         cm.exception.message)
 
   def test_sdk_location_gcs(self):