Add functional tests for Storage Policy

* additional container tests
 * refactor test cross policy copy
 * make functional tests cleanup better

In-process functional tests only define a single ring and will skip some of
the multi-storage policy tests, but have been updated to reload_policies with
the patched swift.conf.

DocImpact
Implements: blueprint storage-policies
Change-Id: If17bc7b9737558d3b9a54eeb6ff3e6b51463f002

swift/proxy/controllers/container.py

@@ -35,7 +35,7 @@ class ContainerController(Controller):
     # Ensure these are all lowercase
     pass_through_headers = ['x-container-read', 'x-container-write',
                             'x-container-sync-key', 'x-container-sync-to',
-                            'x-versions-location', POLICY_INDEX.lower()]
+                            'x-versions-location']
 
     def __init__(self, app, account_name, container_name, **kwargs):
         Controller.__init__(self, app)

test/functional/__init__.py

@@ -21,6 +21,7 @@ import locale
 import eventlet
 import eventlet.debug
 import functools
+import random
 from time import time, sleep
 from httplib import HTTPException
 from urlparse import urlparse
@@ -37,7 +38,7 @@ from test.functional.swift_test_client import Connection, ResponseError
 # on file systems that don't support extended attributes.
 from test.unit import debug_logger, FakeMemcache
 
-from swift.common import constraints, utils, ring
+from swift.common import constraints, utils, ring, storage_policy
 from swift.common.wsgi import monkey_patch_mimetools
 from swift.common.middleware import catch_errors, gatekeeper, healthcheck, \
     proxy_logging, container_sync, bulk, tempurl, slo, dlo, ratelimit, \
@@ -151,6 +152,8 @@ def in_process_setup(the_object_server=object_server):
     orig_swift_conf_name = utils.SWIFT_CONF_FILE
     utils.SWIFT_CONF_FILE = swift_conf
     constraints.reload_constraints()
+    storage_policy.SWIFT_CONF_FILE = swift_conf
+    storage_policy.reload_storage_policies()
     global config
     if constraints.SWIFT_CONSTRAINTS_LOADED:
         # Use the swift constraints that are loaded for the test framework
@@ -344,7 +347,7 @@ def get_cluster_info():
         # test.conf data
         pass
     else:
-        eff_constraints.update(cluster_info['swift'])
+        eff_constraints.update(cluster_info.get('swift', {}))
 
     # Finally, we'll allow any constraint present in the swift-constraints
     # section of test.conf to override everything. Note that only those
@@ -620,6 +623,18 @@ def load_constraint(name):
     return c
 
 
+def get_storage_policy_from_cluster_info(info):
+    policies = info['swift'].get('policies', {})
+    default_policy = []
+    non_default_policies = []
+    for p in policies:
+        if p.get('default', {}):
+            default_policy.append(p)
+        else:
+            non_default_policies.append(p)
+    return default_policy, non_default_policies
+
+
 def reset_acl():
     def post(url, token, parsed, conn):
         conn.request('POST', parsed.path, '', {
@@ -650,3 +665,65 @@ def requires_acls(f):
             reset_acl()
         return rv
     return wrapper
+
+
+class FunctionalStoragePolicyCollection(object):
+
+    def __init__(self, policies):
+        self._all = policies
+        self.default = None
+        for p in self:
+            if p.get('default', False):
+                assert self.default is None, 'Found multiple default ' \
+                    'policies %r and %r' % (self.default, p)
+                self.default = p
+
+    @classmethod
+    def from_info(cls, info=None):
+        if not (info or cluster_info):
+            get_cluster_info()
+        info = info or cluster_info
+        try:
+            policy_info = info['swift']['policies']
+        except KeyError:
+            raise AssertionError('Did not find any policy info in %r' % info)
+        policies = cls(policy_info)
+        assert policies.default, \
+            'Did not find default policy in %r' % policy_info
+        return policies
+
+    def __len__(self):
+        return len(self._all)
+
+    def __iter__(self):
+        return iter(self._all)
+
+    def __getitem__(self, index):
+        return self._all[index]
+
+    def filter(self, **kwargs):
+        return self.__class__([p for p in self if all(
+            p.get(k) == v for k, v in kwargs.items())])
+
+    def exclude(self, **kwargs):
+        return self.__class__([p for p in self if all(
+            p.get(k) != v for k, v in kwargs.items())])
+
+    def select(self):
+        return random.choice(self)
+
+
+def requires_policies(f):
+    @functools.wraps(f)
+    def wrapper(self, *args, **kwargs):
+        if skip:
+            raise SkipTest
+        try:
+            self.policies = FunctionalStoragePolicyCollection.from_info()
+        except AssertionError:
+            raise SkipTest("Unable to determine available policies")
+        if len(self.policies) < 2:
+            raise SkipTest("Multiple policies not enabled")
+        return f(self, *args, **kwargs)
+
+    return wrapper

test/functional/swift_test_client.py

@@ -186,7 +186,7 @@ class Connection(object):
         """
         status = self.make_request('GET', '/info',
                                    cfg={'absolute_path': True})
-        if status == 404:
+        if status // 100 == 4:
             return {}
         if not 200 <= status <= 299:
             raise ResponseError(self.response, 'GET', '/info')

test/functional/test_account.py

@@ -36,6 +36,36 @@ class TestAccount(unittest.TestCase):
         self.max_meta_overall_size = load_constraint('max_meta_overall_size')
         self.max_meta_value_length = load_constraint('max_meta_value_length')
 
+        def head(url, token, parsed, conn):
+            conn.request('HEAD', parsed.path, '', {'X-Auth-Token': token})
+            return check_response(conn)
+        resp = retry(head)
+        self.existing_metadata = set([
+            k for k, v in resp.getheaders() if
+            k.lower().startswith('x-account-meta')])
+
+    def tearDown(self):
+        def head(url, token, parsed, conn):
+            conn.request('HEAD', parsed.path, '', {'X-Auth-Token': token})
+            return check_response(conn)
+        resp = retry(head)
+        resp.read()
+        new_metadata = set(
+            [k for k, v in resp.getheaders() if
+             k.lower().startswith('x-account-meta')])
+
+        def clear_meta(url, token, parsed, conn, remove_metadata_keys):
+            headers = {'X-Auth-Token': token}
+            headers.update((k, '') for k in remove_metadata_keys)
+            conn.request('POST', parsed.path, '', headers)
+            return check_response(conn)
+        extra_metadata = list(self.existing_metadata ^ new_metadata)
+        for i in range(0, len(extra_metadata), 90):
+            batch = extra_metadata[i:i + 90]
+            resp = retry(clear_meta, batch)
+            resp.read()
+            self.assertEqual(resp.status // 100, 2)
+
     def test_metadata(self):
         if tf.skip:
             raise SkipTest

test/functional/test_container.py

@@ -21,7 +21,7 @@ from nose import SkipTest
 from uuid import uuid4
 
 from test.functional import check_response, retry, requires_acls, \
-    load_constraint
+    load_constraint, requires_policies
 import test.functional as tf
 
 
@@ -31,6 +31,8 @@ class TestContainer(unittest.TestCase):
         if tf.skip:
             raise SkipTest
         self.name = uuid4().hex
+        # this container isn't created by default, but will be cleaned up
+        self.container = uuid4().hex
 
         def put(url, token, parsed, conn):
             conn.request('PUT', parsed.path + '/' + self.name, '',
@@ -50,38 +52,47 @@ class TestContainer(unittest.TestCase):
         if tf.skip:
             raise SkipTest
 
-        def get(url, token, parsed, conn):
+        def get(url, token, parsed, conn, container):
-            conn.request('GET', parsed.path + '/' + self.name + '?format=json',
+            conn.request(
-                         '', {'X-Auth-Token': token})
+                'GET', parsed.path + '/' + container + '?format=json', '',
-            return check_response(conn)
-
-        def delete(url, token, parsed, conn, obj):
-            conn.request('DELETE',
-                         '/'.join([parsed.path, self.name, obj['name']]), '',
                 {'X-Auth-Token': token})
             return check_response(conn)
 
+        def delete(url, token, parsed, conn, container, obj):
+            conn.request(
+                'DELETE', '/'.join([parsed.path, container, obj['name']]), '',
+                {'X-Auth-Token': token})
+            return check_response(conn)
+
+        for container in (self.name, self.container):
             while True:
-            resp = retry(get)
+                resp = retry(get, container)
                 body = resp.read()
+                if resp.status == 404:
+                    break
                 self.assert_(resp.status // 100 == 2, resp.status)
                 objs = json.loads(body)
                 if not objs:
                     break
                 for obj in objs:
-                resp = retry(delete, obj)
+                    resp = retry(delete, container, obj)
                     resp.read()
                     self.assertEqual(resp.status, 204)
 
-        def delete(url, token, parsed, conn):
+        def delete(url, token, parsed, conn, container):
-            conn.request('DELETE', parsed.path + '/' + self.name, '',
+            conn.request('DELETE', parsed.path + '/' + container, '',
                          {'X-Auth-Token': token})
             return check_response(conn)
 
-        resp = retry(delete)
+        resp = retry(delete, self.name)
         resp.read()
         self.assertEqual(resp.status, 204)
 
+        # container may have not been created
+        resp = retry(delete, self.container)
+        resp.read()
+        self.assert_(resp.status in (204, 404))
+
     def test_multi_metadata(self):
         if tf.skip:
             raise SkipTest
@@ -1342,6 +1353,163 @@ class TestContainer(unittest.TestCase):
             self.assertEqual(resp.read(), 'Invalid UTF8 or contains NULL')
             self.assertEqual(resp.status, 412)
 
+    def test_create_container_gets_default_policy_by_default(self):
+        try:
+            default_policy = \
+                tf.FunctionalStoragePolicyCollection.from_info().default
+        except AssertionError:
+            raise SkipTest()
+
+        def put(url, token, parsed, conn):
+            conn.request('PUT', parsed.path + '/' + self.container, '',
+                         {'X-Auth-Token': token})
+            return check_response(conn)
+        resp = retry(put)
+        resp.read()
+        self.assertEqual(resp.status // 100, 2)
+
+        def head(url, token, parsed, conn):
+            conn.request('HEAD', parsed.path + '/' + self.container, '',
+                         {'X-Auth-Token': token})
+            return check_response(conn)
+        resp = retry(head)
+        resp.read()
+        headers = dict((k.lower(), v) for k, v in resp.getheaders())
+        self.assertEquals(headers.get('x-storage-policy'),
+                          default_policy['name'])
+
+    def test_error_invalid_storage_policy_name(self):
+        def put(url, token, parsed, conn, headers):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('PUT', parsed.path + '/' + self.container, '',
+                         new_headers)
+            return check_response(conn)
+
+        # create
+        resp = retry(put, {'X-Storage-Policy': uuid4().hex})
+        resp.read()
+        self.assertEqual(resp.status, 400)
+
+    @requires_policies
+    def test_create_non_default_storage_policy_container(self):
+        policy = self.policies.exclude(default=True).select()
+
+        def put(url, token, parsed, conn, headers=None):
+            base_headers = {'X-Auth-Token': token}
+            if headers:
+                base_headers.update(headers)
+            conn.request('PUT', parsed.path + '/' + self.container, '',
+                         base_headers)
+            return check_response(conn)
+        headers = {'X-Storage-Policy': policy['name']}
+        resp = retry(put, headers=headers)
+        resp.read()
+        self.assertEqual(resp.status, 201)
+
+        def head(url, token, parsed, conn):
+            conn.request('HEAD', parsed.path + '/' + self.container, '',
+                         {'X-Auth-Token': token})
+            return check_response(conn)
+        resp = retry(head)
+        resp.read()
+        headers = dict((k.lower(), v) for k, v in resp.getheaders())
+        self.assertEquals(headers.get('x-storage-policy'),
+                          policy['name'])
+
+        # and test recreate with-out specifiying Storage Policy
+        resp = retry(put)
+        resp.read()
+        self.assertEqual(resp.status, 202)
+        # should still be original storage policy
+        resp = retry(head)
+        resp.read()
+        headers = dict((k.lower(), v) for k, v in resp.getheaders())
+        self.assertEquals(headers.get('x-storage-policy'),
+                          policy['name'])
+
+        # delete it
+        def delete(url, token, parsed, conn):
+            conn.request('DELETE', parsed.path + '/' + self.container, '',
+                         {'X-Auth-Token': token})
+            return check_response(conn)
+        resp = retry(delete)
+        resp.read()
+        self.assertEqual(resp.status, 204)
+
+        # verify no policy header
+        resp = retry(head)
+        resp.read()
+        headers = dict((k.lower(), v) for k, v in resp.getheaders())
+        self.assertEquals(headers.get('x-storage-policy'), None)
+
+    @requires_policies
+    def test_conflict_change_storage_policy_with_put(self):
+        def put(url, token, parsed, conn, headers):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('PUT', parsed.path + '/' + self.container, '',
+                         new_headers)
+            return check_response(conn)
+
+        # create
+        policy = self.policies.select()
+        resp = retry(put, {'X-Storage-Policy': policy['name']})
+        resp.read()
+        self.assertEqual(resp.status, 201)
+
+        # can't change it
+        other_policy = self.policies.exclude(name=policy['name']).select()
+        resp = retry(put, {'X-Storage-Policy': other_policy['name']})
+        resp.read()
+        self.assertEqual(resp.status, 409)
+
+        def head(url, token, parsed, conn):
+            conn.request('HEAD', parsed.path + '/' + self.container, '',
+                         {'X-Auth-Token': token})
+            return check_response(conn)
+        # still original policy
+        resp = retry(head)
+        resp.read()
+        headers = dict((k.lower(), v) for k, v in resp.getheaders())
+        self.assertEquals(headers.get('x-storage-policy'),
+                          policy['name'])
+
+    @requires_policies
+    def test_noop_change_storage_policy_with_post(self):
+        def put(url, token, parsed, conn, headers):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('PUT', parsed.path + '/' + self.container, '',
+                         new_headers)
+            return check_response(conn)
+
+        # create
+        policy = self.policies.select()
+        resp = retry(put, {'X-Storage-Policy': policy['name']})
+        resp.read()
+        self.assertEqual(resp.status, 201)
+
+        def post(url, token, parsed, conn, headers):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('POST', parsed.path + '/' + self.container, '',
+                         new_headers)
+            return check_response(conn)
+        # attempt update
+        for header in ('X-Storage-Policy', 'X-Storage-Policy-Index'):
+            other_policy = self.policies.exclude(name=policy['name']).select()
+            resp = retry(post, {header: other_policy['name']})
+            resp.read()
+            self.assertEqual(resp.status, 204)
+
+        def head(url, token, parsed, conn):
+            conn.request('HEAD', parsed.path + '/' + self.container, '',
+                         {'X-Auth-Token': token})
+            return check_response(conn)
+        # still original policy
+        resp = retry(head)
+        resp.read()
+        headers = dict((k.lower(), v) for k, v in resp.getheaders())
+        self.assertEquals(headers.get('x-storage-policy'),
+                          policy['name'])
+
 
 if __name__ == '__main__':
     unittest.main()

test/functional/test_object.py

@@ -21,7 +21,8 @@ from uuid import uuid4
 
 from swift.common.utils import json
 
-from test.functional import check_response, retry, requires_acls
+from test.functional import check_response, retry, requires_acls, \
+    requires_policies
 import test.functional as tf
 
 
@@ -32,13 +33,9 @@ class TestObject(unittest.TestCase):
             raise SkipTest
         self.container = uuid4().hex
 
-        def put(url, token, parsed, conn):
+        self.containers = []
-            conn.request('PUT', parsed.path + '/' + self.container, '',
+        self._create_container(self.container)
-                         {'X-Auth-Token': token})
+
-            return check_response(conn)
-        resp = retry(put)
-        resp.read()
-        self.assertEqual(resp.status, 201)
         self.obj = uuid4().hex
 
         def put(url, token, parsed, conn):
@@ -50,40 +47,65 @@ class TestObject(unittest.TestCase):
         resp.read()
         self.assertEqual(resp.status, 201)
 
+    def _create_container(self, name=None, headers=None):
+        if not name:
+            name = uuid4().hex
+        self.containers.append(name)
+        headers = headers or {}
+
+        def put(url, token, parsed, conn, name):
+            new_headers = dict({'X-Auth-Token': token}, **headers)
+            conn.request('PUT', parsed.path + '/' + name, '',
+                         new_headers)
+            return check_response(conn)
+        resp = retry(put, name)
+        resp.read()
+        self.assertEqual(resp.status, 201)
+        return name
+
     def tearDown(self):
         if tf.skip:
             raise SkipTest
 
-        def delete(url, token, parsed, conn, obj):
-            conn.request('DELETE',
-                         '%s/%s/%s' % (parsed.path, self.container, obj),
-                         '', {'X-Auth-Token': token})
-            return check_response(conn)
-
         # get list of objects in container
-        def list(url, token, parsed, conn):
+        def get(url, token, parsed, conn, container):
-            conn.request('GET',
+            conn.request(
-                         '%s/%s' % (parsed.path, self.container),
+                'GET', parsed.path + '/' + container + '?format=json', '',
-                         '', {'X-Auth-Token': token})
+                {'X-Auth-Token': token})
             return check_response(conn)
-        resp = retry(list)
-        object_listing = resp.read()
-        self.assertEqual(resp.status, 200)
 
-        # iterate over object listing and delete all objects
+        # delete an object
-        for obj in object_listing.splitlines():
+        def delete(url, token, parsed, conn, container, obj):
-            resp = retry(delete, obj)
+            conn.request(
+                'DELETE', '/'.join([parsed.path, container, obj['name']]), '',
+                {'X-Auth-Token': token})
+            return check_response(conn)
+
+        for container in self.containers:
+            while True:
+                resp = retry(get, container)
+                body = resp.read()
+                if resp.status == 404:
+                    break
+                self.assert_(resp.status // 100 == 2, resp.status)
+                objs = json.loads(body)
+                if not objs:
+                    break
+                for obj in objs:
+                    resp = retry(delete, container, obj)
                     resp.read()
                     self.assertEqual(resp.status, 204)
 
         # delete the container
-        def delete(url, token, parsed, conn):
+        def delete(url, token, parsed, conn, name):
-            conn.request('DELETE', parsed.path + '/' + self.container, '',
+            conn.request('DELETE', parsed.path + '/' + name, '',
                          {'X-Auth-Token': token})
             return check_response(conn)
-        resp = retry(delete)
+
+        for container in self.containers:
+            resp = retry(delete, container)
             resp.read()
-        self.assertEqual(resp.status, 204)
+            self.assert_(resp.status in (204, 404))
 
     def test_if_none_match(self):
         def put(url, token, parsed, conn):
@@ -996,6 +1018,64 @@ class TestObject(unittest.TestCase):
             self.assertEquals(headers.get('access-control-allow-origin'),
                               'http://m.com')
 
+    @requires_policies
+    def test_cross_policy_copy(self):
+        # create container in first policy
+        policy = self.policies.select()
+        container = self._create_container(
+            headers={'X-Storage-Policy': policy['name']})
+        obj = uuid4().hex
+
+        # create a container in second policy
+        other_policy = self.policies.exclude(name=policy['name']).select()
+        other_container = self._create_container(
+            headers={'X-Storage-Policy': other_policy['name']})
+        other_obj = uuid4().hex
+
+        def put_obj(url, token, parsed, conn, container, obj):
+            # to keep track of things, use the original path as the body
+            content = '%s/%s' % (container, obj)
+            path = '%s/%s' % (parsed.path, content)
+            conn.request('PUT', path, content, {'X-Auth-Token': token})
+            return check_response(conn)
+
+        # create objects
+        for c, o in zip((container, other_container), (obj, other_obj)):
+            resp = retry(put_obj, c, o)
+            resp.read()
+            self.assertEqual(resp.status, 201)
+
+        def put_copy_from(url, token, parsed, conn, container, obj, source):
+            dest_path = '%s/%s/%s' % (parsed.path, container, obj)
+            conn.request('PUT', dest_path, '',
+                         {'X-Auth-Token': token,
+                          'Content-Length': '0',
+                          'X-Copy-From': source})
+            return check_response(conn)
+
+        copy_requests = (
+            (container, other_obj, '%s/%s' % (other_container, other_obj)),
+            (other_container, obj, '%s/%s' % (container, obj)),
+        )
+
+        # copy objects
+        for c, o, source in copy_requests:
+            resp = retry(put_copy_from, c, o, source)
+            resp.read()
+            self.assertEqual(resp.status, 201)
+
+        def get_obj(url, token, parsed, conn, container, obj):
+            path = '%s/%s/%s' % (parsed.path, container, obj)
+            conn.request('GET', path, '', {'X-Auth-Token': token})
+            return check_response(conn)
+
+        # validate contents, contents should be source
+        validate_requests = copy_requests
+        for c, o, body in validate_requests:
+            resp = retry(get_obj, c, o)
+            self.assertEqual(resp.status, 200)
+            self.assertEqual(body, resp.read())
+
 
 if __name__ == '__main__':
     unittest.main()

test/functional/tests.py

@@ -28,6 +28,8 @@ import uuid
 import eventlet
 from nose import SkipTest
 
+from swift.common.storage_policy import POLICY
+
 from test.functional import normalized_urls, load_constraint, cluster_info
 import test.functional as tf
 from test.functional.swift_test_client import Account, Connection, File, \
@@ -2077,6 +2079,61 @@ class TestObjectVersioningEnv(object):
         cls.versioning_enabled = 'versions' in container_info
 
 
+class TestCrossPolicyObjectVersioningEnv(object):
+    # tri-state: None initially, then True/False
+    versioning_enabled = None
+    multiple_policies_enabled = None
+    policies = None
+
+    @classmethod
+    def setUp(cls):
+        cls.conn = Connection(tf.config)
+        cls.conn.authenticate()
+
+        if cls.multiple_policies_enabled is None:
+            try:
+                cls.policies = tf.FunctionalStoragePolicyCollection.from_info()
+            except AssertionError:
+                pass
+
+        if cls.policies and len(cls.policies) > 1:
+            cls.multiple_policies_enabled = True
+        else:
+            cls.multiple_policies_enabled = False
+            # We have to lie here that versioning is enabled. We actually
+            # don't know, but it does not matter. We know these tests cannot
+            # run without multiple policies present. If multiple policies are
+            # present, we won't be setting this field to any value, so it
+            # should all still work.
+            cls.versioning_enabled = True
+            return
+
+        policy = cls.policies.select()
+        version_policy = cls.policies.exclude(name=policy['name']).select()
+
+        cls.account = Account(cls.conn, tf.config.get('account',
+                                                      tf.config['username']))
+
+        # avoid getting a prefix that stops halfway through an encoded
+        # character
+        prefix = Utils.create_name().decode("utf-8")[:10].encode("utf-8")
+
+        cls.versions_container = cls.account.container(prefix + "-versions")
+        if not cls.versions_container.create(
+                {POLICY: policy['name']}):
+            raise ResponseError(cls.conn.response)
+
+        cls.container = cls.account.container(prefix + "-objs")
+        if not cls.container.create(
+                hdrs={'X-Versions-Location': cls.versions_container.name,
+                      POLICY: version_policy['name']}):
+            raise ResponseError(cls.conn.response)
+
+        container_info = cls.container.info()
+        # if versioning is off, then X-Versions-Location won't persist
+        cls.versioning_enabled = 'versions' in container_info
+
+
 class TestObjectVersioning(Base):
     env = TestObjectVersioningEnv
     set_up = False
@@ -2127,6 +2184,21 @@ class TestObjectVersioningUTF8(Base2, TestObjectVersioning):
     set_up = False
 
 
+class TestCrossPolicyObjectVersioning(TestObjectVersioning):
+    env = TestCrossPolicyObjectVersioningEnv
+    set_up = False
+
+    def setUp(self):
+        super(TestCrossPolicyObjectVersioning, self).setUp()
+        if self.env.multiple_policies_enabled is False:
+            raise SkipTest('Cross policy test requires multiple policies')
+        elif self.env.multiple_policies_enabled is not True:
+            # just some sanity checking
+            raise Exception("Expected multiple_policies_enabled "
+                            "to be True/False, got %r" % (
+                                self.env.versioning_enabled,))
+
+
 class TestTempurlEnv(object):
     tempurl_enabled = None  # tri-state: None initially, then True/False