Merge "Move all DLO functionality to middleware"

doc/saio/swift/proxy-server.conf

@@ -8,7 +8,7 @@ eventlet_debug = true
 [pipeline:main]
 # Yes, proxy-logging appears twice. This is so that
 # middleware-originated requests get logged too.
-pipeline = catch_errors gatekeeper healthcheck proxy-logging cache bulk slo ratelimit crossdomain tempurl tempauth staticweb container-quotas account-quotas proxy-logging proxy-server
+pipeline = catch_errors gatekeeper healthcheck proxy-logging cache bulk slo dlo ratelimit crossdomain tempurl tempauth staticweb container-quotas account-quotas proxy-logging proxy-server
 
 [filter:catch_errors]
 use = egg:swift#catch_errors
@@ -28,6 +28,9 @@ use = egg:swift#ratelimit
 [filter:crossdomain]
 use = egg:swift#crossdomain
 
+[filter:dlo]
+use = egg:swift#dlo
+
 [filter:slo]
 use = egg:swift#slo
 

doc/source/middleware.rst

@@ -160,6 +160,13 @@ Static Large Objects
     :members:
     :show-inheritance:
 
+Dynamic Large Objects
+====================
+
+.. automodule:: swift.common.middleware.dlo
+    :members:
+    :show-inheritance:
+
 List Endpoints
 ==============
 

etc/proxy-server.conf-sample

@@ -69,7 +69,7 @@
 # eventlet_debug = false
 
 [pipeline:main]
-pipeline = catch_errors gatekeeper healthcheck proxy-logging cache container_sync bulk slo ratelimit tempauth container-quotas account-quotas proxy-logging proxy-server
+pipeline = catch_errors gatekeeper healthcheck proxy-logging cache container_sync bulk slo dlo ratelimit tempauth container-quotas account-quotas proxy-logging proxy-server
 
 [app:proxy-server]
 use = egg:swift#proxy
@@ -143,14 +143,6 @@ use = egg:swift#proxy
 # Depth of the proxy put queue.
 # put_queue_depth = 10
 #
-# Start rate-limiting object segment serving after the Nth segment of a
-# segmented object.
-# rate_limit_after_segment = 10
-#
-# Once segment rate-limiting kicks in for an object, limit segments served
-# to N per second.
-# rate_limit_segments_per_sec = 1
-#
 # Storage nodes can be chosen at random (shuffle), by using timing
 # measurements (timing), or by using an explicit match (affinity).
 # Using timing measurements may allow for lower overall latency, while
@@ -537,6 +529,22 @@ use = egg:swift#slo
 # Time limit on GET requests (seconds)
 # max_get_time = 86400
 
+# Note: Put before both ratelimit and auth in the pipeline, but after
+# gatekeeper, catch_errors, and proxy_logging (the first instance).
+# If you don't put it in the pipeline, it will be inserted for you.
+[filter:dlo]
+use = egg:swift#dlo
+# Start rate-limiting DLO segment serving after the Nth segment of a
+# segmented object.
+# rate_limit_after_segment = 10
+#
+# Once segment rate-limiting kicks in for an object, limit segments served
+# to N per second. 0 means no rate-limiting.
+# rate_limit_segments_per_sec = 1
+#
+# Time limit on GET requests (seconds)
+# max_get_time = 86400
+
 [filter:account-quotas]
 use = egg:swift#account_quotas
 

setup.cfg

@@ -83,6 +83,7 @@ paste.filter_factory =
     container_quotas = swift.common.middleware.container_quotas:filter_factory
     account_quotas = swift.common.middleware.account_quotas:filter_factory
     proxy_logging = swift.common.middleware.proxy_logging:filter_factory
+    dlo = swift.common.middleware.dlo:filter_factory
     slo = swift.common.middleware.slo:filter_factory
     list_endpoints = swift.common.middleware.list_endpoints:filter_factory
     gatekeeper = swift.common.middleware.gatekeeper:filter_factory

swift/common/constraints.py

@@ -145,18 +145,6 @@ def check_object_creation(req, object_name):
     if not check_utf8(req.headers['Content-Type']):
         return HTTPBadRequest(request=req, body='Invalid Content-Type',
                               content_type='text/plain')
-    if 'x-object-manifest' in req.headers:
-        value = req.headers['x-object-manifest']
-        container = prefix = None
-        try:
-            container, prefix = value.split('/', 1)
-        except ValueError:
-            pass
-        if not container or not prefix or '?' in value or '&' in value or \
-                prefix[0] == '/':
-            return HTTPBadRequest(
-                request=req,
-                body='X-Object-Manifest must in the format container/prefix')
     return check_metadata(req, 'object')
 
 

swift/common/middleware/dlo.py

@@ -0,0 +1,332 @@
+# Copyright (c) 2013 OpenStack Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from ConfigParser import ConfigParser, NoSectionError, NoOptionError
+from hashlib import md5
+from swift.common.constraints import CONTAINER_LISTING_LIMIT
+from swift.common.exceptions import ListingIterError
+from swift.common.http import is_success
+from swift.common.swob import Request, Response, \
+    HTTPRequestedRangeNotSatisfiable, HTTPBadRequest
+from swift.common.utils import get_logger, json, SegmentedIterable, \
+    RateLimitedIterator, read_conf_dir, quote
+from swift.common.wsgi import WSGIContext
+from urllib import unquote
+
+
+class GetContext(WSGIContext):
+    def __init__(self, dlo, logger):
+        super(GetContext, self).__init__(dlo.app)
+        self.dlo = dlo
+        self.logger = logger
+
+    def _get_container_listing(self, req, version, account, container,
+                               prefix, marker=''):
+        con_req = req.copy_get()
+        con_req.script_name = ''
+        con_req.range = None
+        con_req.path_info = '/'.join(['', version, account, container])
+        con_req.query_string = 'format=json&prefix=%s' % quote(prefix)
+        con_req.user_agent = '%s DLO MultipartGET' % con_req.user_agent
+        if marker:
+            con_req.query_string += '&marker=%s' % quote(marker)
+
+        con_resp = con_req.get_response(self.dlo.app)
+        if not is_success(con_resp.status_int):
+            return con_resp, None
+        return None, json.loads(''.join(con_resp.app_iter))
+
+    def _segment_listing_iterator(self, req, version, account, container,
+                                  prefix, segments, first_byte=None,
+                                  last_byte=None):
+        # It's sort of hokey that this thing takes in the first page of
+        # segments as an argument, but we need to compute the etag and content
+        # length from the first page, and it's better to have a hokey
+        # interface than to make redundant requests.
+        if first_byte is None:
+            first_byte = 0
+        if last_byte is None:
+            last_byte = float("inf")
+
+        marker = ''
+        while True:
+            for segment in segments:
+                seg_length = int(segment['bytes'])
+
+                if first_byte >= seg_length:
+                    # don't need any bytes from this segment
+                    first_byte = max(first_byte - seg_length, -1)
+                    last_byte = max(last_byte - seg_length, -1)
+                    continue
+                elif last_byte < 0:
+                    # no bytes are needed from this or any future segment
+                    break
+
+                seg_name = segment['name']
+                if isinstance(seg_name, unicode):
+                    seg_name = seg_name.encode("utf-8")
+
+                # (obj path, etag, size, first byte, last byte)
+                yield ("/" + "/".join((version, account, container,
+                                       seg_name)),
+                       # We deliberately omit the etag and size here;
+                       # SegmentedIterable will check size and etag if
+                       # specified, but we don't want it to. DLOs only care
+                       # that the objects' names match the specified prefix.
+                       None, None,
+                       (None if first_byte <= 0 else first_byte),
+                       (None if last_byte >= seg_length - 1 else last_byte))
+
+                first_byte = max(first_byte - seg_length, -1)
+                last_byte = max(last_byte - seg_length, -1)
+
+            if len(segments) < CONTAINER_LISTING_LIMIT:
+                # a short page means that we're done with the listing
+                break
+            elif last_byte < 0:
+                break
+
+            marker = segments[-1]['name']
+            error_response, segments = self._get_container_listing(
+                req, version, account, container, prefix, marker)
+            if error_response:
+                # we've already started sending the response body to the
+                # client, so all we can do is raise an exception to make the
+                # WSGI server close the connection early
+                raise ListingIterError(
+                    "Got status %d listing container /%s/%s" %
+                    (error_response.status_int, account, container))
+
+    def get_or_head_response(self, req, x_object_manifest,
+                             response_headers=None):
+        if response_headers is None:
+            response_headers = self._response_headers
+
+        container, obj_prefix = x_object_manifest.split('/', 1)
+        container = unquote(container)
+        obj_prefix = unquote(obj_prefix)
+
+        # manifest might point to a different container
+        req.acl = None
+        version, account, _junk = req.split_path(2, 3, True)
+        error_response, segments = self._get_container_listing(
+            req, version, account, container, obj_prefix)
+        if error_response:
+            return error_response
+        have_complete_listing = len(segments) < CONTAINER_LISTING_LIMIT
+
+        first_byte = last_byte = None
+        content_length = None
+        if req.range and len(req.range.ranges) == 1:
+            content_length = sum(o['bytes'] for o in segments)
+
+            # This is a hack to handle suffix byte ranges (e.g. "bytes=-5"),
+            # which we can't honor unless we have a complete listing.
+            _junk, range_end = req.range.ranges_for_length(float("inf"))[0]
+
+            # If this is all the segments, we know whether or not this
+            # range request is satisfiable.
+            #
+            # Alternately, we may not have all the segments, but this range
+            # falls entirely within the first page's segments, so we know
+            # whether or not it's satisfiable.
+            if have_complete_listing or range_end < content_length:
+                byteranges = req.range.ranges_for_length(content_length)
+                if not byteranges:
+                    return HTTPRequestedRangeNotSatisfiable(request=req)
+                first_byte, last_byte = byteranges[0]
+                # For some reason, swob.Range.ranges_for_length adds 1 to the
+                # last byte's position.
+                last_byte -= 1
+            else:
+                # The range may or may not be satisfiable, but we can't tell
+                # based on just one page of listing, and we're not going to go
+                # get more pages because that would use up too many resources,
+                # so we ignore the Range header and return the whole object.
+                content_length = None
+                req.range = None
+
+        response_headers = [
+            (h, v) for h, v in response_headers
+            if h.lower() not in ("content-length", "content-range")]
+
+        if content_length is not None:
+            # Here, we have to give swob a big-enough content length so that
+            # it can compute the actual content length based on the Range
+            # header. This value will not be visible to the client; swob will
+            # substitute its own Content-Length.
+            #
+            # Note: if the manifest points to at least CONTAINER_LISTING_LIMIT
+            # segments, this may be less than the sum of all the segments'
+            # sizes. However, it'll still be greater than the last byte in the
+            # Range header, so it's good enough for swob.
+            response_headers.append(('Content-Length', str(content_length)))
+        elif have_complete_listing:
+            response_headers.append(('Content-Length',
+                                     str(sum(o['bytes'] for o in segments))))
+
+        if have_complete_listing:
+            response_headers = [(h, v) for h, v in response_headers
+                                if h.lower() != "etag"]
+            etag = md5()
+            for seg_dict in segments:
+                etag.update(seg_dict['hash'].strip('"'))
+            response_headers.append(('Etag', '"%s"' % etag.hexdigest()))
+
+        listing_iter = RateLimitedIterator(
+            self._segment_listing_iterator(
+                req, version, account, container, obj_prefix, segments,
+                first_byte=first_byte, last_byte=last_byte),
+            self.dlo.rate_limit_segments_per_sec,
+            limit_after=self.dlo.rate_limit_after_segment)
+        resp = Response(request=req, headers=response_headers,
+                        conditional_response=True,
+                        app_iter=SegmentedIterable(
+                            req, self.dlo.app, listing_iter,
+                            ua_suffix="DLO MultipartGET",
+                            name=req.path, logger=self.logger,
+                            max_get_time=self.dlo.max_get_time))
+        resp.app_iter.response = resp
+        return resp
+
+    def handle_request(self, req, start_response):
+        """
+        Take a GET or HEAD request, and if it is for a dynamic large object
+        manifest, return an appropriate response.
+
+        Otherwise, simply pass it through.
+        """
+        resp_iter = self._app_call(req.environ)
+
+        # make sure this response is for a dynamic large object manifest
+        for header, value in self._response_headers:
+            if (header.lower() == 'x-object-manifest'):
+                response = self.get_or_head_response(req, value)
+                return response(req.environ, start_response)
+        else:
+            # Not a dynamic large object manifest; just pass it through.
+            start_response(self._response_status,
+                           self._response_headers,
+                           self._response_exc_info)
+            return resp_iter
+
+
+class DynamicLargeObject(object):
+    def __init__(self, app, conf):
+        self.app = app
+        self.logger = get_logger(conf, log_route='dlo')
+
+        # DLO functionality used to live in the proxy server, not middleware,
+        # so let's try to go find config values in the proxy's config section
+        # to ease cluster upgrades.
+        self._populate_config_from_old_location(conf)
+
+        self.max_get_time = int(conf.get('max_get_time', '86400'))
+        self.rate_limit_after_segment = int(conf.get(
+            'rate_limit_after_segment', '10'))
+        self.rate_limit_segments_per_sec = int(conf.get(
+            'rate_limit_segments_per_sec', '1'))
+
+    def _populate_config_from_old_location(self, conf):
+        if ('rate_limit_after_segment' in conf or
+                'rate_limit_segments_per_sec' in conf or
+                'max_get_time' in conf or
+                '__file__' not in conf):
+            return
+
+        cp = ConfigParser()
+        if os.path.isdir(conf['__file__']):
+            read_conf_dir(cp, conf['__file__'])
+        else:
+            cp.read(conf['__file__'])
+
+        try:
+            pipe = cp.get("pipeline:main", "pipeline")
+        except (NoSectionError, NoOptionError):
+            return
+
+        proxy_name = pipe.rsplit(None, 1)[-1]
+        proxy_section = "app:" + proxy_name
+        for setting in ('rate_limit_after_segment',
+                        'rate_limit_segments_per_sec',
+                        'max_get_time'):
+            try:
+                conf[setting] = cp.get(proxy_section, setting)
+            except (NoSectionError, NoOptionError):
+                pass
+
+    def __call__(self, env, start_response):
+        """
+        WSGI entry point
+        """
+        req = Request(env)
+        try:
+            vrs, account, container, obj = req.split_path(4, 4, True)
+        except ValueError:
+            return self.app(env, start_response)
+
+        # install our COPY-callback hook
+        env['swift.copy_response_hook'] = self.copy_response_hook(
+            env.get('swift.copy_response_hook', lambda req, resp: resp))
+
+        if ((req.method == 'GET' or req.method == 'HEAD') and
+                req.params.get('multipart-manifest') != 'get'):
+            return GetContext(self, self.logger).\
+                handle_request(req, start_response)
+        elif req.method == 'PUT':
+            error_response = self.validate_x_object_manifest_header(
+                req, start_response)
+            if error_response:
+                return error_response(env, start_response)
+        return self.app(env, start_response)
+
+    def validate_x_object_manifest_header(self, req, start_response):
+        """
+        Make sure that X-Object-Manifest is valid if present.
+        """
+        if 'X-Object-Manifest' in req.headers:
+            value = req.headers['X-Object-Manifest']
+            container = prefix = None
+            try:
+                container, prefix = value.split('/', 1)
+            except ValueError:
+                pass
+            if not container or not prefix or '?' in value or '&' in value or \
+                    prefix[0] == '/':
+                return HTTPBadRequest(
+                    request=req,
+                    body=('X-Object-Manifest must be in the '
+                          'format container/prefix'))
+
+    def copy_response_hook(self, inner_hook):
+
+        def dlo_copy_hook(req, resp):
+            x_o_m = resp.headers.get('X-Object-Manifest')
+            if (x_o_m and req.params.get('multipart-manifest') != 'get'):
+                resp = GetContext(self, self.logger).get_or_head_response(
+                    req, x_o_m, resp.headers.items())
+            return inner_hook(req, resp)
+
+        return dlo_copy_hook
+
+
+def filter_factory(global_conf, **local_conf):
+    conf = global_conf.copy()
+    conf.update(local_conf)
+
+    def dlo_filter(app):
+        return DynamicLargeObject(app, conf)
+    return dlo_filter

swift/common/middleware/slo.py

@@ -134,22 +134,19 @@ the manifest and the segments it's referring to) in the container and account
 metadata which can be used for stats purposes.
 """
 
-from contextlib import contextmanager
-from time import time
-from urllib import quote
 from cStringIO import StringIO
 from datetime import datetime
-from sys import exc_info
 import mimetypes
 from hashlib import md5
-from swift.common.exceptions import ListingIterError, SegmentError
+from swift.common.exceptions import ListingIterError
 from swift.common.swob import Request, HTTPBadRequest, HTTPServerError, \
     HTTPMethodNotAllowed, HTTPRequestEntityTooLarge, HTTPLengthRequired, \
     HTTPOk, HTTPPreconditionFailed, HTTPException, HTTPNotFound, \
     HTTPUnauthorized, HTTPRequestedRangeNotSatisfiable, Response
 from swift.common.utils import json, get_logger, config_true_value, \
     get_valid_utf8_str, override_bytes_from_content_type, split_path, \
-    register_swift_info, RateLimitedIterator
+    register_swift_info, RateLimitedIterator, SegmentedIterable, \
+    closing_if_possible, close_if_possible, quote
 from swift.common.constraints import check_utf8, MAX_BUFFERED_SLO_SEGMENTS
 from swift.common.http import HTTP_NOT_FOUND, HTTP_UNAUTHORIZED, is_success
 from swift.common.wsgi import WSGIContext
@@ -204,139 +201,6 @@ class SloPutContext(WSGIContext):
         return app_resp
 
 
-def close_if_possible(maybe_closable):
-    close_method = getattr(maybe_closable, 'close', None)
-    if callable(close_method):
-        return close_method()
-
-
-@contextmanager
-def closing_if_possible(maybe_closable):
-    """
-    Like contextlib.closing(), but doesn't crash if the object lacks a close()
-    method.
-
-    PEP 333 (WSGI) says: "If the iterable returned by the application has a
-    close() method, the server or gateway must call that method upon
-    completion of the current request[.]" This function makes that easier.
-    """
-    yield maybe_closable
-    close_if_possible(maybe_closable)
-
-
-class SloIterable(object):
-    """
-    Iterable that returns the object contents for a large object.
-
-    :param req: original request object
-    :param app: WSGI application from which segments will come
-    :param listing_iter: iterable yielding the object segments to fetch,
-                         along with the byte subranges to fetch, in the
-                         form of a tuple (object-path, first-byte, last-byte)
-                         or (object-path, None, None) to fetch the whole thing.
-    :param max_get_time: maximum permitted duration of a GET request (seconds)
-    :param logger: logger object
-    :param ua_suffix: string to append to user-agent.
-    :param name: name of manifest (used in logging only)
-    """
-    def __init__(self, req, app, listing_iter, max_get_time,
-                 logger, ua_suffix, name='<not specified>'):
-        self.req = req
-        self.app = app
-        self.listing_iter = listing_iter
-        self.max_get_time = max_get_time
-        self.logger = logger
-        self.ua_suffix = ua_suffix
-        self.name = name
-
-    def app_iter_range(self, *a, **kw):
-        """
-        swob.Response will only respond with a 206 status in certain cases; one
-        of those is if the body iterator responds to .app_iter_range().
-
-        However, this object (or really, its listing iter) is smart enough to
-        handle the range stuff internally, so we just no-op this out to fool
-        swob.Response.
-
-        """
-        return self
-
-    def __iter__(self):
-        start_time = time()
-        have_yielded_data = False
-        try:
-            for seg_path, seg_etag, seg_size, first_byte, last_byte \
-                    in self.listing_iter:
-                if time() - start_time > self.max_get_time:
-                    raise SegmentError(
-                        'ERROR: While processing manifest %s, '
-                        'max LO GET time of %ds exceeded' %
-                        (self.name, self.max_get_time))
-                seg_req = self.req.copy_get()
-                seg_req.range = None
-                seg_req.environ['PATH_INFO'] = seg_path
-                seg_req.user_agent = "%s %s" % (seg_req.user_agent,
-                                                self.ua_suffix)
-                if first_byte is not None or last_byte is not None:
-                    seg_req.headers['Range'] = "bytes=%s-%s" % (
-                        # The 0 is to avoid having a range like "bytes=-10",
-                        # which actually means the *last* 10 bytes.
-                        '0' if first_byte is None else first_byte,
-                        '' if last_byte is None else last_byte)
-
-                seg_resp = seg_req.get_response(self.app)
-                if not is_success(seg_resp.status_int):
-                    close_if_possible(seg_resp.app_iter)
-                    raise SegmentError(
-                        'ERROR: While processing manifest %s, '
-                        'got %d while retrieving %s' %
-                        (self.name, seg_resp.status_int, seg_path))
-
-                elif ((seg_resp.etag != seg_etag) or
-                        (seg_resp.content_length != seg_size and
-                         not seg_req.range)):
-                    # The content-length check is for security reasons. Seems
-                    # possible that an attacker could upload a >1mb object and
-                    # then replace it with a much smaller object with same
-                    # etag.  Then create a big nested SLO that calls that
-                    # object many times which would hammer our obj servers. If
-                    # this is a range request, don't check content-length
-                    # because it won't match.
-                    close_if_possible(seg_resp.app_iter)
-                    raise SegmentError(
-                        'Object segment no longer valid: '
-                        '%(path)s etag: %(r_etag)s != %(s_etag)s or '
-                        '%(r_size)s != %(s_size)s.' %
-                        {'path': seg_req.path, 'r_etag': seg_resp.etag,
-                         'r_size': seg_resp.content_length,
-                         's_etag': seg_etag,
-                         's_size': seg_size})
-
-                with closing_if_possible(seg_resp.app_iter):
-                    for chunk in seg_resp.app_iter:
-                        yield chunk
-                        have_yielded_data = True
-        except ListingIterError as ex:
-            # I have to save this error because yielding the ' ' below clears
-            # the exception from the current stack frame.
-            err = exc_info()
-            self.logger.error('ERROR: While processing manifest %s, %s',
-                              self.name, ex)
-            # Normally, exceptions before any data has been yielded will
-            # cause Eventlet to send a 5xx response. In this particular
-            # case of ListingIterError we don't want that and we'd rather
-            # just send the normal 2xx response and then hang up early
-            # since 5xx codes are often used to judge Service Level
-            # Agreements and this ListingIterError indicates the user has
-            # created an invalid condition.
-            if not have_yielded_data:
-                yield ' '
-            raise err
-        except SegmentError:
-            self.logger.exception("Error getting segment")
-            raise
-
-
 class SloGetContext(WSGIContext):
 
     max_slo_recursion_depth = 10
@@ -383,8 +247,8 @@ class SloGetContext(WSGIContext):
         # submanifest referencing 50 MiB total, but first_byte falls in the
         # 51st MiB, then we can avoid fetching the first submanifest.
         #
-        # If we were to let SloIterable handle all the range calculations, we
-        # would be unable to make this optimization.
+        # If we were to make SegmentedIterable handle all the range
+        # calculations, we would be unable to make this optimization.
         total_length = sum(int(seg['bytes']) for seg in segments)
         if first_byte is None:
             first_byte = 0
@@ -545,8 +409,8 @@ class SloGetContext(WSGIContext):
             limit_after=self.slo.rate_limit_after_segment)
 
         # self._segment_listing_iterator gives us 3-tuples of (segment dict,
-        # start byte, end byte), but SloIterable wants (obj path, etag, size,
-        # start byte, end byte), so we clean that up here
+        # start byte, end byte), but SegmentedIterable wants (obj path, etag,
+        # size, start byte, end byte), so we clean that up here
         segment_listing_iter = (
             ("/{ver}/{acc}/{conobj}".format(
                 ver=ver, acc=account, conobj=seg_dict['name'].lstrip('/')),
@@ -557,7 +421,7 @@ class SloGetContext(WSGIContext):
         response = Response(request=req, content_length=content_length,
                             headers=response_headers,
                             conditional_response=True,
-                            app_iter=SloIterable(
+                            app_iter=SegmentedIterable(
                                 req, self.slo.app, segment_listing_iter,
                                 name=req.path, logger=self.slo.logger,
                                 ua_suffix="SLO MultipartGET",

swift/common/utils.py

@@ -61,8 +61,10 @@ utf8_decoder = codecs.getdecoder('utf-8')
 utf8_encoder = codecs.getencoder('utf-8')
 
 from swift import gettext_ as _
-from swift.common.exceptions import LockTimeout, MessageTimeout
-from swift.common.http import is_success, is_redirection, HTTP_NOT_FOUND
+from swift.common.exceptions import LockTimeout, MessageTimeout, \
+    ListingIterError, SegmentError
+from swift.common.http import is_success, is_redirection, HTTP_NOT_FOUND, \
+    HTTP_SERVICE_UNAVAILABLE
 
 # logging doesn't import patched as cleanly as one would like
 from logging.handlers import SysLogHandler
@@ -2557,3 +2559,141 @@ def quote(value, safe='/'):
     Patched version of urllib.quote that encodes utf-8 strings before quoting
     """
     return _quote(get_valid_utf8_str(value), safe)
+
+
+def close_if_possible(maybe_closable):
+    close_method = getattr(maybe_closable, 'close', None)
+    if callable(close_method):
+        return close_method()
+
+
+@contextmanager
+def closing_if_possible(maybe_closable):
+    """
+    Like contextlib.closing(), but doesn't crash if the object lacks a close()
+    method.
+
+    PEP 333 (WSGI) says: "If the iterable returned by the application has a
+    close() method, the server or gateway must call that method upon
+    completion of the current request[.]" This function makes that easier.
+    """
+    yield maybe_closable
+    close_if_possible(maybe_closable)
+
+
+class SegmentedIterable(object):
+    """
+    Iterable that returns the object contents for a large object.
+
+    :param req: original request object
+    :param app: WSGI application from which segments will come
+    :param listing_iter: iterable yielding the object segments to fetch,
+                         along with the byte subranges to fetch, in the
+                         form of a tuple (object-path, first-byte, last-byte)
+                         or (object-path, None, None) to fetch the whole thing.
+    :param max_get_time: maximum permitted duration of a GET request (seconds)
+    :param logger: logger object
+    :param ua_suffix: string to append to user-agent.
+    :param name: name of manifest (used in logging only)
+    :param response: optional response object for the response being sent
+                     to the client. Only affects logs.
+    """
+    def __init__(self, req, app, listing_iter, max_get_time,
+                 logger, ua_suffix, name='<not specified>', response=None):
+        self.req = req
+        self.app = app
+        self.listing_iter = listing_iter
+        self.max_get_time = max_get_time
+        self.logger = logger
+        self.ua_suffix = " " + ua_suffix
+        self.name = name
+        self.response = response
+
+    def app_iter_range(self, *a, **kw):
+        """
+        swob.Response will only respond with a 206 status in certain cases; one
+        of those is if the body iterator responds to .app_iter_range().
+
+        However, this object (or really, its listing iter) is smart enough to
+        handle the range stuff internally, so we just no-op this out for swob.
+        """
+        return self
+
+    def __iter__(self):
+        start_time = time.time()
+        have_yielded_data = False
+        try:
+            for seg_path, seg_etag, seg_size, first_byte, last_byte \
+                    in self.listing_iter:
+                if time.time() - start_time > self.max_get_time:
+                    raise SegmentError(
+                        'ERROR: While processing manifest %s, '
+                        'max LO GET time of %ds exceeded' %
+                        (self.name, self.max_get_time))
+                seg_req = self.req.copy_get()
+                seg_req.range = None
+                seg_req.environ['PATH_INFO'] = seg_path
+                seg_req.user_agent = "%s %s" % (seg_req.user_agent,
+                                                self.ua_suffix)
+                if first_byte is not None or last_byte is not None:
+                    seg_req.headers['Range'] = "bytes=%s-%s" % (
+                        # The 0 is to avoid having a range like "bytes=-10",
+                        # which actually means the *last* 10 bytes.
+                        '0' if first_byte is None else first_byte,
+                        '' if last_byte is None else last_byte)
+
+                seg_resp = seg_req.get_response(self.app)
+                if not is_success(seg_resp.status_int):
+                    close_if_possible(seg_resp.app_iter)
+                    raise SegmentError(
+                        'ERROR: While processing manifest %s, '
+                        'got %d while retrieving %s' %
+                        (self.name, seg_resp.status_int, seg_path))
+
+                elif ((seg_etag and (seg_resp.etag != seg_etag)) or
+                        (seg_size and (seg_resp.content_length != seg_size) and
+                         not seg_req.range)):
+                    # The content-length check is for security reasons. Seems
+                    # possible that an attacker could upload a >1mb object and
+                    # then replace it with a much smaller object with same
+                    # etag. Then create a big nested SLO that calls that
+                    # object many times which would hammer our obj servers. If
+                    # this is a range request, don't check content-length
+                    # because it won't match.
+                    close_if_possible(seg_resp.app_iter)
+                    raise SegmentError(
+                        'Object segment no longer valid: '
+                        '%(path)s etag: %(r_etag)s != %(s_etag)s or '
+                        '%(r_size)s != %(s_size)s.' %
+                        {'path': seg_req.path, 'r_etag': seg_resp.etag,
+                         'r_size': seg_resp.content_length,
+                         's_etag': seg_etag,
+                         's_size': seg_size})
+
+                for chunk in seg_resp.app_iter:
+                    yield chunk
+                    have_yielded_data = True
+                close_if_possible(seg_resp.app_iter)
+        except ListingIterError as err:
+            # I have to save this error because yielding the ' ' below clears
+            # the exception from the current stack frame.
+            excinfo = sys.exc_info()
+            self.logger.exception('ERROR: While processing manifest %s, %s',
+                                  self.name, err)
+            # Normally, exceptions before any data has been yielded will
+            # cause Eventlet to send a 5xx response. In this particular
+            # case of ListingIterError we don't want that and we'd rather
+            # just send the normal 2xx response and then hang up early
+            # since 5xx codes are often used to judge Service Level
+            # Agreements and this ListingIterError indicates the user has
+            # created an invalid condition.
+            if not have_yielded_data:
+                yield ' '
+            raise excinfo
+        except SegmentError as err:
+            self.logger.exception(err)
+            # This doesn't actually change the response status (we're too
+            # late for that), but this does make it to the logs.
+            if self.response:
+                self.response.status = HTTP_SERVICE_UNAVAILABLE
+            raise

swift/proxy/controllers/obj.py

@@ -26,16 +26,12 @@
 
 import itertools
 import mimetypes
-import re
 import time
 import math
-from datetime import datetime
 from swift import gettext_ as _
 from urllib import unquote, quote
-from hashlib import md5
-from sys import exc_info
 
-from eventlet import sleep, GreenPile
+from eventlet import GreenPile
 from eventlet.queue import Queue
 from eventlet.timeout import Timeout
 
@@ -44,32 +40,23 @@ from swift.common.utils import ContextPool, normalize_timestamp, \
     quorum_size, GreenAsyncPile, normalize_delete_at_timestamp
 from swift.common.bufferedhttp import http_connect
 from swift.common.constraints import check_metadata, check_object_creation, \
-    CONTAINER_LISTING_LIMIT, MAX_FILE_SIZE, check_copy_from_header
+    MAX_FILE_SIZE, check_copy_from_header
 from swift.common.exceptions import ChunkReadTimeout, \
     ChunkWriteTimeout, ConnectionTimeout, ListingIterNotFound, \
-    ListingIterNotAuthorized, ListingIterError, SegmentError
+    ListingIterNotAuthorized, ListingIterError
 from swift.common.http import is_success, is_client_error, HTTP_CONTINUE, \
-    HTTP_CREATED, HTTP_MULTIPLE_CHOICES, HTTP_NOT_FOUND, HTTP_CONFLICT, \
+    HTTP_CREATED, HTTP_MULTIPLE_CHOICES, HTTP_NOT_FOUND, \
     HTTP_INTERNAL_SERVER_ERROR, HTTP_SERVICE_UNAVAILABLE, \
     HTTP_INSUFFICIENT_STORAGE
 from swift.proxy.controllers.base import Controller, delay_denial, \
     cors_validation
 from swift.common.swob import HTTPAccepted, HTTPBadRequest, HTTPNotFound, \
     HTTPPreconditionFailed, HTTPRequestEntityTooLarge, HTTPRequestTimeout, \
-    HTTPServerError, HTTPServiceUnavailable, Request, Response, \
+    HTTPServerError, HTTPServiceUnavailable, Request, \
     HTTPClientDisconnect, HTTPNotImplemented
 from swift.common.request_helpers import is_user_meta
 
 
-def segment_listing_iter(listing):
-    listing = iter(listing)
-    while True:
-        seg_dict = listing.next()
-        if isinstance(seg_dict['name'], unicode):
-            seg_dict['name'] = seg_dict['name'].encode('utf-8')
-        yield seg_dict
-
-
 def copy_headers_into(from_r, to_r):
     """
     Will copy desired headers from from_r to to_r
@@ -92,238 +79,6 @@ def check_content_type(req):
     return None
 
 
-class SegmentedIterable(object):
-    """
-    Iterable that returns the object contents for a segmented object in Swift.
-
-    If there's a failure that cuts the transfer short, the response's
-    `status_int` will be updated (again, just for logging since the original
-    status would have already been sent to the client).
-
-    :param controller: The ObjectController instance to work with.
-    :param container: The container the object segments are within. If
-                      container is None will derive container from elements
-                      in listing using split('/', 1).
-    :param listing: The listing of object segments to iterate over; this may
-                    be an iterator or list that returns dicts with 'name' and
-                    'bytes' keys.
-    :param response: The swob.Response this iterable is associated with, if
-                     any (default: None)
-    :param max_lo_time: Defaults to 86400. The connection for the
-                        SegmentedIterable will drop after that many seconds.
-    """
-
-    def __init__(self, controller, container, listing, response=None,
-                 max_lo_time=86400):
-        self.controller = controller
-        self.container = container
-        self.listing = segment_listing_iter(listing)
-        self.max_lo_time = max_lo_time
-        self.ratelimit_index = 0
-        self.segment_dict = None
-        self.segment_peek = None
-        self.seek = 0
-        self.length = None
-        self.segment_iter = None
-        # See NOTE: swift_conn at top of file about this.
-        self.segment_iter_swift_conn = None
-        self.position = 0
-        self.have_yielded_data = False
-        self.response = response
-        if not self.response:
-            self.response = Response()
-        self.next_get_time = 0
-        self.start_time = time.time()
-
-    def _load_next_segment(self):
-        """
-        Loads the self.segment_iter with the next object segment's contents.
-
-        :raises: StopIteration when there are no more object segments
-        """
-        try:
-            self.ratelimit_index += 1
-            if time.time() - self.start_time > self.max_lo_time:
-                raise SegmentError(
-                    _('Max LO GET time of %s exceeded.') % self.max_lo_time)
-            self.segment_dict = self.segment_peek or self.listing.next()
-            self.segment_peek = None
-            if self.container is None:
-                container, obj = \
-                    self.segment_dict['name'].lstrip('/').split('/', 1)
-            else:
-                container, obj = self.container, self.segment_dict['name']
-            partition = self.controller.app.object_ring.get_part(
-                self.controller.account_name, container, obj)
-            path = '/%s/%s/%s' % (self.controller.account_name,
-                                  container, obj)
-            req = Request.blank('/v1' + path)
-            if self.seek or (self.length and self.length > 0):
-                bytes_available = \
-                    self.segment_dict['bytes'] - self.seek
-                range_tail = ''
-                if self.length:
-                    if bytes_available >= self.length:
-                        range_tail = self.seek + self.length - 1
-                        self.length = 0
-                    else:
-                        self.length -= bytes_available
-                if self.seek or range_tail:
-                    req.range = 'bytes=%s-%s' % (self.seek, range_tail)
-                self.seek = 0
-            if self.ratelimit_index > \
-                    self.controller.app.rate_limit_after_segment:
-                sleep(max(self.next_get_time - time.time(), 0))
-            self.next_get_time = time.time() + \
-                1.0 / self.controller.app.rate_limit_segments_per_sec
-            resp = self.controller.GETorHEAD_base(
-                req, _('Object'), self.controller.app.object_ring, partition,
-                path)
-            if not is_success(resp.status_int):
-                raise Exception(_(
-                    'Could not load object segment %(path)s:'
-                    ' %(status)s') % {'path': path, 'status': resp.status_int})
-            self.segment_iter = resp.app_iter
-            # See NOTE: swift_conn at top of file about this.
-            self.segment_iter_swift_conn = getattr(resp, 'swift_conn', None)
-        except StopIteration:
-            raise
-        except SegmentError as err:
-            if not getattr(err, 'swift_logged', False):
-                self.controller.app.logger.error(_(
-                    'ERROR: While processing manifest '
-                    '/%(acc)s/%(cont)s/%(obj)s, %(err)s'),
-                    {'acc': self.controller.account_name,
-                     'cont': self.controller.container_name,
-                     'obj': self.controller.object_name, 'err': err})
-                err.swift_logged = True
-                self.response.status_int = HTTP_CONFLICT
-            raise
-        except (Exception, Timeout) as err:
-            if not getattr(err, 'swift_logged', False):
-                self.controller.app.logger.exception(_(
-                    'ERROR: While processing manifest '
-                    '/%(acc)s/%(cont)s/%(obj)s'),
-                    {'acc': self.controller.account_name,
-                     'cont': self.controller.container_name,
-                     'obj': self.controller.object_name})
-                err.swift_logged = True
-                self.response.status_int = HTTP_SERVICE_UNAVAILABLE
-            raise
-
-    def next(self):
-        return iter(self).next()
-
-    def __iter__(self):
-        """Standard iterator function that returns the object's contents."""
-        try:
-            while True:
-                if not self.segment_iter:
-                    self._load_next_segment()
-                while True:
-                    with ChunkReadTimeout(self.controller.app.node_timeout):
-                        try:
-                            chunk = self.segment_iter.next()
-                            break
-                        except StopIteration:
-                            if self.length is None or self.length > 0:
-                                self._load_next_segment()
-                            else:
-                                return
-                self.position += len(chunk)
-                self.have_yielded_data = True
-                yield chunk
-        except StopIteration:
-            raise
-        except SegmentError:
-            # I have to save this error because yielding the ' ' below clears
-            # the exception from the current stack frame.
-            err = exc_info()
-            if not self.have_yielded_data:
-                # Normally, exceptions before any data has been yielded will
-                # cause Eventlet to send a 5xx response. In this particular
-                # case of SegmentError we don't want that and we'd rather
-                # just send the normal 2xx response and then hang up early
-                # since 5xx codes are often used to judge Service Level
-                # Agreements and this SegmentError indicates the user has
-                # created an invalid condition.
-                yield ' '
-            raise err
-        except (Exception, Timeout) as err:
-            if not getattr(err, 'swift_logged', False):
-                self.controller.app.logger.exception(_(
-                    'ERROR: While processing manifest '
-                    '/%(acc)s/%(cont)s/%(obj)s'),
-                    {'acc': self.controller.account_name,
-                     'cont': self.controller.container_name,
-                     'obj': self.controller.object_name})
-                err.swift_logged = True
-                self.response.status_int = HTTP_SERVICE_UNAVAILABLE
-            raise
-
-    def app_iter_range(self, start, stop):
-        """
-        Non-standard iterator function for use with Swob in serving Range
-        requests more quickly. This will skip over segments and do a range
-        request on the first segment to return data from, if needed.
-
-        :param start: The first byte (zero-based) to return. None for 0.
-        :param stop: The last byte (zero-based) to return. None for end.
-        """
-        try:
-            if start:
-                self.segment_peek = self.listing.next()
-                while start >= self.position + self.segment_peek['bytes']:
-                    self.position += self.segment_peek['bytes']
-                    self.segment_peek = self.listing.next()
-                self.seek = start - self.position
-            else:
-                start = 0
-            if stop is not None:
-                length = stop - start
-                self.length = length
-            else:
-                length = None
-            for chunk in self:
-                if length is not None:
-                    length -= len(chunk)
-                    if length <= 0:
-                        if length < 0:
-                            # Chop off the extra:
-                            yield chunk[:length]
-                        else:
-                            yield chunk
-                        break
-                yield chunk
-            # See NOTE: swift_conn at top of file about this.
-            if self.segment_iter_swift_conn:
-                try:
-                    self.segment_iter_swift_conn.close()
-                except Exception:
-                    pass
-                self.segment_iter_swift_conn = None
-            if self.segment_iter:
-                try:
-                    while self.segment_iter.next():
-                        pass
-                except Exception:
-                    pass
-                self.segment_iter = None
-        except StopIteration:
-            raise
-        except (Exception, Timeout) as err:
-            if not getattr(err, 'swift_logged', False):
-                self.controller.app.logger.exception(_(
-                    'ERROR: While processing manifest '
-                    '/%(acc)s/%(cont)s/%(obj)s'),
-                    {'acc': self.controller.account_name,
-                     'cont': self.controller.container_name,
-                     'obj': self.controller.object_name})
-                err.swift_logged = True
-                self.response.status_int = HTTP_SERVICE_UNAVAILABLE
-            raise
-
-
 class ObjectController(Controller):
     """WSGI controller for object requests."""
     server_type = 'Object'
@@ -455,72 +210,6 @@ class ObjectController(Controller):
                 resp.headers['content-type'].rsplit(';', 1)
             if check_extra_meta.lstrip().startswith('swift_bytes='):
                 resp.content_type = content_type
-
-        large_object = None
-        if 'x-object-manifest' in resp.headers and \
-                req.params.get('multipart-manifest') != 'get':
-            large_object = 'DLO'
-            lcontainer, lprefix = \
-                resp.headers['x-object-manifest'].split('/', 1)
-            lcontainer = unquote(lcontainer)
-            lprefix = unquote(lprefix)
-            try:
-                pages_iter = iter(self._listing_pages_iter(lcontainer, lprefix,
-                                                           req.environ))
-                listing_page1 = pages_iter.next()
-                listing = itertools.chain(listing_page1,
-                                          self._remaining_items(pages_iter))
-            except ListingIterNotFound:
-                return HTTPNotFound(request=req)
-            except ListingIterNotAuthorized as err:
-                return err.aresp
-            except ListingIterError:
-                return HTTPServerError(request=req)
-            except StopIteration:
-                listing_page1 = listing = ()
-
-        if large_object:
-            if len(listing_page1) >= CONTAINER_LISTING_LIMIT:
-                resp = Response(headers=resp.headers, request=req,
-                                conditional_response=True)
-                resp.app_iter = SegmentedIterable(
-                    self, lcontainer, listing, resp,
-                    max_lo_time=self.app.max_large_object_get_time)
-            else:
-                # For objects with a reasonable number of segments, we'll serve
-                # them with a set content-length and computed etag.
-                listing = list(listing)
-                if listing:
-                    try:
-                        content_length = sum(o['bytes'] for o in listing)
-                        last_modified = \
-                            max(o['last_modified'] for o in listing)
-                        last_modified = datetime(*map(int, re.split('[^\d]',
-                                                 last_modified)[:-1]))
-                        etag = md5(
-                            ''.join(o['hash'] for o in listing)).hexdigest()
-                    except KeyError:
-                        return HTTPServerError('Invalid Manifest File',
-                                               request=req)
-
-                else:
-                    content_length = 0
-                    last_modified = resp.last_modified
-                    etag = md5().hexdigest()
-                resp = Response(headers=resp.headers, request=req,
-                                conditional_response=True)
-                resp.app_iter = SegmentedIterable(
-                    self, lcontainer, listing, resp,
-                    max_lo_time=self.app.max_large_object_get_time)
-                resp.content_length = content_length
-                resp.last_modified = last_modified
-                resp.etag = etag
-            resp.headers['accept-ranges'] = 'bytes'
-            # In case of a manifest file of nonzero length, the
-            # backend may have sent back a Content-Range header for
-            # the manifest. It's wrong for the client, though.
-            resp.content_range = None
-
         return resp
 
     @public
@@ -826,9 +515,7 @@ class ObjectController(Controller):
         if error_response:
             return error_response
         if object_versions and not req.environ.get('swift_versioned_copy'):
-            is_manifest = 'x-object-manifest' in req.headers or \
-                          'x-object-manifest' in hresp.headers
-            if hresp.status_int != HTTP_NOT_FOUND and not is_manifest:
+            if hresp.status_int != HTTP_NOT_FOUND:
                 # This is a version manifest and needs to be handled
                 # differently. First copy the existing data to a new object,
                 # then write the data from this request to the version manifest

swift/proxy/server.py

@@ -44,24 +44,22 @@ from swift.common.swob import HTTPBadRequest, HTTPForbidden, \
 #
 # "name" (required) is the entry point name from setup.py.
 #
-# "after" (optional) is a list of middlewares that this middleware should come
-# after. Default is for the middleware to go at the start of the pipeline. Any
-# middlewares in the "after" list that are not present in the pipeline will be
-# ignored, so you can safely name optional middlewares to come after. For
-# example, 'after: ["catch_errors", "bulk"]' would install this middleware
-# after catch_errors and bulk if both were present, but if bulk were absent,
-# would just install it after catch_errors.
-#
 # "after_fn" (optional) a function that takes a PipelineWrapper object as its
-# single argument and returns a list of middlewares that this middleware should
-# come after. This list overrides any defined by the "after" field.
+# single argument and returns a list of middlewares that this middleware
+# should come after. Any middlewares in the returned list that are not present
+# in the pipeline will be ignored, so you can safely name optional middlewares
+# to come after. For example, ["catch_errors", "bulk"] would install this
+# middleware after catch_errors and bulk if both were present, but if bulk
+# were absent, would just install it after catch_errors.
+
 required_filters = [
     {'name': 'catch_errors'},
     {'name': 'gatekeeper',
      'after_fn': lambda pipe: (['catch_errors']
                                if pipe.startswith("catch_errors")
-                               else [])}
-]
+                               else [])},
+    {'name': 'dlo', 'after_fn': lambda _junk: ['catch_errors', 'gatekeeper',
+                                               'proxy_logging']}]
 
 
 class Application(object):
@@ -528,10 +526,7 @@ class Application(object):
         for filter_spec in reversed(required_filters):
             filter_name = filter_spec['name']
             if filter_name not in pipe:
-                if 'after_fn' in filter_spec:
-                    afters = filter_spec['after_fn'](pipe)
-                else:
-                    afters = filter_spec.get('after', [])
+                afters = filter_spec.get('after_fn', lambda _junk: [])(pipe)
                 insert_at = 0
                 for after in afters:
                     try:

test/unit/common/middleware/helpers.py

@@ -0,0 +1,98 @@
+# Copyright (c) 2013 OpenStack Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This stuff can't live in test/unit/__init__.py due to its swob dependency.
+
+from copy import deepcopy
+from hashlib import md5
+from swift.common import swob
+from swift.common.utils import split_path
+
+
+class FakeSwift(object):
+    """
+    A good-enough fake Swift proxy server to use in testing middleware.
+    """
+
+    def __init__(self):
+        self._calls = []
+        self.req_method_paths = []
+        self.uploaded = {}
+        # mapping of (method, path) --> (response class, headers, body)
+        self._responses = {}
+
+    def __call__(self, env, start_response):
+        method = env['REQUEST_METHOD']
+        path = env['PATH_INFO']
+        _, acc, cont, obj = split_path(env['PATH_INFO'], 0, 4,
+                                       rest_with_last=True)
+        if env.get('QUERY_STRING'):
+            path += '?' + env['QUERY_STRING']
+
+        headers = swob.Request(env).headers
+        self._calls.append((method, path, headers))
+
+        try:
+            resp_class, raw_headers, body = self._responses[(method, path)]
+            headers = swob.HeaderKeyDict(raw_headers)
+        except KeyError:
+            if (env.get('QUERY_STRING')
+                    and (method, env['PATH_INFO']) in self._responses):
+                resp_class, raw_headers, body = self._responses[
+                    (method, env['PATH_INFO'])]
+                headers = swob.HeaderKeyDict(raw_headers)
+            elif method == 'HEAD' and ('GET', path) in self._responses:
+                resp_class, raw_headers, _ = self._responses[('GET', path)]
+                body = None
+                headers = swob.HeaderKeyDict(raw_headers)
+            elif method == 'GET' and obj and path in self.uploaded:
+                resp_class = swob.HTTPOk
+                headers, body = self.uploaded[path]
+            else:
+                print "Didn't find %r in allowed responses" % ((method, path),)
+                raise
+
+        # simulate object PUT
+        if method == 'PUT' and obj:
+            input = env['wsgi.input'].read()
+            etag = md5(input).hexdigest()
+            headers.setdefault('Etag', etag)
+            headers.setdefault('Content-Length', len(input))
+
+            # keep it for subsequent GET requests later
+            self.uploaded[path] = (deepcopy(headers), input)
+            if "CONTENT_TYPE" in env:
+                self.uploaded[path][0]['Content-Type'] = env["CONTENT_TYPE"]
+
+        # range requests ought to work, hence conditional_response=True
+        req = swob.Request(env)
+        resp = resp_class(req=req, headers=headers, body=body,
+                          conditional_response=True)
+        return resp(env, start_response)
+
+    @property
+    def calls(self):
+        return [(method, path) for method, path, headers in self._calls]
+
+    @property
+    def calls_with_headers(self):
+        return self._calls
+
+    @property
+    def call_count(self):
+        return len(self._calls)
+
+    def register(self, method, path, response_class, headers, body):
+        self._responses[(method, path)] = (response_class, headers, body)

test/unit/common/middleware/test_dlo.py

@@ -0,0 +1,792 @@
+#-*- coding:utf-8 -*-
+# Copyright (c) 2013 OpenStack Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import contextlib
+import hashlib
+import json
+import mock
+import tempfile
+import time
+import unittest
+from swift.common import exceptions, swob
+from swift.common.middleware import dlo
+from test.unit.common.middleware.helpers import FakeSwift
+from textwrap import dedent
+
+LIMIT = 'swift.common.middleware.dlo.CONTAINER_LISTING_LIMIT'
+
+
+class DloTestCase(unittest.TestCase):
+    def call_dlo(self, req, app=None, expect_exception=False):
+        if app is None:
+            app = self.dlo
+
+        req.headers.setdefault("User-Agent", "Soap Opera")
+
+        status = [None]
+        headers = [None]
+
+        def start_response(s, h, ei=None):
+            status[0] = s
+            headers[0] = h
+
+        body_iter = app(req.environ, start_response)
+        body = ''
+        caught_exc = None
+        try:
+            for chunk in body_iter:
+                body += chunk
+        except Exception as exc:
+            if expect_exception:
+                caught_exc = exc
+            else:
+                raise
+
+        if expect_exception:
+            return status[0], headers[0], body, caught_exc
+        else:
+            return status[0], headers[0], body
+
+    def setUp(self):
+        self.app = FakeSwift()
+        self.dlo = dlo.filter_factory({
+            # don't slow down tests with rate limiting
+            'rate_limit_after_segment': '1000000',
+        })(self.app)
+
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/seg_01',
+            swob.HTTPOk, {'Content-Length': '5', 'Etag': 'seg01-etag'},
+            'aaaaa')
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/seg_02',
+            swob.HTTPOk, {'Content-Length': '5', 'Etag': 'seg02-etag'},
+            'bbbbb')
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/seg_03',
+            swob.HTTPOk, {'Content-Length': '5', 'Etag': 'seg03-etag'},
+            'ccccc')
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/seg_04',
+            swob.HTTPOk, {'Content-Length': '5', 'Etag': 'seg04-etag'},
+            'ddddd')
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/seg_05',
+            swob.HTTPOk, {'Content-Length': '5', 'Etag': 'seg05-etag'},
+            'eeeee')
+
+        # an unrelated object (not seg*) to test the prefix matching
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/catpicture.jpg',
+            swob.HTTPOk, {'Content-Length': '9', 'Etag': 'cats-etag'},
+            'meow meow meow meow')
+
+        self.app.register(
+            'GET', '/v1/AUTH_test/mancon/manifest',
+            swob.HTTPOk, {'Content-Length': '17', 'Etag': 'manifest-etag',
+                          'X-Object-Manifest': 'c/seg'},
+            'manifest-contents')
+
+        lm = '2013-11-22T02:42:13.781760'
+        ct = 'application/octet-stream'
+        segs = [{"hash": "seg01-etag", "bytes": 5, "name": "seg_01",
+                 "last_modified": lm, "content_type": ct},
+                {"hash": "seg02-etag", "bytes": 5, "name": "seg_02",
+                 "last_modified": lm, "content_type": ct},
+                {"hash": "seg03-etag", "bytes": 5, "name": "seg_03",
+                 "last_modified": lm, "content_type": ct},
+                {"hash": "seg04-etag", "bytes": 5, "name": "seg_04",
+                 "last_modified": lm, "content_type": ct},
+                {"hash": "seg05-etag", "bytes": 5, "name": "seg_05",
+                 "last_modified": lm, "content_type": ct}]
+
+        full_container_listing = segs + [{"hash": "cats-etag", "bytes": 9,
+                                          "name": "catpicture.jpg",
+                                          "last_modified": lm,
+                                          "content_type": "application/png"}]
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json',
+            swob.HTTPOk, {'Content-Type': 'application/json; charset=utf-8'},
+            json.dumps(full_container_listing))
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=seg',
+            swob.HTTPOk, {'Content-Type': 'application/json; charset=utf-8'},
+            json.dumps(segs))
+
+        # This is to let us test multi-page container listings; we use the
+        # trailing underscore to send small (pagesize=3) listings.
+        #
+        # If you're testing against this, be sure to mock out
+        # CONTAINER_LISTING_LIMIT to 3 in your test.
+        self.app.register(
+            'GET', '/v1/AUTH_test/mancon/manifest-many-segments',
+            swob.HTTPOk, {'Content-Length': '7', 'Etag': 'etag-manyseg',
+                          'X-Object-Manifest': 'c/seg_'},
+            'manyseg')
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=seg_',
+            swob.HTTPOk, {'Content-Type': 'application/json; charset=utf-8'},
+            json.dumps(segs[:3]))
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=seg_&marker=seg_03',
+            swob.HTTPOk, {'Content-Type': 'application/json; charset=utf-8'},
+            json.dumps(segs[3:]))
+
+        # Here's a manifest with 0 segments
+        self.app.register(
+            'GET', '/v1/AUTH_test/mancon/manifest-no-segments',
+            swob.HTTPOk, {'Content-Length': '7', 'Etag': 'noseg',
+                          'X-Object-Manifest': 'c/noseg_'},
+            'noseg')
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=noseg_',
+            swob.HTTPOk, {'Content-Type': 'application/json; charset=utf-8'},
+            json.dumps([]))
+
+
+class TestDloPutManifest(DloTestCase):
+    def setUp(self):
+        super(TestDloPutManifest, self).setUp()
+        self.app.register(
+            'PUT', '/v1/AUTH_test/c/m',
+            swob.HTTPCreated, {}, None)
+
+    def test_validating_x_object_manifest(self):
+        exp_okay = ["c/o",
+                    "c/obj/with/slashes",
+                    "c/obj/with/trailing/slash/",
+                    "c/obj/with//multiple///slashes////adjacent"]
+        exp_bad = ["",
+                   "/leading/slash",
+                   "double//slash",
+                   "container-only",
+                   "whole-container/",
+                   "c/o?short=querystring",
+                   "c/o?has=a&long-query=string"]
+
+        got_okay = []
+        got_bad = []
+        for val in (exp_okay + exp_bad):
+            req = swob.Request.blank("/v1/AUTH_test/c/m",
+                                     environ={'REQUEST_METHOD': 'PUT'},
+                                     headers={"X-Object-Manifest": val})
+            status, _, _ = self.call_dlo(req)
+            if status.startswith("201"):
+                got_okay.append(val)
+            else:
+                got_bad.append(val)
+
+        self.assertEqual(exp_okay, got_okay)
+        self.assertEqual(exp_bad, got_bad)
+
+    def test_validation_watches_manifests_with_slashes(self):
+        self.app.register(
+            'PUT', '/v1/AUTH_test/con/w/x/y/z',
+            swob.HTTPCreated, {}, None)
+
+        req = swob.Request.blank(
+            "/v1/AUTH_test/con/w/x/y/z", environ={'REQUEST_METHOD': 'PUT'},
+            headers={"X-Object-Manifest": 'good/value'})
+        status, _, _ = self.call_dlo(req)
+        self.assertEqual(status, "201 Created")
+
+        req = swob.Request.blank(
+            "/v1/AUTH_test/con/w/x/y/z", environ={'REQUEST_METHOD': 'PUT'},
+            headers={"X-Object-Manifest": '/badvalue'})
+        status, _, _ = self.call_dlo(req)
+        self.assertEqual(status, "400 Bad Request")
+
+    def test_validation_ignores_containers(self):
+        self.app.register(
+            'PUT', '/v1/a/c',
+            swob.HTTPAccepted, {}, None)
+        req = swob.Request.blank(
+            "/v1/a/c", environ={'REQUEST_METHOD': 'PUT'},
+            headers={"X-Object-Manifest": "/superbogus/?wrong=in&every=way"})
+        status, _, _ = self.call_dlo(req)
+        self.assertEqual(status, "202 Accepted")
+
+    def test_validation_ignores_accounts(self):
+        self.app.register(
+            'PUT', '/v1/a',
+            swob.HTTPAccepted, {}, None)
+        req = swob.Request.blank(
+            "/v1/a", environ={'REQUEST_METHOD': 'PUT'},
+            headers={"X-Object-Manifest": "/superbogus/?wrong=in&every=way"})
+        status, _, _ = self.call_dlo(req)
+        self.assertEqual(status, "202 Accepted")
+
+
+class TestDloHeadManifest(DloTestCase):
+    def test_head_large_object(self):
+        expected_etag = '"%s"' % hashlib.md5(
+            "seg01-etag" + "seg02-etag" + "seg03-etag" +
+            "seg04-etag" + "seg05-etag").hexdigest()
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'HEAD'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(headers["Etag"], expected_etag)
+        self.assertEqual(headers["Content-Length"], "25")
+
+    def test_head_large_object_too_many_segments(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-many-segments',
+                                 environ={'REQUEST_METHOD': 'HEAD'})
+        with mock.patch(LIMIT, 3):
+            status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+
+        # etag is manifest's etag
+        self.assertEqual(headers["Etag"], "etag-manyseg")
+        self.assertEqual(headers.get("Content-Length"), None)
+
+    def test_head_large_object_no_segments(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-no-segments',
+                                 environ={'REQUEST_METHOD': 'HEAD'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(headers["Etag"],
+                         '"' + hashlib.md5("").hexdigest() + '"')
+        self.assertEqual(headers["Content-Length"], "0")
+
+        # one request to HEAD the manifest
+        # one request for the first page of listings
+        # *zero* requests for the second page of listings
+        self.assertEqual(
+            self.app.calls,
+            [('HEAD', '/v1/AUTH_test/mancon/manifest-no-segments'),
+             ('GET', '/v1/AUTH_test/c?format=json&prefix=noseg_')])
+
+
+class TestDloGetManifest(DloTestCase):
+    def test_get_manifest(self):
+        expected_etag = '"%s"' % hashlib.md5(
+            "seg01-etag" + "seg02-etag" + "seg03-etag" +
+            "seg04-etag" + "seg05-etag").hexdigest()
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(headers["Etag"], expected_etag)
+        self.assertEqual(headers["Content-Length"], "25")
+        self.assertEqual(body, 'aaaaabbbbbcccccdddddeeeee')
+
+        for _, _, hdrs in self.app.calls_with_headers[1:]:
+            ua = hdrs.get("User-Agent", "")
+            self.assertTrue("DLO MultipartGET" in ua)
+            self.assertFalse("DLO MultipartGET DLO MultipartGET" in ua)
+        # the first request goes through unaltered
+        self.assertFalse(
+            "DLO MultipartGET" in self.app.calls_with_headers[0][2])
+
+    def test_get_non_manifest_passthrough(self):
+        req = swob.Request.blank('/v1/AUTH_test/c/catpicture.jpg',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        status, headers, body = self.call_dlo(req)
+        self.assertEqual(body, "meow meow meow meow")
+
+    def test_get_non_object_passthrough(self):
+        self.app.register('GET', '/info', swob.HTTPOk,
+                          {}, 'useful stuff here')
+        req = swob.Request.blank('/info',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        status, headers, body = self.call_dlo(req)
+        self.assertEqual(status, '200 OK')
+        self.assertEqual(body, 'useful stuff here')
+        self.assertEqual(self.app.call_count, 1)
+
+    def test_get_manifest_passthrough(self):
+        # reregister it with the query param
+        self.app.register(
+            'GET', '/v1/AUTH_test/mancon/manifest?multipart-manifest=get',
+            swob.HTTPOk, {'Content-Length': '17', 'Etag': 'manifest-etag',
+                          'X-Object-Manifest': 'c/seg'},
+            'manifest-contents')
+        req = swob.Request.blank(
+            '/v1/AUTH_test/mancon/manifest',
+            environ={'REQUEST_METHOD': 'GET',
+                     'QUERY_STRING': 'multipart-manifest=get'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(headers["Etag"], "manifest-etag")
+        self.assertEqual(body, "manifest-contents")
+
+    def test_error_passthrough(self):
+        self.app.register(
+            'GET', '/v1/AUTH_test/gone/404ed',
+            swob.HTTPNotFound, {}, None)
+        req = swob.Request.blank('/v1/AUTH_test/gone/404ed',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        status, headers, body = self.call_dlo(req)
+        self.assertEqual(status, '404 Not Found')
+
+    def test_get_range(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=8-17'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "206 Partial Content")
+        self.assertEqual(headers["Content-Length"], "10")
+        self.assertEqual(body, "bbcccccddd")
+        expected_etag = '"%s"' % hashlib.md5(
+            "seg01-etag" + "seg02-etag" + "seg03-etag" +
+            "seg04-etag" + "seg05-etag").hexdigest()
+        self.assertEqual(headers.get("Etag"), expected_etag)
+
+    def test_get_range_on_segment_boundaries(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=10-19'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "206 Partial Content")
+        self.assertEqual(headers["Content-Length"], "10")
+        self.assertEqual(body, "cccccddddd")
+
+    def test_get_range_first_byte(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=0-0'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "206 Partial Content")
+        self.assertEqual(headers["Content-Length"], "1")
+        self.assertEqual(body, "a")
+
+    def test_get_range_last_byte(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=24-24'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "206 Partial Content")
+        self.assertEqual(headers["Content-Length"], "1")
+        self.assertEqual(body, "e")
+
+    def test_get_range_overlapping_end(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=18-30'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "206 Partial Content")
+        self.assertEqual(headers["Content-Length"], "7")
+        self.assertEqual(headers["Content-Range"], "bytes 18-24/25")
+        self.assertEqual(body, "ddeeeee")
+
+    def test_get_range_unsatisfiable(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=25-30'})
+        status, headers, body = self.call_dlo(req)
+        self.assertEqual(status, "416 Requested Range Not Satisfiable")
+
+    def test_get_range_many_segments_satisfiable(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-many-segments',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=3-12'})
+        with mock.patch(LIMIT, 3):
+            status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "206 Partial Content")
+        self.assertEqual(headers["Content-Length"], "10")
+        # The /15 here indicates that this is a 15-byte object. DLO can't tell
+        # if there are more segments or not without fetching more container
+        # listings, though, so we just go with the sum of the lengths of the
+        # segments we can see. In an ideal world, this would be "bytes 3-12/*"
+        # to indicate that we don't know the full object length. However, RFC
+        # 2616 section 14.16 explicitly forbids us from doing that:
+        #
+        #   A response with status code 206 (Partial Content) MUST NOT include
+        #   a Content-Range field with a byte-range-resp-spec of "*".
+        #
+        # Since the truth is forbidden, we lie.
+        self.assertEqual(headers["Content-Range"], "bytes 3-12/15")
+        self.assertEqual(body, "aabbbbbccc")
+
+        self.assertEqual(
+            self.app.calls,
+            [('GET', '/v1/AUTH_test/mancon/manifest-many-segments'),
+             ('GET', '/v1/AUTH_test/c?format=json&prefix=seg_'),
+             ('GET', '/v1/AUTH_test/c/seg_01'),
+             ('GET', '/v1/AUTH_test/c/seg_02'),
+             ('GET', '/v1/AUTH_test/c/seg_03')])
+
+    def test_get_range_many_segments_satisfiability_unknown(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-many-segments',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=10-22'})
+        with mock.patch(LIMIT, 3):
+            status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "200 OK")
+        # this requires multiple pages of container listing, so we can't send
+        # a Content-Length header
+        self.assertEqual(headers.get("Content-Length"), None)
+        self.assertEqual(body, "aaaaabbbbbcccccdddddeeeee")
+
+    def test_get_suffix_range(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=-40'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "206 Partial Content")
+        self.assertEqual(headers["Content-Length"], "25")
+        self.assertEqual(body, "aaaaabbbbbcccccdddddeeeee")
+
+    def test_get_suffix_range_many_segments(self):
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-many-segments',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=-5'})
+        with mock.patch(LIMIT, 3):
+            status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "200 OK")
+        self.assertEqual(headers.get("Content-Length"), None)
+        self.assertEqual(headers.get("Content-Range"), None)
+        self.assertEqual(body, "aaaaabbbbbcccccdddddeeeee")
+
+    def test_get_multi_range(self):
+        # DLO doesn't support multi-range GETs. The way that you express that
+        # in HTTP is to return a 200 response containing the whole entity.
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-many-segments',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=5-9,15-19'})
+        with mock.patch(LIMIT, 3):
+            status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(status, "200 OK")
+        self.assertEqual(headers.get("Content-Length"), None)
+        self.assertEqual(headers.get("Content-Range"), None)
+        self.assertEqual(body, "aaaaabbbbbcccccdddddeeeee")
+
+    def test_error_fetching_first_segment(self):
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/seg_01',
+            swob.HTTPForbidden, {}, None)
+
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        status, headers, body, exc = self.call_dlo(req, expect_exception=True)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertTrue(isinstance(exc, exceptions.SegmentError))
+
+        self.assertEqual(status, "200 OK")
+        self.assertEqual(body, '')  # error right away -> no body bytes sent
+
+    def test_error_fetching_second_segment(self):
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/seg_02',
+            swob.HTTPForbidden, {}, None)
+
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        status, headers, body, exc = self.call_dlo(req, expect_exception=True)
+        headers = swob.HeaderKeyDict(headers)
+
+        self.assertTrue(isinstance(exc, exceptions.SegmentError))
+        self.assertEqual(status, "200 OK")
+        self.assertEqual(''.join(body), "aaaaa")  # first segment made it out
+
+    def test_error_listing_container_first_listing_request(self):
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=seg_',
+            swob.HTTPNotFound, {}, None)
+
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-many-segments',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=-5'})
+        with mock.patch(LIMIT, 3):
+            status, headers, body = self.call_dlo(req)
+        self.assertEqual(status, "404 Not Found")
+
+    def test_error_listing_container_second_listing_request(self):
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=seg_&marker=seg_03',
+            swob.HTTPNotFound, {}, None)
+
+        req = swob.Request.blank('/v1/AUTH_test/mancon/manifest-many-segments',
+                                 environ={'REQUEST_METHOD': 'GET'},
+                                 headers={'Range': 'bytes=-5'})
+        with mock.patch(LIMIT, 3):
+            status, headers, body, exc = self.call_dlo(
+                req, expect_exception=True)
+        self.assertTrue(isinstance(exc, exceptions.ListingIterError))
+        self.assertEqual(status, "200 OK")
+        self.assertEqual(body, "aaaaabbbbbccccc")
+
+    def test_etag_comparison_ignores_quotes(self):
+        # a little future-proofing here in case we ever fix this
+        self.app.register(
+            'HEAD', '/v1/AUTH_test/mani/festo',
+            swob.HTTPOk, {'Content-Length': '0', 'Etag': 'blah',
+                          'X-Object-Manifest': 'c/quotetags'}, None)
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=quotetags',
+            swob.HTTPOk, {'Content-Type': 'application/json; charset=utf-8'},
+            json.dumps([{"hash": "\"abc\"", "bytes": 5, "name": "quotetags1",
+                         "last_modified": "2013-11-22T02:42:14.261620",
+                         "content-type": "application/octet-stream"},
+                        {"hash": "def", "bytes": 5, "name": "quotetags2",
+                         "last_modified": "2013-11-22T02:42:14.261620",
+                         "content-type": "application/octet-stream"}]))
+
+        req = swob.Request.blank('/v1/AUTH_test/mani/festo',
+                                 environ={'REQUEST_METHOD': 'HEAD'})
+        status, headers, body = self.call_dlo(req)
+        headers = swob.HeaderKeyDict(headers)
+        self.assertEqual(headers["Etag"],
+                         '"' + hashlib.md5("abcdef").hexdigest() + '"')
+
+    def test_object_prefix_quoting(self):
+        self.app.register(
+            'GET', '/v1/AUTH_test/man/accent',
+            swob.HTTPOk, {'Content-Length': '0', 'Etag': 'blah',
+                          'X-Object-Manifest': u'c/é'.encode('utf-8')}, None)
+
+        segs = [{"hash": "etag1", "bytes": 5, "name": u"é1"},
+                {"hash": "etag2", "bytes": 5, "name": u"é2"}]
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=%C3%A9',
+            swob.HTTPOk, {'Content-Type': 'application/json'},
+            json.dumps(segs))
+
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/\xC3\xa91',
+            swob.HTTPOk, {'Content-Length': '5', 'Etag': 'etag1'},
+            "AAAAA")
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/\xC3\xA92',
+            swob.HTTPOk, {'Content-Length': '5', 'Etag': 'etag2'},
+            "BBBBB")
+
+        req = swob.Request.blank('/v1/AUTH_test/man/accent',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        status, headers, body = self.call_dlo(req)
+        self.assertEqual(status, "200 OK")
+        self.assertEqual(body, "AAAAABBBBB")
+
+    def test_get_taking_too_long(self):
+        the_time = [time.time()]
+
+        def mock_time():
+            return the_time[0]
+
+        # this is just a convenient place to hang a time jump
+        def mock_is_success(status_int):
+            the_time[0] += 9 * 3600
+            return status_int // 100 == 2
+
+        req = swob.Request.blank(
+            '/v1/AUTH_test/mancon/manifest',
+            environ={'REQUEST_METHOD': 'GET'})
+
+        with contextlib.nested(
+                mock.patch('swift.common.utils.time.time', mock_time),
+                mock.patch('swift.common.utils.is_success', mock_is_success),
+                mock.patch.object(dlo, 'is_success', mock_is_success)):
+            status, headers, body, exc = self.call_dlo(
+                req, expect_exception=True)
+
+        self.assertEqual(status, '200 OK')
+        self.assertEqual(body, 'aaaaabbbbbccccc')
+        self.assertTrue(isinstance(exc, exceptions.SegmentError))
+
+
+def fake_start_response(*args, **kwargs):
+    pass
+
+
+class TestDloCopyHook(DloTestCase):
+    def setUp(self):
+        super(TestDloCopyHook, self).setUp()
+
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/o1', swob.HTTPOk,
+            {'Content-Length': '10', 'Etag': 'o1-etag'},
+            "aaaaaaaaaa")
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/o2', swob.HTTPOk,
+            {'Content-Length': '10', 'Etag': 'o2-etag'},
+            "bbbbbbbbbb")
+        self.app.register(
+            'GET', '/v1/AUTH_test/c/man',
+            swob.HTTPOk, {'X-Object-Manifest': 'c/o'},
+            "manifest-contents")
+
+        lm = '2013-11-22T02:42:13.781760'
+        ct = 'application/octet-stream'
+        segs = [{"hash": "o1-etag", "bytes": 10, "name": "o1",
+                 "last_modified": lm, "content_type": ct},
+                {"hash": "o2-etag", "bytes": 5, "name": "o2",
+                 "last_modified": lm, "content_type": ct}]
+
+        self.app.register(
+            'GET', '/v1/AUTH_test/c?format=json&prefix=o',
+            swob.HTTPOk, {'Content-Type': 'application/json; charset=utf-8'},
+            json.dumps(segs))
+
+        copy_hook = [None]
+
+        # slip this guy in there to pull out the hook
+        def extract_copy_hook(env, sr):
+            copy_hook[0] = env.get('swift.copy_response_hook')
+            return self.app(env, sr)
+
+        self.dlo = dlo.filter_factory({})(extract_copy_hook)
+
+        req = swob.Request.blank('/v1/AUTH_test/c/o1',
+                                 environ={'REQUEST_METHOD': 'GET'})
+        self.dlo(req.environ, fake_start_response)
+        self.copy_hook = copy_hook[0]
+
+        self.assertTrue(self.copy_hook is not None)  # sanity check
+
+    def test_copy_hook_passthrough(self):
+        req = swob.Request.blank('/v1/AUTH_test/c/man')
+        # no X-Object-Manifest header, so do nothing
+        resp = swob.Response(request=req, status=200)
+
+        modified_resp = self.copy_hook(req, resp)
+        self.assertTrue(modified_resp is resp)
+
+    def test_copy_hook_manifest(self):
+        req = swob.Request.blank('/v1/AUTH_test/c/man')
+        resp = swob.Response(request=req, status=200,
+                             headers={"X-Object-Manifest": "c/o"},
+                             app_iter=["manifest"])
+
+        modified_resp = self.copy_hook(req, resp)
+        self.assertTrue(modified_resp is not resp)
+        self.assertEqual(modified_resp.etag,
+                         hashlib.md5("o1-etago2-etag").hexdigest())
+
+
+class TestDloConfiguration(unittest.TestCase):
+    """
+    For backwards compatibility, we will read a couple of values out of the
+    proxy's config section if we don't have any config values.
+    """
+
+    def test_skip_defaults_if_configured(self):
+        # The presence of even one config value in our config section means we
+        # won't go looking for the proxy config at all.
+        proxy_conf = dedent("""
+        [DEFAULT]
+        bind_ip = 10.4.5.6
+
+        [pipeline:main]
+        pipeline = catch_errors dlo ye-olde-proxy-server
+
+        [filter:dlo]
+        use = egg:swift#dlo
+        max_get_time = 3600
+
+        [app:ye-olde-proxy-server]
+        use = egg:swift#proxy
+        rate_limit_segments_per_sec = 7
+        rate_limit_after_segment = 13
+        max_get_time = 2900
+        """)
+
+        conffile = tempfile.NamedTemporaryFile()
+        conffile.write(proxy_conf)
+        conffile.flush()
+
+        mware = dlo.filter_factory({
+            'max_get_time': '3600',
+            '__file__': conffile.name
+        })("no app here")
+
+        self.assertEqual(1, mware.rate_limit_segments_per_sec)
+        self.assertEqual(10, mware.rate_limit_after_segment)
+        self.assertEqual(3600, mware.max_get_time)
+
+    def test_finding_defaults_from_file(self):
+        # If DLO has no config vars, go pull them from the proxy server's
+        # config section
+        proxy_conf = dedent("""
+        [DEFAULT]
+        bind_ip = 10.4.5.6
+
+        [pipeline:main]
+        pipeline = catch_errors dlo ye-olde-proxy-server
+
+        [filter:dlo]
+        use = egg:swift#dlo
+
+        [app:ye-olde-proxy-server]
+        use = egg:swift#proxy
+        rate_limit_after_segment = 13
+        max_get_time = 2900
+        """)
+
+        conffile = tempfile.NamedTemporaryFile()
+        conffile.write(proxy_conf)
+        conffile.flush()
+
+        mware = dlo.filter_factory({
+            '__file__': conffile.name
+        })("no app here")
+
+        self.assertEqual(1, mware.rate_limit_segments_per_sec)
+        self.assertEqual(13, mware.rate_limit_after_segment)
+        self.assertEqual(2900, mware.max_get_time)
+
+    def test_finding_defaults_from_dir(self):
+        # If DLO has no config vars, go pull them from the proxy server's
+        # config section
+        proxy_conf1 = dedent("""
+        [DEFAULT]
+        bind_ip = 10.4.5.6
+
+        [pipeline:main]
+        pipeline = catch_errors dlo ye-olde-proxy-server
+        """)
+
+        proxy_conf2 = dedent("""
+        [filter:dlo]
+        use = egg:swift#dlo
+
+        [app:ye-olde-proxy-server]
+        use = egg:swift#proxy
+        rate_limit_after_segment = 13
+        max_get_time = 2900
+        """)
+
+        conf_dir = tempfile.mkdtemp()
+
+        conffile1 = tempfile.NamedTemporaryFile(dir=conf_dir, suffix='.conf')
+        conffile1.write(proxy_conf1)
+        conffile1.flush()
+
+        conffile2 = tempfile.NamedTemporaryFile(dir=conf_dir, suffix='.conf')
+        conffile2.write(proxy_conf2)
+        conffile2.flush()
+
+        mware = dlo.filter_factory({
+            '__file__': conf_dir
+        })("no app here")
+
+        self.assertEqual(1, mware.rate_limit_segments_per_sec)
+        self.assertEqual(13, mware.rate_limit_after_segment)
+        self.assertEqual(2900, mware.max_get_time)
+
+
+if __name__ == '__main__':
+    unittest.main()

test/unit/common/middleware/test_slo.py

@@ -16,14 +16,15 @@
 
 import time
 import unittest
-from copy import deepcopy
+from contextlib import nested
 from mock import patch
 from hashlib import md5
 from swift.common import swob, utils
 from swift.common.exceptions import ListingIterError, SegmentError
 from swift.common.middleware import slo
-from swift.common.utils import json, split_path
 from swift.common.swob import Request, Response, HTTPException
+from swift.common.utils import json
+from test.unit.common.middleware.helpers import FakeSwift
 
 
 test_xml_data = '''<?xml version="1.0" encoding="UTF-8"?>
@@ -44,71 +45,6 @@ def fake_start_response(*args, **kwargs):
     pass
 
 
-class FakeSwift(object):
-    def __init__(self):
-        self._calls = []
-        self.req_method_paths = []
-        self.uploaded = {}
-        # mapping of (method, path) --> (response class, headers, body)
-        self._responses = {}
-
-    def __call__(self, env, start_response):
-        method = env['REQUEST_METHOD']
-        path = env['PATH_INFO']
-        _, acc, cont, obj = split_path(env['PATH_INFO'], 0, 4,
-                                       rest_with_last=True)
-
-        headers = swob.Request(env).headers
-        self._calls.append((method, path, headers))
-
-        try:
-            resp_class, raw_headers, body = self._responses[(method, path)]
-            headers = swob.HeaderKeyDict(raw_headers)
-        except KeyError:
-            if method == 'HEAD' and ('GET', path) in self._responses:
-                resp_class, raw_headers, _ = self._responses[('GET', path)]
-                body = None
-                headers = swob.HeaderKeyDict(raw_headers)
-            elif method == 'GET' and obj and path in self.uploaded:
-                resp_class = swob.HTTPOk
-                headers, body = self.uploaded[path]
-            else:
-                raise
-
-        # simulate object PUT
-        if method == 'PUT' and obj:
-            input = env['wsgi.input'].read()
-            etag = md5(input).hexdigest()
-            headers.setdefault('Etag', etag)
-            headers.setdefault('Content-Length', len(input))
-
-            # keep it for subsequent GET requests later
-            self.uploaded[path] = (deepcopy(headers), input)
-            if "CONTENT_TYPE" in env:
-                self.uploaded[path][0]['Content-Type'] = env["CONTENT_TYPE"]
-
-        req = swob.Request(env)
-        # range requests ought to work, hence conditional_response=True
-        resp = resp_class(req=req, headers=headers, body=body,
-                          conditional_response=True)
-        return resp(env, start_response)
-
-    @property
-    def calls(self):
-        return [(method, path) for method, path, headers in self._calls]
-
-    @property
-    def calls_with_headers(self):
-        return self._calls
-
-    @property
-    def call_count(self):
-        return len(self._calls)
-
-    def register(self, method, path, response_class, headers, body):
-        self._responses[(method, path)] = (response_class, headers, body)
-
-
 class SloTestCase(unittest.TestCase):
     def setUp(self):
         self.app = FakeSwift()
@@ -372,7 +308,9 @@ class TestSloPutManifest(SloTestCase):
 
         # go behind SLO's back and see what actually got stored
         req = Request.blank(
-            '/v1/AUTH_test/checktest/man_3?multipart-manifest=get',
+            # this string looks weird, but it's just an artifact
+            # of FakeSwift
+            '/v1/AUTH_test/checktest/man_3?multipart-manifest=put',
             environ={'REQUEST_METHOD': 'GET'})
         status, headers, body = self.call_app(req)
         headers = dict(headers)
@@ -440,6 +378,9 @@ class TestSloDeleteManifest(SloTestCase):
                           'X-Static-Large-Object': 'true'},
             json.dumps([{'name': '/deltest/b_2', 'hash': 'a', 'bytes': '1'},
                         {'name': '/deltest/c_3', 'hash': 'b', 'bytes': '2'}]))
+        self.app.register(
+            'DELETE', '/v1/AUTH_test/deltest/man-all-there',
+            swob.HTTPNoContent, {}, None)
         self.app.register(
             'DELETE', '/v1/AUTH_test/deltest/gone',
             swob.HTTPNotFound, {}, None)
@@ -545,8 +486,10 @@ class TestSloDeleteManifest(SloTestCase):
                      'HTTP_ACCEPT': 'application/json'})
         status, headers, body = self.call_slo(req)
         resp_data = json.loads(body)
-        self.assertEquals(self.app.calls,
-                          [('GET', '/v1/AUTH_test/deltest/man_404')])
+        self.assertEquals(
+            self.app.calls,
+            [('GET',
+              '/v1/AUTH_test/deltest/man_404?multipart-manifest=get')])
         self.assertEquals(resp_data['Response Status'], '200 OK')
         self.assertEquals(resp_data['Response Body'], '')
         self.assertEquals(resp_data['Number Deleted'], 0)
@@ -560,11 +503,16 @@ class TestSloDeleteManifest(SloTestCase):
                      'HTTP_ACCEPT': 'application/json'})
         status, headers, body = self.call_slo(req)
         resp_data = json.loads(body)
-        self.assertEquals(self.app.calls,
-                          [('GET', '/v1/AUTH_test/deltest/man'),
-                           ('DELETE', '/v1/AUTH_test/deltest/gone'),
-                           ('DELETE', '/v1/AUTH_test/deltest/b_2'),
-                           ('DELETE', '/v1/AUTH_test/deltest/man')])
+        self.assertEquals(
+            self.app.calls,
+            [('GET',
+              '/v1/AUTH_test/deltest/man?multipart-manifest=get'),
+             ('DELETE',
+              '/v1/AUTH_test/deltest/gone?multipart-manifest=delete'),
+             ('DELETE',
+              '/v1/AUTH_test/deltest/b_2?multipart-manifest=delete'),
+             ('DELETE',
+              '/v1/AUTH_test/deltest/man?multipart-manifest=delete')])
         self.assertEquals(resp_data['Response Status'], '200 OK')
         self.assertEquals(resp_data['Number Deleted'], 2)
         self.assertEquals(resp_data['Number Not Found'], 1)
@@ -574,11 +522,14 @@ class TestSloDeleteManifest(SloTestCase):
             '/v1/AUTH_test/deltest/man-all-there?multipart-manifest=delete',
             environ={'REQUEST_METHOD': 'DELETE'})
         self.call_slo(req)
-        self.assertEquals(self.app.calls,
-                          [('GET', '/v1/AUTH_test/deltest/man-all-there'),
-                           ('DELETE', '/v1/AUTH_test/deltest/b_2'),
-                           ('DELETE', '/v1/AUTH_test/deltest/c_3'),
-                           ('DELETE', '/v1/AUTH_test/deltest/man-all-there')])
+        self.assertEquals(
+            self.app.calls,
+            [('GET',
+              '/v1/AUTH_test/deltest/man-all-there?multipart-manifest=get'),
+             ('DELETE', '/v1/AUTH_test/deltest/b_2?multipart-manifest=delete'),
+             ('DELETE', '/v1/AUTH_test/deltest/c_3?multipart-manifest=delete'),
+             ('DELETE', ('/v1/AUTH_test/deltest/' +
+                         'man-all-there?multipart-manifest=delete'))])
 
     def test_handle_multipart_delete_nested(self):
         req = Request.blank(
@@ -588,15 +539,24 @@ class TestSloDeleteManifest(SloTestCase):
         self.call_slo(req)
         self.assertEquals(
             set(self.app.calls),
-            set([('GET', '/v1/AUTH_test/deltest/manifest-with-submanifest'),
-                 ('GET', '/v1/AUTH_test/deltest/submanifest'),
-                 ('DELETE', '/v1/AUTH_test/deltest/a_1'),
-                 ('DELETE', '/v1/AUTH_test/deltest/b_2'),
-                 ('DELETE', '/v1/AUTH_test/deltest/c_3'),
-                 ('DELETE', '/v1/AUTH_test/deltest/submanifest'),
-                 ('DELETE', '/v1/AUTH_test/deltest/d_3'),
-                 ('DELETE', '/v1/AUTH_test/deltest/' +
-                  'manifest-with-submanifest')]))
+            set([('GET', '/v1/AUTH_test/deltest/' +
+                  'manifest-with-submanifest?multipart-manifest=get'),
+                 ('GET', '/v1/AUTH_test/deltest/' +
+                  'submanifest?multipart-manifest=get'),
+                 ('DELETE',
+                  '/v1/AUTH_test/deltest/a_1?multipart-manifest=delete'),
+                 ('DELETE',
+                  '/v1/AUTH_test/deltest/b_2?multipart-manifest=delete'),
+                 ('DELETE',
+                  '/v1/AUTH_test/deltest/c_3?multipart-manifest=delete'),
+                 ('DELETE',
+                  '/v1/AUTH_test/deltest/' +
+                  'submanifest?multipart-manifest=delete'),
+                 ('DELETE',
+                  '/v1/AUTH_test/deltest/d_3?multipart-manifest=delete'),
+                 ('DELETE',
+                  '/v1/AUTH_test/deltest/' +
+                  'manifest-with-submanifest?multipart-manifest=delete')]))
 
     def test_handle_multipart_delete_nested_too_many_segments(self):
         req = Request.blank(
@@ -620,15 +580,16 @@ class TestSloDeleteManifest(SloTestCase):
                      'HTTP_ACCEPT': 'application/json'})
         status, headers, body = self.call_slo(req)
         resp_data = json.loads(body)
-        self.assertEquals(self.app.calls,
-                          [('GET', '/v1/AUTH_test/deltest/' +
-                            'manifest-missing-submanifest'),
-                           ('DELETE', '/v1/AUTH_test/deltest/a_1'),
-                           ('GET', '/v1/AUTH_test/deltest/' +
-                            'missing-submanifest'),
-                           ('DELETE', '/v1/AUTH_test/deltest/d_3'),
-                           ('DELETE', '/v1/AUTH_test/deltest/' +
-                            'manifest-missing-submanifest')])
+        self.assertEquals(
+            self.app.calls,
+            [('GET', '/v1/AUTH_test/deltest/' +
+              'manifest-missing-submanifest?multipart-manifest=get'),
+             ('DELETE', '/v1/AUTH_test/deltest/a_1?multipart-manifest=delete'),
+             ('GET', '/v1/AUTH_test/deltest/' +
+              'missing-submanifest?multipart-manifest=get'),
+             ('DELETE', '/v1/AUTH_test/deltest/d_3?multipart-manifest=delete'),
+             ('DELETE', '/v1/AUTH_test/deltest/' +
+              'manifest-missing-submanifest?multipart-manifest=delete')])
         self.assertEquals(resp_data['Response Status'], '200 OK')
         self.assertEquals(resp_data['Response Body'], '')
         self.assertEquals(resp_data['Number Deleted'], 3)
@@ -677,8 +638,9 @@ class TestSloDeleteManifest(SloTestCase):
                      'HTTP_ACCEPT': 'application/json'})
         status, headers, body = self.call_slo(req)
         resp_data = json.loads(body)
-        self.assertEquals(self.app.calls,
-                          [('GET', '/v1/AUTH_test/deltest/a_1')])
+        self.assertEquals(
+            self.app.calls,
+            [('GET', '/v1/AUTH_test/deltest/a_1?multipart-manifest=get')])
         self.assertEquals(resp_data['Response Status'], '400 Bad Request')
         self.assertEquals(resp_data['Response Body'], '')
         self.assertEquals(resp_data['Number Deleted'], 0)
@@ -694,7 +656,8 @@ class TestSloDeleteManifest(SloTestCase):
         status, headers, body = self.call_slo(req)
         resp_data = json.loads(body)
         self.assertEquals(self.app.calls,
-                          [('GET', '/v1/AUTH_test/deltest/manifest-badjson')])
+                          [('GET', '/v1/AUTH_test/deltest/' +
+                            'manifest-badjson?multipart-manifest=get')])
         self.assertEquals(resp_data['Response Status'], '400 Bad Request')
         self.assertEquals(resp_data['Response Body'], '')
         self.assertEquals(resp_data['Number Deleted'], 0)
@@ -711,13 +674,15 @@ class TestSloDeleteManifest(SloTestCase):
                      'HTTP_ACCEPT': 'application/json'})
         status, headers, body = self.call_slo(req)
         resp_data = json.loads(body)
-        self.assertEquals(self.app.calls,
-                          [('GET', '/v1/AUTH_test/deltest/' +
-                            'manifest-with-unauth-segment'),
-                           ('DELETE', '/v1/AUTH_test/deltest/a_1'),
-                           ('DELETE', '/v1/AUTH_test/deltest-unauth/q_17'),
-                           ('DELETE', '/v1/AUTH_test/deltest/' +
-                            'manifest-with-unauth-segment')])
+        self.assertEquals(
+            self.app.calls,
+            [('GET', '/v1/AUTH_test/deltest/' +
+              'manifest-with-unauth-segment?multipart-manifest=get'),
+             ('DELETE', '/v1/AUTH_test/deltest/a_1?multipart-manifest=delete'),
+             ('DELETE', '/v1/AUTH_test/deltest-unauth/' +
+              'q_17?multipart-manifest=delete'),
+             ('DELETE', '/v1/AUTH_test/deltest/' +
+              'manifest-with-unauth-segment?multipart-manifest=delete')])
         self.assertEquals(resp_data['Response Status'], '400 Bad Request')
         self.assertEquals(resp_data['Response Body'], '')
         self.assertEquals(resp_data['Number Deleted'], 2)
@@ -1253,8 +1218,9 @@ class TestSloGetManifest(SloTestCase):
             '/v1/AUTH_test/gettest/manifest-abcd',
             environ={'REQUEST_METHOD': 'GET'})
 
-        with patch.object(slo, 'time', mock_time):
-            with patch.object(slo, 'is_success', mock_is_success):
+        with nested(patch.object(slo, 'is_success', mock_is_success),
+                    patch('swift.common.utils.time.time', mock_time),
+                    patch('swift.common.utils.is_success', mock_is_success)):
                 status, headers, body, exc = self.call_slo(
                     req, expect_exception=True)
 

test/unit/common/test_constraints.py

@@ -166,46 +166,6 @@ class TestConstraints(unittest.TestCase):
         self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
         self.assert_('Content-Type' in resp.body)
 
-    def test_check_object_manifest_header(self):
-        resp = constraints.check_object_creation(Request.blank(
-            '/',
-            headers={'X-Object-Manifest': 'container/prefix',
-                     'Content-Length': '0',
-                     'Content-Type': 'text/plain'}), 'manifest')
-        self.assert_(not resp)
-        resp = constraints.check_object_creation(Request.blank(
-            '/',
-            headers={'X-Object-Manifest': 'container',
-                     'Content-Length': '0',
-                     'Content-Type': 'text/plain'}), 'manifest')
-        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
-        resp = constraints.check_object_creation(Request.blank(
-            '/',
-            headers={'X-Object-Manifest': '/container/prefix',
-                     'Content-Length': '0',
-                     'Content-Type': 'text/plain'}), 'manifest')
-        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
-        resp = constraints.check_object_creation(Request.blank(
-            '/',
-            headers={'X-Object-Manifest': 'container/prefix?query=param',
-                     'Content-Length': '0',
-                     'Content-Type': 'text/plain'}), 'manifest')
-        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
-        resp = constraints.check_object_creation(Request.blank(
-            '/',
-            headers={'X-Object-Manifest': 'container/prefix&query=param',
-                     'Content-Length': '0',
-                     'Content-Type': 'text/plain'}), 'manifest')
-        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
-        resp = constraints.check_object_creation(
-            Request.blank(
-                '/', headers={
-                    'X-Object-Manifest': 'http://host/container/prefix',
-                    'Content-Length': '0',
-                    'Content-Type': 'text/plain'}),
-                'manifest')
-        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
-
     def test_check_mount(self):
         self.assertFalse(constraints.check_mount('', ''))
         with mock.patch("swift.common.constraints.ismount", MockTrue()):

test/unit/common/test_wsgi.py

@@ -141,15 +141,23 @@ class TestWSGI(unittest.TestCase):
             _fake_rings(t)
             app, conf, logger, log_name = wsgi.init_request_processor(
                 conf_file, 'proxy-server')
-        # verify pipeline is catch_errors -> proxy-server
+        # verify pipeline is catch_errors -> dlo -> proxy-server
         expected = swift.common.middleware.catch_errors.CatchErrorMiddleware
         self.assert_(isinstance(app, expected))
+
         app = app.app
         expected = swift.common.middleware.gatekeeper.GatekeeperMiddleware
         self.assert_(isinstance(app, expected))
-        self.assert_(isinstance(app.app, swift.proxy.server.Application))
+
+        app = app.app
+        expected = swift.common.middleware.dlo.DynamicLargeObject
+        self.assert_(isinstance(app, expected))
+
+        app = app.app
+        expected = swift.proxy.server.Application
+        self.assert_(isinstance(app, expected))
         # config settings applied to app instance
-        self.assertEquals(0.2, app.app.conn_timeout)
+        self.assertEquals(0.2, app.conn_timeout)
         # appconfig returns values from 'proxy-server' section
         expected = {
             '__file__': conf_file,
@@ -841,7 +849,8 @@ class TestPipelineModification(unittest.TestCase):
 
         self.assertEqual(self.pipeline_modules(app),
                          ['swift.common.middleware.catch_errors',
-                         'swift.common.middleware.gatekeeper',
+                          'swift.common.middleware.gatekeeper',
+                          'swift.common.middleware.dlo',
                           'swift.proxy.server'])
 
     def test_proxy_modify_wsgi_pipeline(self):
@@ -870,9 +879,10 @@ class TestPipelineModification(unittest.TestCase):
 
         self.assertEqual(self.pipeline_modules(app),
                          ['swift.common.middleware.catch_errors',
-                         'swift.common.middleware.gatekeeper',
-                         'swift.common.middleware.healthcheck',
-                         'swift.proxy.server'])
+                          'swift.common.middleware.gatekeeper',
+                          'swift.common.middleware.dlo',
+                          'swift.common.middleware.healthcheck',
+                          'swift.proxy.server'])
 
     def test_proxy_modify_wsgi_pipeline_ordering(self):
         config = """
@@ -904,10 +914,10 @@ class TestPipelineModification(unittest.TestCase):
             {'name': 'catch_errors'},
             # already in pipeline
             {'name': 'proxy_logging',
-             'after': ['catch_errors']},
+             'after_fn': lambda _: ['catch_errors']},
             # not in pipeline, comes after more than one thing
             {'name': 'container_quotas',
-             'after': ['catch_errors', 'bulk']}]
+             'after_fn': lambda _: ['catch_errors', 'bulk']}]
 
         contents = dedent(config)
         with temptree(['proxy-server.conf']) as t:
@@ -967,6 +977,7 @@ class TestPipelineModification(unittest.TestCase):
         self.assertEqual(self.pipeline_modules(app), [
             'swift.common.middleware.catch_errors',
             'swift.common.middleware.gatekeeper',
+            'swift.common.middleware.dlo',
             'swift.common.middleware.healthcheck',
             'swift.proxy.server'])
 
@@ -979,6 +990,7 @@ class TestPipelineModification(unittest.TestCase):
             'swift.common.middleware.gatekeeper',
             'swift.common.middleware.healthcheck',
             'swift.common.middleware.catch_errors',
+            'swift.common.middleware.dlo',
             'swift.proxy.server'])
 
     def test_catch_errors_gatekeeper_configured_not_at_start(self):
@@ -990,6 +1002,7 @@ class TestPipelineModification(unittest.TestCase):
             'swift.common.middleware.healthcheck',
             'swift.common.middleware.catch_errors',
             'swift.common.middleware.gatekeeper',
+            'swift.common.middleware.dlo',
             'swift.proxy.server'])
 
 if __name__ == '__main__':

test/unit/obj/test_server.py

@@ -1793,56 +1793,6 @@ class TestObjectController(unittest.TestCase):
         self.assertEquals(resp.status_int, 200)
         self.assertEquals(resp.headers['content-encoding'], 'gzip')
 
-    def test_manifest_header(self):
-        timestamp = normalize_timestamp(time())
-        req = Request.blank(
-            '/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
-            headers={'X-Timestamp': timestamp,
-                     'Content-Type': 'text/plain',
-                     'Content-Length': '0',
-                     'X-Object-Manifest': 'c/o/'})
-        resp = req.get_response(self.object_controller)
-        self.assertEquals(resp.status_int, 201)
-        objfile = os.path.join(
-            self.testdir, 'sda1',
-            storage_directory(diskfile.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
-            timestamp + '.data')
-        self.assert_(os.path.isfile(objfile))
-        self.assertEquals(
-            diskfile.read_metadata(objfile),
-            {'X-Timestamp': timestamp,
-             'Content-Length': '0',
-             'Content-Type': 'text/plain',
-             'name': '/a/c/o',
-             'X-Object-Manifest': 'c/o/',
-             'ETag': 'd41d8cd98f00b204e9800998ecf8427e'})
-        req = Request.blank('/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'GET'})
-        resp = req.get_response(self.object_controller)
-        self.assertEquals(resp.status_int, 200)
-        self.assertEquals(resp.headers.get('x-object-manifest'), 'c/o/')
-
-    def test_manifest_head_request(self):
-        timestamp = normalize_timestamp(time())
-        req = Request.blank(
-            '/sda1/p/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
-            headers={'X-Timestamp': timestamp,
-                     'Content-Type': 'text/plain',
-                     'Content-Length': '0',
-                     'X-Object-Manifest': 'c/o/'})
-        req.body = 'hi'
-        resp = req.get_response(self.object_controller)
-        objfile = os.path.join(
-            self.testdir, 'sda1',
-            storage_directory(diskfile.DATADIR, 'p',
-                              hash_path('a', 'c', 'o')),
-            timestamp + '.data')
-        self.assert_(os.path.isfile(objfile))
-        req = Request.blank('/sda1/p/a/c/o',
-                            environ={'REQUEST_METHOD': 'HEAD'})
-        resp = req.get_response(self.object_controller)
-        self.assertEquals(resp.body, '')
-
     def test_async_update_http_connect(self):
         given_args = []
 

test/unit/proxy/test_mem_server.py

@@ -50,10 +50,6 @@ class TestAccountControllerFakeGetResponse(
     pass
 
 
-class TestSegmentedIterable(test_server.TestSegmentedIterable):
-    pass
-
-
 if __name__ == '__main__':
     setup()
     try:

test/unit/proxy/test_server.py

@@ -49,14 +49,12 @@ from swift.common.constraints import MAX_META_NAME_LENGTH, \
 from swift.common import utils
 from swift.common.utils import mkdirs, normalize_timestamp, NullLogger
 from swift.common.wsgi import monkey_patch_mimetools
-from swift.proxy.controllers.obj import SegmentedIterable
 from swift.proxy.controllers import base as proxy_base
 from swift.proxy.controllers.base import get_container_memcache_key, \
     get_account_memcache_key, cors_validation
 import swift.proxy.controllers
-from swift.common.swob import Request, Response, HTTPNotFound, \
-    HTTPUnauthorized
 from swift.common.request_helpers import get_sys_meta_prefix
+from swift.common.swob import Request, Response, HTTPUnauthorized
 
 # mocks
 logging.getLogger().addHandler(logging.StreamHandler(sys.stdout))
@@ -85,8 +83,6 @@ def do_setup(the_object_server):
     mkdirs(os.path.join(_testdir, 'sda1', 'tmp'))
     mkdirs(os.path.join(_testdir, 'sdb1'))
     mkdirs(os.path.join(_testdir, 'sdb1', 'tmp'))
-    _orig_container_listing_limit = \
-        swift.proxy.controllers.obj.CONTAINER_LISTING_LIMIT
     conf = {'devices': _testdir, 'swift_dir': _testdir,
             'mount_check': 'false', 'allowed_headers':
             'content-encoding, x-object-manifest, content-disposition, foo',
@@ -184,8 +180,6 @@ def setup():
 def teardown():
     for server in _test_coros:
         server.kill()
-    swift.proxy.controllers.obj.CONTAINER_LISTING_LIMIT = \
-        _orig_container_listing_limit
     rmtree(os.path.dirname(_testdir))
     utils.SysLogHandler = _orig_SysLogHandler
 
@@ -3289,400 +3283,6 @@ class TestObjectController(unittest.TestCase):
         headers = readuntil2crlfs(fd)
         self.assertEquals(headers[:len(exp)], exp)
 
-    def test_chunked_put_lobjects_with_nonzero_size_manifest_file(self):
-        # Create a container for our segmented/manifest object testing
-        (prolis, acc1lis, acc2lis, con1lis, con2lis, obj1lis, obj2lis) = \
-            _test_sockets
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/segmented_nonzero HTTP/1.1\r\nHost: localhost\r\n'
-                 'Connection: close\r\nX-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        # Create the object segments
-        segment_etags = []
-        for segment in xrange(5):
-            sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-            fd = sock.makefile()
-            fd.write('PUT /v1/a/segmented_nonzero/name/%s HTTP/1.1\r\nHost: '
-                     'localhost\r\nConnection: close\r\nX-Storage-Token: '
-                     't\r\nContent-Length: 5\r\n\r\n1234 ' % str(segment))
-            fd.flush()
-            headers = readuntil2crlfs(fd)
-            exp = 'HTTP/1.1 201'
-            self.assertEquals(headers[:len(exp)], exp)
-            segment_etags.append(md5('1234 ').hexdigest())
-
-        # Create the nonzero size manifest file
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/segmented_nonzero/name HTTP/1.1\r\nHost: '
-                 'localhost\r\nConnection: close\r\nX-Storage-Token: '
-                 't\r\nContent-Length: 5\r\n\r\nabcd ')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-
-        # Create the object manifest file
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('POST /v1/a/segmented_nonzero/name HTTP/1.1\r\nHost: '
-                 'localhost\r\nConnection: close\r\nX-Storage-Token: t\r\n'
-                 'X-Object-Manifest: segmented_nonzero/name/\r\n'
-                 'Foo: barbaz\r\nContent-Type: text/jibberish\r\n'
-                 '\r\n\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 202'
-        self.assertEquals(headers[:len(exp)], exp)
-
-        # Ensure retrieving the manifest file gets the whole object
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented_nonzero/name HTTP/1.1\r\nHost: '
-                 'localhost\r\nConnection: close\r\nX-Auth-Token: '
-                 't\r\n\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('X-Object-Manifest: segmented_nonzero/name/' in headers)
-        self.assert_('Content-Type: text/jibberish' in headers)
-        self.assert_('Foo: barbaz' in headers)
-        expected_etag = md5(''.join(segment_etags)).hexdigest()
-        self.assert_('Etag: "%s"' % expected_etag in headers)
-        body = fd.read()
-        self.assertEquals(body, '1234 1234 1234 1234 1234 ')
-
-        # Get lobjects with Range smaller than manifest file
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented_nonzero/name HTTP/1.1\r\nHost: '
-                 'localhost\r\nConnection: close\r\nX-Auth-Token: t\r\n'
-                 'Range: bytes=0-4\r\n\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 206'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('X-Object-Manifest: segmented_nonzero/name/' in headers)
-        self.assert_('Content-Type: text/jibberish' in headers)
-        self.assert_('Foo: barbaz' in headers)
-        expected_etag = md5(''.join(segment_etags)).hexdigest()
-        body = fd.read()
-        self.assertEquals(body, '1234 ')
-
-        # Get lobjects with Range bigger than manifest file
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented_nonzero/name HTTP/1.1\r\nHost: '
-                 'localhost\r\nConnection: close\r\nX-Auth-Token: t\r\n'
-                 'Range: bytes=11-15\r\n\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 206'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('X-Object-Manifest: segmented_nonzero/name/' in headers)
-        self.assert_('Content-Type: text/jibberish' in headers)
-        self.assert_('Foo: barbaz' in headers)
-        expected_etag = md5(''.join(segment_etags)).hexdigest()
-        body = fd.read()
-        self.assertEquals(body, '234 1')
-
-    def test_chunked_put_lobjects(self):
-        # Create a container for our segmented/manifest object testing
-        (prolis, acc1lis, acc2lis, con1lis, con2lis, obj1lis,
-         obj2lis) = _test_sockets
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/segmented%20object HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        # Create the object segments
-        segment_etags = []
-        for segment in xrange(5):
-            sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-            fd = sock.makefile()
-            fd.write('PUT /v1/a/segmented%%20object/object%%20name/%s '
-                     'HTTP/1.1\r\n'
-                     'Host: localhost\r\n'
-                     'Connection: close\r\n'
-                     'X-Storage-Token: t\r\n'
-                     'Content-Length: 5\r\n'
-                     '\r\n'
-                     '1234 ' % str(segment))
-            fd.flush()
-            headers = readuntil2crlfs(fd)
-            exp = 'HTTP/1.1 201'
-            self.assertEquals(headers[:len(exp)], exp)
-            segment_etags.append(md5('1234 ').hexdigest())
-        # Create the object manifest file
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/segmented%20object/object%20name HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n'
-                 'X-Object-Manifest: segmented%20object/object%20name/\r\n'
-                 'Content-Type: text/jibberish\r\n'
-                 'Foo: barbaz\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        # Check retrieving the listing the manifest would retrieve
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented%20object?prefix=object%20name/ '
-                 'HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        body = fd.read()
-        self.assertEquals(
-            body,
-            'object name/0\n'
-            'object name/1\n'
-            'object name/2\n'
-            'object name/3\n'
-            'object name/4\n')
-        # Ensure retrieving the manifest file gets the whole object
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented%20object/object%20name HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('X-Object-Manifest: segmented%20object/object%20name/' in
-                     headers)
-        self.assert_('Content-Type: text/jibberish' in headers)
-        self.assert_('Foo: barbaz' in headers)
-        expected_etag = md5(''.join(segment_etags)).hexdigest()
-        self.assert_('Etag: "%s"' % expected_etag in headers)
-        body = fd.read()
-        self.assertEquals(body, '1234 1234 1234 1234 1234 ')
-        # Do it again but exceeding the container listing limit
-        swift.proxy.controllers.obj.CONTAINER_LISTING_LIMIT = 2
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented%20object/object%20name HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('X-Object-Manifest: segmented%20object/object%20name/' in
-                     headers)
-        self.assert_('Content-Type: text/jibberish' in headers)
-        body = fd.read()
-        # A bit fragile of a test; as it makes the assumption that all
-        # will be sent in a single chunk.
-        self.assertEquals(
-            body, '19\r\n1234 1234 1234 1234 1234 \r\n0\r\n\r\n')
-        # Make a copy of the manifested object, which should
-        # error since the number of segments exceeds
-        # CONTAINER_LISTING_LIMIT.
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/segmented%20object/copy HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 'X-Copy-From: segmented%20object/object%20name\r\n'
-                 'Content-Length: 0\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 413'
-        self.assertEquals(headers[:len(exp)], exp)
-        body = fd.read()
-        # After adjusting the CONTAINER_LISTING_LIMIT, make a copy of
-        # the manifested object which should consolidate the segments.
-        swift.proxy.controllers.obj.CONTAINER_LISTING_LIMIT = 10000
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/segmented%20object/copy HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 'X-Copy-From: segmented%20object/object%20name\r\n'
-                 'Content-Length: 0\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        body = fd.read()
-        # Retrieve and validate the copy.
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented%20object/copy HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('x-object-manifest:' not in headers.lower())
-        self.assert_('Content-Length: 25\r' in headers)
-        body = fd.read()
-        self.assertEquals(body, '1234 1234 1234 1234 1234 ')
-        # Create an object manifest file pointing to nothing
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/segmented%20object/empty HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n'
-                 'X-Object-Manifest: segmented%20object/empty/\r\n'
-                 'Content-Type: text/jibberish\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        # Ensure retrieving the manifest file gives a zero-byte file
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/segmented%20object/empty HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('X-Object-Manifest: segmented%20object/empty/' in headers)
-        self.assert_('Content-Type: text/jibberish' in headers)
-        body = fd.read()
-        self.assertEquals(body, '')
-        # Check copy content type
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/c/obj HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n'
-                 'Content-Type: text/jibberish\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/c/obj2 HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n'
-                 'X-Copy-From: c/obj\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        # Ensure getting the copied file gets original content-type
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/c/obj2 HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('Content-Type: text/jibberish' in headers)
-        # Check set content type
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/c/obj3 HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n'
-                 'Content-Type: foo/bar\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        # Ensure getting the copied file gets original content-type
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/c/obj3 HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('Content-Type: foo/bar' in
-                     headers.split('\r\n'), repr(headers.split('\r\n')))
-        # Check set content type with charset
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('PUT /v1/a/c/obj4 HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Storage-Token: t\r\n'
-                 'Content-Length: 0\r\n'
-                 'Content-Type: foo/bar; charset=UTF-8\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 201'
-        self.assertEquals(headers[:len(exp)], exp)
-        # Ensure getting the copied file gets original content-type
-        sock = connect_tcp(('localhost', prolis.getsockname()[1]))
-        fd = sock.makefile()
-        fd.write('GET /v1/a/c/obj4 HTTP/1.1\r\n'
-                 'Host: localhost\r\n'
-                 'Connection: close\r\n'
-                 'X-Auth-Token: t\r\n'
-                 '\r\n')
-        fd.flush()
-        headers = readuntil2crlfs(fd)
-        exp = 'HTTP/1.1 200'
-        self.assertEquals(headers[:len(exp)], exp)
-        self.assert_('Content-Type: foo/bar; charset=UTF-8' in
-                     headers.split('\r\n'), repr(headers.split('\r\n')))
-
     def test_mismatched_etags(self):
         with save_globals():
             # no etag supplied, object servers return success w/ diff values
@@ -6115,313 +5715,6 @@ class Stub(object):
     pass
 
 
-class TestSegmentedIterable(unittest.TestCase):
-
-    def setUp(self):
-        self.controller = FakeObjectController()
-
-    def test_load_next_segment_unexpected_error(self):
-        # Iterator value isn't a dict
-        self.assertRaises(Exception,
-                          SegmentedIterable(self.controller, None,
-                                            [None])._load_next_segment)
-        self.assert_(self.controller.exception_args[0].startswith(
-                     'ERROR: While processing manifest'))
-
-    def test_load_next_segment_with_no_segments(self):
-        self.assertRaises(StopIteration,
-                          SegmentedIterable(self.controller, 'lc',
-                                            [])._load_next_segment)
-
-    def test_load_next_segment_with_one_segment(self):
-        segit = SegmentedIterable(self.controller, 'lc', [{'name':
-                                  'o1'}])
-        segit._load_next_segment()
-        self.assertEquals(
-            self.controller.GETorHEAD_base_args[0][4], '/a/lc/o1')
-        data = ''.join(segit.segment_iter)
-        self.assertEquals(data, '1')
-
-    def test_load_next_segment_with_two_segments(self):
-        segit = SegmentedIterable(self.controller, 'lc', [{'name':
-                                  'o1'}, {'name': 'o2'}])
-        segit._load_next_segment()
-        self.assertEquals(
-            self.controller.GETorHEAD_base_args[-1][4], '/a/lc/o1')
-        data = ''.join(segit.segment_iter)
-        self.assertEquals(data, '1')
-        segit._load_next_segment()
-        self.assertEquals(
-            self.controller.GETorHEAD_base_args[-1][4], '/a/lc/o2')
-        data = ''.join(segit.segment_iter)
-        self.assertEquals(data, '22')
-
-    def test_load_next_segment_rate_limiting(self):
-        sleep_calls = []
-
-        def _stub_sleep(sleepy_time):
-            sleep_calls.append(sleepy_time)
-        orig_sleep = swift.proxy.controllers.obj.sleep
-        try:
-            swift.proxy.controllers.obj.sleep = _stub_sleep
-            segit = SegmentedIterable(
-                self.controller, 'lc', [
-                    {'name': 'o1'}, {'name': 'o2'}, {'name': 'o3'},
-                    {'name': 'o4'}, {'name': 'o5'}])
-
-            # rate_limit_after_segment == 3, so the first 3 segments should
-            # invoke no sleeping.
-            for _ in xrange(3):
-                segit._load_next_segment()
-            self.assertEquals([], sleep_calls)
-            self.assertEquals(self.controller.GETorHEAD_base_args[-1][4],
-                              '/a/lc/o3')
-
-            # Loading of next (4th) segment starts rate-limiting.
-            segit._load_next_segment()
-            self.assertAlmostEqual(0.5, sleep_calls[0], places=2)
-            self.assertEquals(self.controller.GETorHEAD_base_args[-1][4],
-                              '/a/lc/o4')
-
-            sleep_calls = []
-            segit._load_next_segment()
-            self.assertAlmostEqual(0.5, sleep_calls[0], places=2)
-            self.assertEquals(self.controller.GETorHEAD_base_args[-1][4],
-                              '/a/lc/o5')
-        finally:
-            swift.proxy.controllers.obj.sleep = orig_sleep
-
-    def test_load_next_segment_range_req_rate_limiting(self):
-        sleep_calls = []
-
-        def _stub_sleep(sleepy_time):
-            sleep_calls.append(sleepy_time)
-        orig_sleep = swift.proxy.controllers.obj.sleep
-        try:
-            swift.proxy.controllers.obj.sleep = _stub_sleep
-            segit = SegmentedIterable(
-                self.controller, 'lc', [
-                    {'name': 'o0', 'bytes': 5}, {'name': 'o1', 'bytes': 5},
-                    {'name': 'o2', 'bytes': 1}, {'name': 'o3'}, {'name': 'o4'},
-                    {'name': 'o5'}, {'name': 'o6'}])
-
-            # this tests for a range request which skips over the whole first
-            # segment, after that 3 segments will be read in because the
-            # rate_limit_after_segment == 3, then sleeping starts
-            segit_iter = segit.app_iter_range(10, None)
-            segit_iter.next()
-            for _ in xrange(2):
-                # this is set to 2 instead of 3 because o2 was loaded after
-                # o0 and o1 were skipped.
-                segit._load_next_segment()
-            self.assertEquals([], sleep_calls)
-            self.assertEquals(self.controller.GETorHEAD_base_args[-1][4],
-                              '/a/lc/o4')
-
-            # Loading of next (5th) segment starts rate-limiting.
-            segit._load_next_segment()
-            self.assertAlmostEqual(0.5, sleep_calls[0], places=2)
-            self.assertEquals(self.controller.GETorHEAD_base_args[-1][4],
-                              '/a/lc/o5')
-
-            sleep_calls = []
-            segit._load_next_segment()
-            self.assertAlmostEqual(0.5, sleep_calls[0], places=2)
-            self.assertEquals(self.controller.GETorHEAD_base_args[-1][4],
-                              '/a/lc/o6')
-        finally:
-            swift.proxy.controllers.obj.sleep = orig_sleep
-
-    def test_load_next_segment_with_two_segments_skip_first(self):
-        segit = SegmentedIterable(self.controller, 'lc', [{'name':
-                                  'o1'}, {'name': 'o2'}])
-        segit.ratelimit_index = 0
-        segit.listing.next()
-        segit._load_next_segment()
-        self.assertEquals(
-            self.controller.GETorHEAD_base_args[-1][4], '/a/lc/o2')
-        data = ''.join(segit.segment_iter)
-        self.assertEquals(data, '22')
-
-    def test_load_next_segment_with_seek(self):
-        segit = SegmentedIterable(self.controller, 'lc',
-                                  [{'name': 'o1', 'bytes': 1},
-                                   {'name': 'o2', 'bytes': 2}])
-        segit.ratelimit_index = 0
-        segit.listing.next()
-        segit.seek = 1
-        segit._load_next_segment()
-        self.assertEquals(
-            self.controller.GETorHEAD_base_args[-1][4], '/a/lc/o2')
-        self.assertEquals(
-            str(self.controller.GETorHEAD_base_args[-1][0].range),
-            'bytes=1-')
-        data = ''.join(segit.segment_iter)
-        self.assertEquals(data, '2')
-
-    def test_fetching_only_what_you_need(self):
-        segit = SegmentedIterable(self.controller, 'lc',
-                                  [{'name': 'o7', 'bytes': 7},
-                                   {'name': 'o8', 'bytes': 8},
-                                   {'name': 'o9', 'bytes': 9}])
-
-        body = ''.join(segit.app_iter_range(10, 20))
-        self.assertEqual('8888899999', body)
-
-        GoH_args = self.controller.GETorHEAD_base_args
-        self.assertEquals(2, len(GoH_args))
-
-        # Either one is fine, as they both indicate "from byte 3 to (the last)
-        # byte 8".
-        self.assert_(str(GoH_args[0][0].range) in ['bytes=3-', 'bytes=3-8'])
-
-        # This one must ask only for the bytes it needs; otherwise we waste
-        # bandwidth pulling bytes from the object server and then throwing
-        # them out
-        self.assertEquals(str(GoH_args[1][0].range), 'bytes=0-4')
-
-    def test_load_next_segment_with_get_error(self):
-
-        def local_GETorHEAD_base(*args):
-            return HTTPNotFound()
-
-        self.controller.GETorHEAD_base = local_GETorHEAD_base
-        self.assertRaises(Exception,
-                          SegmentedIterable(self.controller, 'lc',
-                          [{'name': 'o1'}])._load_next_segment)
-        self.assert_(self.controller.exception_args[0].startswith(
-                     'ERROR: While processing manifest'))
-        self.assertEquals(str(self.controller.exception_info[1]),
-                          'Could not load object segment /a/lc/o1: 404')
-
-    def test_iter_unexpected_error(self):
-        # Iterator value isn't a dict
-        self.assertRaises(Exception, ''.join,
-                          SegmentedIterable(self.controller, None, [None]))
-        self.assert_(self.controller.exception_args[0].startswith(
-            'ERROR: While processing manifest'))
-
-    def test_iter_with_no_segments(self):
-        segit = SegmentedIterable(self.controller, 'lc', [])
-        self.assertEquals(''.join(segit), '')
-
-    def test_iter_with_one_segment(self):
-        segit = SegmentedIterable(self.controller, 'lc', [{'name':
-                                  'o1'}])
-        segit.response = Stub()
-        self.assertEquals(''.join(segit), '1')
-
-    def test_iter_with_two_segments(self):
-        segit = SegmentedIterable(self.controller, 'lc', [{'name':
-                                  'o1'}, {'name': 'o2'}])
-        segit.response = Stub()
-        self.assertEquals(''.join(segit), '122')
-
-    def test_iter_with_get_error(self):
-
-        def local_GETorHEAD_base(*args):
-            return HTTPNotFound()
-
-        self.controller.GETorHEAD_base = local_GETorHEAD_base
-        self.assertRaises(Exception, ''.join,
-                          SegmentedIterable(self.controller, 'lc', [{'name':
-                                                                    'o1'}]))
-        self.assert_(self.controller.exception_args[0].startswith(
-                     'ERROR: While processing manifest'))
-        self.assertEquals(str(self.controller.exception_info[1]),
-                          'Could not load object segment /a/lc/o1: 404')
-
-    def test_app_iter_range_unexpected_error(self):
-        # Iterator value isn't a dict
-        self.assertRaises(Exception,
-                          SegmentedIterable(self.controller, None,
-                                            [None]).app_iter_range(None,
-                                                                   None).next)
-        self.assert_(self.controller.exception_args[0].startswith(
-            'ERROR: While processing manifest'))
-
-    def test_app_iter_range_with_no_segments(self):
-        self.assertEquals(''.join(SegmentedIterable(
-            self.controller, 'lc', []).app_iter_range(None, None)), '')
-        self.assertEquals(''.join(SegmentedIterable(
-            self.controller, 'lc', []).app_iter_range(3, None)), '')
-        self.assertEquals(''.join(SegmentedIterable(
-            self.controller, 'lc', []).app_iter_range(3, 5)), '')
-        self.assertEquals(''.join(SegmentedIterable(
-            self.controller, 'lc', []).app_iter_range(None, 5)), '')
-
-    def test_app_iter_range_with_one_segment(self):
-        listing = [{'name': 'o1', 'bytes': 1}]
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(None, None)), '1')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        self.assertEquals(''.join(segit.app_iter_range(3, None)), '')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        self.assertEquals(''.join(segit.app_iter_range(3, 5)), '')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(None, 5)), '1')
-
-    def test_app_iter_range_with_two_segments(self):
-        listing = [{'name': 'o1', 'bytes': 1}, {'name': 'o2', 'bytes': 2}]
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(None, None)), '122')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(1, None)), '22')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(1, 5)), '22')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(None, 2)), '12')
-
-    def test_app_iter_range_with_many_segments(self):
-        listing = [{'name': 'o1', 'bytes': 1}, {'name': 'o2', 'bytes': 2},
-                   {'name': 'o3', 'bytes': 3}, {'name': 'o4', 'bytes': 4},
-                   {'name': 'o5', 'bytes': 5}]
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(None, None)),
-                          '122333444455555')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(3, None)),
-                          '333444455555')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(5, None)), '3444455555')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(None, 6)), '122333')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(None, 7)), '1223334')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(3, 7)), '3334')
-
-        segit = SegmentedIterable(self.controller, 'lc', listing)
-        segit.response = Stub()
-        self.assertEquals(''.join(segit.app_iter_range(5, 7)), '34')
-
-
 class TestProxyObjectPerformance(unittest.TestCase):
 
     def setUp(self):