Add option to skip container PUT during upload

Currently, a user with read/write access to a container (but without
access to creat new containers) recieves a warning every time they
upload. Now, allow them to avoid the extra request and warning by
specifying --skip-container-put on the command line.

This is also useful when testing: developers can HEAD a container to
ensure it's in memcache, shut down all container servers, then upload
and creaate a bunch of async pendings. Previously, the 503 on container
PUT would prevent the object upload from even being attempted.

Closes-Bug: 1317956
Related-Bug: 1204558
Change-Id: I3d9129a0b6b65c6c6187ae6af003b221afceef47
Related-Change: If1f8a02ee7459ea2158ffa6e958f67d299ec529e

3 files changed, 102 insertions, 48 deletions

diff --git a/swiftclient/service.py b/swiftclient/service.py
index 8e2c7b0..685b748 100644
--- a/swiftclient/service.py
+++ b/swiftclient/service.py
@@ -202,6 +202,7 @@ _default_local_options = {
     'leave_segments': False,
     'changed': None,
     'skip_identical': False,
+    'skip_container_put': False,
     'version_id': None,
     'yes_all': False,
     'read_acl': None,
@@ -1462,6 +1463,7 @@ class SwiftService(object):
                                 'leave_segments': False,
                                 'changed': None,
                                 'skip_identical': False,
+                                'skip_container_put': False,
                                 'fail_fast': False,
                                 'dir_marker': False  # Only for None sources
                             }
@@ -1487,54 +1489,57 @@ class SwiftService(object):
         # the object name. (same as passing --object-name).
         container, _sep, pseudo_folder = container.partition('/')
 
-        # Try to create the container, just in case it doesn't exist. If this
-        # fails, it might just be because the user doesn't have container PUT
-        # permissions, so we'll ignore any error. If there's really a problem,
-        # it'll surface on the first object PUT.
-        policy_header = {}
-        _header = split_headers(options["header"])
-        if POLICY in _header:
-            policy_header[POLICY] = \
-                _header[POLICY]
-        create_containers = [
-            self.thread_manager.container_pool.submit(
-                self._create_container_job, container, headers=policy_header)
-        ]
+        if not options['skip_container_put']:
+            # Try to create the container, just in case it doesn't exist. If
+            # this fails, it might just be because the user doesn't have
+            # container PUT permissions, so we'll ignore any error. If there's
+            # really a problem, it'll surface on the first object PUT.
+            policy_header = {}
+            _header = split_headers(options["header"])
+            if POLICY in _header:
+                policy_header[POLICY] = \
+                    _header[POLICY]
+            create_containers = [
+                self.thread_manager.container_pool.submit(
+                    self._create_container_job, container,
+                    headers=policy_header)
+            ]
 
-        # wait for first container job to complete before possibly attempting
-        # segment container job because segment container job may attempt
-        # to HEAD the first container
-        for r in interruptable_as_completed(create_containers):
-            res = r.result()
-            yield res
+            # wait for first container job to complete before possibly
+            # attempting segment container job because segment container job
+            # may attempt to HEAD the first container
+            for r in interruptable_as_completed(create_containers):
+                res = r.result()
+                yield res
 
-        if segment_size:
-            seg_container = container + '_segments'
-            if options['segment_container']:
-                seg_container = options['segment_container']
-            if seg_container != container:
-                if not policy_header:
-                    # Since no storage policy was specified on the command
-                    # line, rather than just letting swift pick the default
-                    # storage policy, we'll try to create the segments
-                    # container with the same policy as the upload container
-                    create_containers = [
-                        self.thread_manager.container_pool.submit(
-                            self._create_container_job, seg_container,
-                            policy_source=container
-                        )
-                    ]
-                else:
-                    create_containers = [
-                        self.thread_manager.container_pool.submit(
-                            self._create_container_job, seg_container,
-                            headers=policy_header
-                        )
-                    ]
+            if segment_size:
+                seg_container = container + '_segments'
+                if options['segment_container']:
+                    seg_container = options['segment_container']
+                if seg_container != container:
+                    if not policy_header:
+                        # Since no storage policy was specified on the command
+                        # line, rather than just letting swift pick the default
+                        # storage policy, we'll try to create the segments
+                        # container with the same policy as the upload
+                        # container
+                        create_containers = [
+                            self.thread_manager.container_pool.submit(
+                                self._create_container_job, seg_container,
+                                policy_source=container
+                            )
+                        ]
+                    else:
+                        create_containers = [
+                            self.thread_manager.container_pool.submit(
+                                self._create_container_job, seg_container,
+                                headers=policy_header
+                            )
+                        ]
 
-                for r in interruptable_as_completed(create_containers):
-                    res = r.result()
-                    yield res
+                    for r in interruptable_as_completed(create_containers):
+                        res = r.result()
+                        yield res
 
         # We maintain a results queue here and a separate thread to monitor
         # the futures because we want to get results back from potential
diff --git a/swiftclient/shell.py b/swiftclient/shell.py
index 6da9d66..76473fd 100755
--- a/swiftclient/shell.py
+++ b/swiftclient/shell.py
@@ -985,8 +985,9 @@ def st_copy(parser, args, output_manager, return_parser=False):
 st_upload_options = '''[--changed] [--skip-identical] [--segment-size <size>]
                     [--segment-container <container>] [--leave-segments]
                     [--object-threads <thread>] [--segment-threads <threads>]
-                    [--meta <name:value>] [--header <header>] [--use-slo]
-                    [--ignore-checksum] [--object-name <object-name>]
+                    [--meta <name:value>] [--header <header>]
+                    [--use-slo] [--ignore-checksum] [--skip-container-put]
+                    [--object-name <object-name>]
                     <container> <file_or_directory> [<file_or_directory>] [...]
 '''
 
@@ -1032,11 +1033,13 @@ Optional arguments:
   --use-slo             When used in conjunction with --segment-size it will
                         create a Static Large Object instead of the default
                         Dynamic Large Object.
+  --ignore-checksum     Turn off checksum validation for uploads.
+  --skip-container-put  Assume all necessary containers already exist; don't
+                        automatically try to create them.
   --object-name <object-name>
                         Upload file and name object to <object-name> or upload
                         dir and use <object-name> as object prefix instead of
                         folder name.
-  --ignore-checksum     Turn off checksum validation for uploads.
 '''.strip('\n')
 
 
@@ -1052,6 +1055,10 @@ def st_upload(parser, args, output_manager, return_parser=False):
         default=False, help='Skip uploading files that are identical on '
         'both sides.')
     parser.add_argument(
+        '--skip-container-put', action='store_true', dest='skip_container_put',
+        default=False, help='Assume all necessary containers already exist; '
+        "don't automatically try to create them.")
+    parser.add_argument(
         '-S', '--segment-size', dest='segment_size', help='Upload files '
         'in segments no larger than <size> (in Bytes) and then create a '
         '"manifest" file that will download all the segments as if it were '
diff --git a/test/unit/test_shell.py b/test/unit/test_shell.py
index 295c918..2331eaa 100644
--- a/test/unit/test_shell.py
+++ b/test/unit/test_shell.py
@@ -912,6 +912,48 @@ class TestShell(unittest.TestCase):
             query_string='multipart-manifest=put',
             response_dict=mock.ANY)
 
+    @mock.patch('swiftclient.shell.walk')
+    @mock.patch('swiftclient.service.Connection')
+    def test_upload_skip_container_put(self, connection, walk):
+        connection.return_value.head_object.return_value = {
+            'content-length': '0'}
+        connection.return_value.put_object.return_value = EMPTY_ETAG
+        connection.return_value.attempts = 0
+        argv = ["", "upload", "container", "--skip-container-put",
+                self.tmpfile, "-H", "X-Storage-Policy:one",
+                "--meta", "Color:Blue"]
+        swiftclient.shell.main(argv)
+        connection.return_value.put_container.assert_not_called()
+
+        connection.return_value.put_object.assert_called_with(
+            'container',
+            self.tmpfile.lstrip('/'),
+            mock.ANY,
+            content_length=0,
+            headers={'x-object-meta-mtime': mock.ANY,
+                     'X-Storage-Policy': 'one',
+                     'X-Object-Meta-Color': 'Blue'},
+            response_dict={})
+
+        # Upload in segments
+        connection.return_value.head_container.return_value = {
+            'x-storage-policy': 'one'}
+        argv = ["", "upload", "container", "--skip-container-put",
+                self.tmpfile, "-S", "10"]
+        with open(self.tmpfile, "wb") as fh:
+            fh.write(b'12345678901234567890')
+        swiftclient.shell.main(argv)
+        # Both base and segments container are assumed to exist already
+        connection.return_value.put_container.assert_not_called()
+        connection.return_value.put_object.assert_called_with(
+            'container',
+            self.tmpfile.lstrip('/'),
+            '',
+            content_length=0,
+            headers={'x-object-manifest': mock.ANY,
+                     'x-object-meta-mtime': mock.ANY},
+            response_dict={})
+
     @mock.patch('swiftclient.service.SwiftService.upload')
     def test_upload_object_with_account_readonly(self, upload):
         argv = ["", "upload", "container", self.tmpfile]