#!/usr/bin/python
# Copyright (c) 2010-2015 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 hmac
import unittest
import itertools
import hashlib
import six
import time
from six.moves import urllib
from uuid import uuid4
from swift.common.http import is_success
from swift.common.utils import json, MD5_OF_EMPTY_STRING
from swift.common.middleware.slo import SloGetContext
from test.functional import check_response, retry, requires_acls, \
cluster_info, SkipTest
from test.functional.tests import Base, TestFileComparisonEnv, Utils, BaseEnv
from test.functional.test_slo import TestSloEnv
from test.functional.test_dlo import TestDloEnv
from test.functional.test_tempurl import TestContainerTempurlEnv, \
TestTempurlEnv
from test.functional.swift_test_client import ResponseError
import test.functional as tf
from test.unit import group_by_byte
TARGET_BODY = b'target body'
def setUpModule():
tf.setup_package()
if 'symlink' not in cluster_info:
raise SkipTest("Symlinks not enabled")
def tearDownModule():
tf.teardown_package()
class TestSymlinkEnv(BaseEnv):
link_cont = uuid4().hex
tgt_cont = uuid4().hex
tgt_obj = uuid4().hex
@classmethod
def setUp(cls):
if tf.skip or tf.skip2:
raise SkipTest
cls._create_container(cls.tgt_cont) # use_account=1
cls._create_container(cls.link_cont) # use_account=1
# container in account 2
cls._create_container(cls.link_cont, use_account=2)
cls._create_tgt_object()
@classmethod
def containers(cls):
return (cls.link_cont, cls.tgt_cont)
@classmethod
def target_content_location(cls, override_obj=None, override_account=None):
account = override_account or tf.parsed[0].path.split('/', 2)[2]
return '/v1/%s/%s/%s' % (account, cls.tgt_cont,
override_obj or cls.tgt_obj)
@classmethod
def _make_request(cls, url, token, parsed, conn, method,
container, obj='', headers=None, body=b'',
query_args=None):
headers = headers or {}
headers.update({'X-Auth-Token': token})
path = '%s/%s/%s' % (parsed.path, container, obj) if obj \
else '%s/%s' % (parsed.path, container)
if query_args:
path += '?%s' % query_args
conn.request(method, path, body, headers)
resp = check_response(conn)
# to read the buffer and keep it in the attribute, call resp.content
resp.content
return resp
@classmethod
def _create_container(cls, name, headers=None, use_account=1):
headers = headers or {}
resp = retry(cls._make_request, method='PUT', container=name,
headers=headers, use_account=use_account)
if resp.status not in (201, 202):
raise ResponseError(resp)
return name
@classmethod
def _create_tgt_object(cls, body=TARGET_BODY):
resp = retry(cls._make_request, method='PUT',
headers={'Content-Type': 'application/target'},
container=cls.tgt_cont, obj=cls.tgt_obj,
body=body)
if resp.status != 201:
raise ResponseError(resp)
# sanity: successful put response has content-length 0
cls.tgt_length = str(len(body))
cls.tgt_etag = resp.getheader('etag')
resp = retry(cls._make_request, method='GET',
container=cls.tgt_cont, obj=cls.tgt_obj)
if resp.status != 200 and resp.content != body:
raise ResponseError(resp)
@classmethod
def tearDown(cls):
delete_containers = [
(use_account, containers) for use_account, containers in
enumerate([cls.containers(), [cls.link_cont]], 1)]
# delete objects inside container
for use_account, containers in delete_containers:
if use_account == 2 and tf.skip2:
continue
for container in containers:
while True:
cont = container
resp = retry(cls._make_request, method='GET',
container=cont, query_args='format=json',
use_account=use_account)
if resp.status == 404:
break
if not is_success(resp.status):
raise ResponseError(resp)
objs = json.loads(resp.content)
if not objs:
break
for obj in objs:
resp = retry(cls._make_request, method='DELETE',
container=container, obj=obj['name'],
use_account=use_account)
if resp.status not in (204, 404):
raise ResponseError(resp)
# delete the containers
for use_account, containers in delete_containers:
if use_account == 2 and tf.skip2:
continue
for container in containers:
resp = retry(cls._make_request, method='DELETE',
container=container,
use_account=use_account)
if resp.status not in (204, 404):
raise ResponseError(resp)
class TestSymlink(Base):
env = TestSymlinkEnv
@classmethod
def setUpClass(cls):
# To skip env setup for class setup, instead setUp the env for each
# test method
pass
def setUp(self):
self.env.setUp()
def object_name_generator():
while True:
yield uuid4().hex
self.obj_name_gen = object_name_generator()
self._account_name = None
def tearDown(self):
self.env.tearDown()
@property
def account_name(self):
if not self._account_name:
self._account_name = tf.parsed[0].path.split('/', 2)[2]
return self._account_name
def _make_request(self, url, token, parsed, conn, method,
container, obj='', headers=None, body=b'',
query_args=None, allow_redirects=True):
headers = headers or {}
headers.update({'X-Auth-Token': token})
path = '%s/%s/%s' % (parsed.path, container, obj) if obj \
else '%s/%s' % (parsed.path, container)
if query_args:
path += '?%s' % query_args
conn.requests_args['allow_redirects'] = allow_redirects
conn.request(method, path, body, headers)
resp = check_response(conn)
# to read the buffer and keep it in the attribute, call resp.content
resp.content
return resp
def _make_request_with_symlink_get(self, url, token, parsed, conn, method,
container, obj, headers=None, body=b''):
resp = self._make_request(
url, token, parsed, conn, method, container, obj, headers, body,
query_args='symlink=get')
return resp
def _test_put_symlink(self, link_cont, link_obj, tgt_cont, tgt_obj):
headers = {'X-Symlink-Target': '%s/%s' % (tgt_cont, tgt_obj)}
resp = retry(self._make_request, method='PUT',
container=link_cont, obj=link_obj,
headers=headers)
self.assertEqual(resp.status, 201)
def _test_put_symlink_with_etag(self, link_cont, link_obj, tgt_cont,
tgt_obj, etag, headers=None):
headers = headers or {}
headers.update({'X-Symlink-Target': '%s/%s' % (tgt_cont, tgt_obj),
'X-Symlink-Target-Etag': etag})
resp = retry(self._make_request, method='PUT',
container=link_cont, obj=link_obj,
headers=headers)
self.assertEqual(resp.status, 201, resp.content)
def _test_get_as_target_object(
self, link_cont, link_obj, expected_content_location,
use_account=1):
resp = retry(
self._make_request, method='GET',
container=link_cont, obj=link_obj, use_account=use_account)
self.assertEqual(resp.status, 200, resp.content)
self.assertEqual(resp.content, TARGET_BODY)
self.assertEqual(resp.getheader('content-length'),
str(self.env.tgt_length))
self.assertEqual(resp.getheader('etag'), self.env.tgt_etag)
self.assertIn('Content-Location', resp.headers)
self.assertEqual(expected_content_location,
resp.getheader('content-location'))
return resp
def _test_head_as_target_object(self, link_cont, link_obj, use_account=1):
resp = retry(
self._make_request, method='HEAD',
container=link_cont, obj=link_obj, use_account=use_account)
self.assertEqual(resp.status, 200)
def _assertLinkObject(self, link_cont, link_obj, use_account=1):
# HEAD on link_obj itself
resp = retry(
self._make_request_with_symlink_get, method='HEAD',
container=link_cont, obj=link_obj, use_account=use_account)
self.assertEqual(resp.status, 200)
self.assertTrue(resp.getheader('x-symlink-target'))
# GET on link_obj itself
resp = retry(
self._make_request_with_symlink_get, method='GET',
container=link_cont, obj=link_obj, use_account=use_account)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, b'')
self.assertEqual(resp.getheader('content-length'), str(0))
self.assertTrue(resp.getheader('x-symlink-target'))
def _assertSymlink(self, link_cont, link_obj,
expected_content_location=None, use_account=1):
expected_content_location = \
expected_content_location or self.env.target_content_location()
# sanity: HEAD/GET on link_obj
self._assertLinkObject(link_cont, link_obj, use_account)
# HEAD target object via symlink
self._test_head_as_target_object(
link_cont=link_cont, link_obj=link_obj, use_account=use_account)
# GET target object via symlink
self._test_get_as_target_object(
link_cont=link_cont, link_obj=link_obj, use_account=use_account,
expected_content_location=expected_content_location)
def test_symlink_with_encoded_target_name(self):
# makes sure to test encoded characters as symlink target
target_obj = 'dealde%2Fl04 011e%204c8df/flash.png'
link_obj = uuid4().hex
# create target using unnormalized path
resp = retry(
self._make_request, method='PUT', container=self.env.tgt_cont,
obj=target_obj, body=TARGET_BODY)
self.assertEqual(resp.status, 201)
# you can get it using either name
resp = retry(
self._make_request, method='GET', container=self.env.tgt_cont,
obj=target_obj)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, TARGET_BODY)
normalized_quoted_obj = 'dealde/l04%20011e%204c8df/flash.png'
self.assertEqual(normalized_quoted_obj, urllib.parse.quote(
urllib.parse.unquote(target_obj)))
resp = retry(
self._make_request, method='GET', container=self.env.tgt_cont,
obj=normalized_quoted_obj)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, TARGET_BODY)
# create a symlink using the un-normalized target path
self._test_put_symlink(link_cont=self.env.link_cont, link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=target_obj)
# and it's normalized
self._assertSymlink(
self.env.link_cont, link_obj,
expected_content_location=self.env.target_content_location(
normalized_quoted_obj))
# create a symlink using the normalized target path
self._test_put_symlink(link_cont=self.env.link_cont, link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=normalized_quoted_obj)
# and it's ALSO normalized
self._assertSymlink(
self.env.link_cont, link_obj,
expected_content_location=self.env.target_content_location(
normalized_quoted_obj))
def test_symlink_put_head_get(self):
link_obj = uuid4().hex
# PUT link_obj
self._test_put_symlink(link_cont=self.env.link_cont, link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
self._assertSymlink(self.env.link_cont, link_obj)
def test_symlink_with_etag_put_head_get(self):
link_obj = uuid4().hex
# PUT link_obj
self._test_put_symlink_with_etag(link_cont=self.env.link_cont,
link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj,
etag=self.env.tgt_etag)
self._assertSymlink(self.env.link_cont, link_obj)
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj,
headers={'If-Match': self.env.tgt_etag})
self.assertEqual(resp.status, 200)
self.assertEqual(resp.getheader('content-location'),
self.env.target_content_location())
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj,
headers={'If-Match': 'not-the-etag'})
self.assertEqual(resp.status, 412)
self.assertEqual(resp.getheader('content-location'),
self.env.target_content_location())
def test_static_symlink_with_bad_etag_put_head_get(self):
link_obj = uuid4().hex
# PUT link_obj
self._test_put_symlink_with_etag(link_cont=self.env.link_cont,
link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj,
etag=self.env.tgt_etag)
# overwrite tgt object
self.env._create_tgt_object(body='updated target body')
resp = retry(
self._make_request, method='HEAD',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 409)
# but we still know where it points
self.assertEqual(resp.getheader('content-location'),
self.env.target_content_location())
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 409)
self.assertEqual(resp.getheader('content-location'),
self.env.target_content_location())
# uses a mechanism entirely divorced from if-match
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj,
headers={'If-Match': self.env.tgt_etag})
self.assertEqual(resp.status, 409)
self.assertEqual(resp.getheader('content-location'),
self.env.target_content_location())
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj,
headers={'If-Match': 'not-the-etag'})
self.assertEqual(resp.status, 409)
self.assertEqual(resp.getheader('content-location'),
self.env.target_content_location())
resp = retry(
self._make_request, method='DELETE',
container=self.env.tgt_cont, obj=self.env.tgt_obj)
# not-found-ness trumps if-match-ness
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 404)
self.assertEqual(resp.getheader('content-location'),
self.env.target_content_location())
def test_dynamic_link_to_static_link(self):
static_link_obj = uuid4().hex
# PUT static_link to tgt_obj
self._test_put_symlink_with_etag(link_cont=self.env.link_cont,
link_obj=static_link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj,
etag=self.env.tgt_etag)
symlink_obj = uuid4().hex
# PUT symlink to static_link
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=symlink_obj,
tgt_cont=self.env.link_cont,
tgt_obj=static_link_obj)
self._test_get_as_target_object(
link_cont=self.env.link_cont, link_obj=symlink_obj,
expected_content_location=self.env.target_content_location())
def test_static_link_to_dynamic_link(self):
symlink_obj = uuid4().hex
# PUT symlink to tgt_obj
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=symlink_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
static_link_obj = uuid4().hex
# PUT a static_link to the symlink
self._test_put_symlink_with_etag(link_cont=self.env.link_cont,
link_obj=static_link_obj,
tgt_cont=self.env.link_cont,
tgt_obj=symlink_obj,
etag=MD5_OF_EMPTY_STRING)
self._test_get_as_target_object(
link_cont=self.env.link_cont, link_obj=static_link_obj,
expected_content_location=self.env.target_content_location())
def test_static_link_to_nowhere(self):
missing_obj = uuid4().hex
static_link_obj = uuid4().hex
# PUT a static_link to the missing name
headers = {
'X-Symlink-Target': '%s/%s' % (self.env.link_cont, missing_obj),
'X-Symlink-Target-Etag': MD5_OF_EMPTY_STRING}
resp = retry(self._make_request, method='PUT',
container=self.env.link_cont, obj=static_link_obj,
headers=headers)
self.assertEqual(resp.status, 409)
self.assertEqual(resp.content, b'X-Symlink-Target does not exist')
def test_static_link_to_broken_symlink(self):
symlink_obj = uuid4().hex
# PUT symlink to tgt_obj
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=symlink_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
static_link_obj = uuid4().hex
# PUT a static_link to the symlink
self._test_put_symlink_with_etag(link_cont=self.env.link_cont,
link_obj=static_link_obj,
tgt_cont=self.env.link_cont,
tgt_obj=symlink_obj,
etag=MD5_OF_EMPTY_STRING)
# break the symlink
resp = retry(
self._make_request, method='DELETE',
container=self.env.tgt_cont, obj=self.env.tgt_obj)
self.assertEqual(resp.status // 100, 2)
# sanity
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=symlink_obj)
self.assertEqual(resp.status, 404)
# static_link is broken too!
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=static_link_obj)
self.assertEqual(resp.status, 404)
# interestingly you may create a static_link to a broken symlink
broken_static_link_obj = uuid4().hex
# PUT a static_link to the broken symlink
self._test_put_symlink_with_etag(link_cont=self.env.link_cont,
link_obj=broken_static_link_obj,
tgt_cont=self.env.link_cont,
tgt_obj=symlink_obj,
etag=MD5_OF_EMPTY_STRING)
def test_symlink_get_ranged(self):
link_obj = uuid4().hex
# PUT symlink
self._test_put_symlink(link_cont=self.env.link_cont, link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
headers = {'Range': 'bytes=7-10'}
resp = retry(self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj,
headers=headers)
self.assertEqual(resp.status, 206)
self.assertEqual(resp.content, b'body')
def test_create_symlink_before_target(self):
link_obj = uuid4().hex
target_obj = uuid4().hex
# PUT link_obj before target object is written
# PUT, GET, HEAD (on symlink) should all work ok without target object
self._test_put_symlink(link_cont=self.env.link_cont, link_obj=link_obj,
tgt_cont=self.env.tgt_cont, tgt_obj=target_obj)
# Try to GET target via symlink.
# 404 will be returned with Content-Location of target path.
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj, use_account=1)
self.assertEqual(resp.status, 404)
self.assertIn('Content-Location', resp.headers)
self.assertEqual(self.env.target_content_location(target_obj),
resp.getheader('content-location'))
# HEAD on target object via symlink should return a 404 since target
# object has not yet been written
resp = retry(
self._make_request, method='HEAD',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 404)
# GET on target object directly
resp = retry(
self._make_request, method='GET',
container=self.env.tgt_cont, obj=target_obj)
self.assertEqual(resp.status, 404)
# Now let's write target object and symlink will be able to return
# object
resp = retry(
self._make_request, method='PUT', container=self.env.tgt_cont,
obj=target_obj, body=TARGET_BODY)
self.assertEqual(resp.status, 201)
# successful put response has content-length 0
target_length = str(len(TARGET_BODY))
target_etag = resp.getheader('etag')
# sanity: HEAD/GET on link_obj itself
self._assertLinkObject(self.env.link_cont, link_obj)
# HEAD target object via symlink
self._test_head_as_target_object(
link_cont=self.env.link_cont, link_obj=link_obj)
# GET target object via symlink
resp = retry(self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, TARGET_BODY)
self.assertEqual(resp.getheader('content-length'), str(target_length))
self.assertEqual(resp.getheader('etag'), target_etag)
self.assertIn('Content-Location', resp.headers)
self.assertEqual(self.env.target_content_location(target_obj),
resp.getheader('content-location'))
def test_symlink_chain(self):
# Testing to symlink chain like symlink -> symlink -> target.
symloop_max = cluster_info['symlink']['symloop_max']
# create symlink chain in a container. To simplify,
# use target container for all objects (symlinks and target) here
previous = self.env.tgt_obj
container = self.env.tgt_cont
for link_obj in itertools.islice(self.obj_name_gen, symloop_max):
# PUT link_obj point to tgt_obj
self._test_put_symlink(
link_cont=container, link_obj=link_obj,
tgt_cont=container, tgt_obj=previous)
# set corrent link_obj to previous
previous = link_obj
# the last link is valid for symloop_max constraint
max_chain_link = link_obj
self._assertSymlink(link_cont=container, link_obj=max_chain_link)
# PUT a new link_obj points to the max_chain_link
# that will result in 409 error on the HEAD/GET.
too_many_chain_link = next(self.obj_name_gen)
self._test_put_symlink(
link_cont=container, link_obj=too_many_chain_link,
tgt_cont=container, tgt_obj=max_chain_link)
# try to HEAD to target object via too_many_chain_link
resp = retry(self._make_request, method='HEAD',
container=container,
obj=too_many_chain_link)
self.assertEqual(resp.status, 409)
self.assertEqual(resp.content, b'')
# try to GET to target object via too_many_chain_link
resp = retry(self._make_request, method='GET',
container=container,
obj=too_many_chain_link)
self.assertEqual(resp.status, 409)
self.assertEqual(
resp.content,
b'Too many levels of symbolic links, maximum allowed is %d' %
symloop_max)
# However, HEAD/GET to the (just) link is still ok
self._assertLinkObject(container, too_many_chain_link)
def test_symlink_chain_with_etag(self):
# Testing to symlink chain like symlink -> symlink -> target.
symloop_max = cluster_info['symlink']['symloop_max']
# create symlink chain in a container. To simplify,
# use target container for all objects (symlinks and target) here
previous = self.env.tgt_obj
container = self.env.tgt_cont
for link_obj in itertools.islice(self.obj_name_gen, symloop_max):
# PUT link_obj point to tgt_obj
self._test_put_symlink_with_etag(link_cont=container,
link_obj=link_obj,
tgt_cont=container,
tgt_obj=previous,
etag=self.env.tgt_etag)
# set current link_obj to previous
previous = link_obj
# the last link is valid for symloop_max constraint
max_chain_link = link_obj
self._assertSymlink(link_cont=container, link_obj=max_chain_link)
# chained etag validation works as long as the target symlink works
headers = {'X-Symlink-Target': '%s/%s' % (container, max_chain_link),
'X-Symlink-Target-Etag': 'not-the-real-etag'}
resp = retry(self._make_request, method='PUT',
container=container, obj=uuid4().hex,
headers=headers)
self.assertEqual(resp.status, 409)
# PUT a new link_obj pointing to the max_chain_link can validate the
# ETag but will result in 409 error on the HEAD/GET.
too_many_chain_link = next(self.obj_name_gen)
self._test_put_symlink_with_etag(
link_cont=container, link_obj=too_many_chain_link,
tgt_cont=container, tgt_obj=max_chain_link,
etag=self.env.tgt_etag)
# try to HEAD to target object via too_many_chain_link
resp = retry(self._make_request, method='HEAD',
container=container,
obj=too_many_chain_link)
self.assertEqual(resp.status, 409)
self.assertEqual(resp.content, b'')
# try to GET to target object via too_many_chain_link
resp = retry(self._make_request, method='GET',
container=container,
obj=too_many_chain_link)
self.assertEqual(resp.status, 409)
self.assertEqual(
resp.content,
b'Too many levels of symbolic links, maximum allowed is %d' %
symloop_max)
# However, HEAD/GET to the (just) link is still ok
self._assertLinkObject(container, too_many_chain_link)
def test_symlink_and_slo_manifest_chain(self):
if 'slo' not in cluster_info:
raise SkipTest
symloop_max = cluster_info['symlink']['symloop_max']
# create symlink chain in a container. To simplify,
# use target container for all objects (symlinks and target) here
previous = self.env.tgt_obj
container = self.env.tgt_cont
# make symlink and slo manifest chain
# e.g. slo -> symlink -> symlink -> slo -> symlink -> symlink -> target
for _ in range(SloGetContext.max_slo_recursion_depth or 1):
for link_obj in itertools.islice(self.obj_name_gen, symloop_max):
# PUT link_obj point to previous object
self._test_put_symlink(
link_cont=container, link_obj=link_obj,
tgt_cont=container, tgt_obj=previous)
# next link will point to this link
previous = link_obj
else:
# PUT a manifest with single segment to the symlink
manifest_obj = next(self.obj_name_gen)
manifest = json.dumps(
[{'path': '/%s/%s' % (container, link_obj)}])
resp = retry(self._make_request, method='PUT',
container=container, obj=manifest_obj,
body=manifest,
query_args='multipart-manifest=put')
self.assertEqual(resp.status, 201) # sanity
previous = manifest_obj
# From the last manifest to the final target obj length is
# symloop_max * max_slo_recursion_depth
max_recursion_manifest = previous
# Check GET to max_recursion_manifest returns valid target object
resp = retry(
self._make_request, method='GET', container=container,
obj=max_recursion_manifest)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, TARGET_BODY)
self.assertEqual(resp.getheader('content-length'),
str(self.env.tgt_length))
# N.B. since the last manifest is slo so it will remove
# content-location info from the response header
self.assertNotIn('Content-Location', resp.headers)
# sanity: one more link to the slo can work still
one_more_link = next(self.obj_name_gen)
self._test_put_symlink(
link_cont=container, link_obj=one_more_link,
tgt_cont=container, tgt_obj=max_recursion_manifest)
resp = retry(
self._make_request, method='GET', container=container,
obj=one_more_link)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, TARGET_BODY)
self.assertEqual(resp.getheader('content-length'),
str(self.env.tgt_length))
self.assertIn('Content-Location', resp.headers)
self.assertIn('%s/%s' % (container, max_recursion_manifest),
resp.getheader('content-location'))
# PUT a new slo manifest point to the max_recursion_manifest
# Symlink and slo manifest chain from the new manifest to the final
# target has (max_slo_recursion_depth + 1) manifests.
too_many_recursion_manifest = next(self.obj_name_gen)
manifest = json.dumps(
[{'path': '/%s/%s' % (container, max_recursion_manifest)}])
resp = retry(self._make_request, method='PUT',
container=container, obj=too_many_recursion_manifest,
body=manifest.encode('ascii'),
query_args='multipart-manifest=put')
self.assertEqual(resp.status, 201) # sanity
# Check GET to too_many_recursion_mani returns 409 error
resp = retry(self._make_request, method='GET',
container=container, obj=too_many_recursion_manifest)
self.assertEqual(resp.status, 409)
# N.B. This error message is from slo middleware that uses default.
self.assertEqual(
resp.content,
b'
Conflict
There was a conflict when trying to'
b' complete your request.
')
def test_symlink_put_missing_target_container(self):
link_obj = uuid4().hex
# set only object, no container in the prefix
headers = {'X-Symlink-Target': self.env.tgt_obj}
resp = retry(self._make_request, method='PUT',
container=self.env.link_cont, obj=link_obj,
headers=headers)
self.assertEqual(resp.status, 412)
self.assertEqual(resp.content,
b'X-Symlink-Target header must be of the form'
b' /