-
-
Save clayg/a82f475edf46e61eb4b6 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
diff --git a/tests/unit/test_service.py b/tests/unit/test_service.py | |
index c478351..8b74b64 100644 | |
--- a/tests/unit/test_service.py | |
+++ b/tests/unit/test_service.py | |
@@ -12,8 +12,10 @@ | |
# implied. | |
# See the License for the specific language governing permissions and | |
# limitations under the License. | |
+import contextlib | |
import mock | |
import os | |
+import shutil | |
import tempfile | |
import testtools | |
from hashlib import md5 | |
@@ -22,7 +24,7 @@ from six.moves.queue import Queue, Empty as QueueEmptyError | |
import swiftclient | |
from swiftclient.service import SwiftService, SwiftError | |
-from swiftclient.client import Connection | |
+from swiftclient.client import Connection, ClientException | |
clean_os_environ = {} | |
@@ -539,6 +541,18 @@ class TestSwiftUploadObject(testtools.TestCase): | |
self.assertRaises(SwiftError, self.suo, []) | |
+@contextlib.contextmanager | |
+def in_temp_dir(): | |
+ cwd = os.getcwd() | |
+ tempdir = tempfile.mkdtemp() | |
+ try: | |
+ os.chdir(tempdir) | |
+ yield | |
+ finally: | |
+ os.chdir(cwd) | |
+ shutil.rmtree(tempdir) | |
+ | |
+ | |
class TestService(testtools.TestCase): | |
def test_upload_with_bad_segment_size(self): | |
@@ -551,3 +565,23 @@ class TestService(testtools.TestCase): | |
except SwiftError as exc: | |
self.assertEqual('Segment size should be an integer value', | |
exc.value) | |
+ | |
+ def test_uplaod_with_realative_path(self): | |
+ service = SwiftService({}) | |
+ with mock.patch('swiftclient.service.Connection') as mock_conn: | |
+ # make the segment check request 404 | |
+ mock_conn.return_value.head_object.side_effect = ClientException( | |
+ 'Not Found', http_status=404) | |
+ with in_temp_dir(): | |
+ with open('./test', 'w') as f: | |
+ f.write('asdf') | |
+ resp_iter = service.upload('c', ['./test']) | |
+ responses = [x for x in resp_iter] | |
+ for resp in responses: | |
+ self.assertTrue(resp['success']) | |
+ self.assertEqual(2, len(responses)) | |
+ create_container_resp, upload_object_resp = responses | |
+ self.assertEqual(create_container_resp['action'], 'create_container') | |
+ self.assertEqual(upload_object_resp['action'], 'upload_object') | |
+ self.assertEqual(upload_object_resp['object'], 'test') | |
+ self.assertEqual(upload_object_resp['path'], './test') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment