Add use_default_tempdir cloud config options

aboudreault · aboudreault · commit c2bf6255cb8c · 2020-05-21T11:09:58.000-04:00
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -5,6 +5,7 @@ Not released
 Features
 --------
 * Make geomet an optional dependency at runtime (PYTHON-1237)
+* Add use_default_tempdir cloud config options (PYTHON-1245)
 
 Bug Fixes
 ---------
diff --git a/cassandra/datastax/cloud/__init__.py b/cassandra/datastax/cloud/__init__.py
@@ -97,8 +97,9 @@ def get_cloud_config(cloud_config, create_pyopenssl_context=False):
 
 def read_cloud_config_from_zip(cloud_config, create_pyopenssl_context):
     secure_bundle = cloud_config['secure_connect_bundle']
+    use_default_tempdir = cloud_config.get('use_default_tempdir', None)
     with ZipFile(secure_bundle) as zipfile:
-        base_dir = os.path.dirname(secure_bundle)
+        base_dir = tempfile.gettempdir() if use_default_tempdir else os.path.dirname(secure_bundle)
         tmp_dir = tempfile.mkdtemp(dir=base_dir)
         try:
             zipfile.extractall(path=tmp_dir)
diff --git a/tests/unit/advanced/cloud/test_cloud.py b/tests/unit/advanced/cloud/test_cloud.py
@@ -6,23 +6,29 @@
 # You may obtain a copy of the License at
 #
 # http://www.datastax.com/terms/datastax-dse-driver-license-terms
+import tempfile
+import os
+import shutil
+import six
+
 try:
     import unittest2 as unittest
 except ImportError:
     import unittest  # noqa
 
-import os
-
+from cassandra import DriverException
 from cassandra.datastax import cloud
 
 from mock import patch
 
+from tests import notwindows
 
 class CloudTests(unittest.TestCase):
 
     current_path = os.path.dirname(os.path.abspath(__file__))
+    creds_path = os.path.join(current_path, './creds.zip')
     config_zip = {
-        'secure_connect_bundle': os.path.join(current_path, './creds.zip')
+        'secure_connect_bundle': creds_path
     }
     metadata_json = """
         {"region":"local",
@@ -75,3 +81,33 @@ def test_parse_metadata_info(self):
         ]
         for host_id in host_ids:
             self.assertIn(host_id, config.host_ids)
+
+    @notwindows
+    def test_use_default_tempdir(self):
+        tmpdir = tempfile.mkdtemp()
+
+        def clean_tmp_dir():
+            os.chmod(tmpdir, 0o777)
+            shutil.rmtree(tmpdir)
+        self.addCleanup(clean_tmp_dir)
+
+        tmp_creds_path = os.path.join(tmpdir, 'creds.zip')
+        shutil.copyfile(self.creds_path, tmp_creds_path)
+        os.chmod(tmpdir, 0o544)
+        config = {
+            'secure_connect_bundle': tmp_creds_path
+        }
+
+        # The directory is not writtable.. we expect a permission error
+        exc = PermissionError if six.PY3 else OSError
+        with self.assertRaises(exc):
+            cloud.get_cloud_config(config)
+
+        # With use_default_tempdir, we expect an connection refused
+        # since the cluster doesn't exist
+        with self.assertRaises(DriverException):
+            config = {
+                'secure_connect_bundle': tmp_creds_path,
+                'use_default_tempdir': True
+            }
+            cloud.get_cloud_config(config)
