#!/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 unittest2
import itertools
import hashlib
import time
from six.moves import urllib
from uuid import uuid4
from unittest2 import SkipTest
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
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
TARGET_BODY = '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):
return '%s/%s' % (cls.tgt_cont, cls.tgt_obj)
@classmethod
def _make_request(cls, url, token, parsed, conn, method,
container, obj='', headers=None, body='',
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 != 201:
raise ResponseError(resp)
return name
@classmethod
def _create_tgt_object(cls):
resp = retry(cls._make_request, method='PUT',
container=cls.tgt_cont, obj=cls.tgt_obj,
body=TARGET_BODY)
if resp.status != 201:
raise ResponseError(resp)
# sanity: successful put response has content-length 0
cls.tgt_length = str(len(TARGET_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 != TARGET_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:
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 resp.status // 100 != 2:
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 != 204):
raise ResponseError(resp)
# delete the containers
for use_account, containers in delete_containers:
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()
def tearDown(self):
self.env.tearDown()
def _make_request(self, url, token, parsed, conn, method,
container, obj='', headers=None, body='',
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=''):
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_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)
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)
# TODO: content-location is a full path so it's better to assert
# with the value, instead of assertIn
self.assertIn(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, '')
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
# Now let's write a new 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)
# PUT symlink
self._test_put_symlink(link_cont=self.env.link_cont, link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=target_obj)
self._assertSymlink(
self.env.link_cont, link_obj,
expected_content_location="%s/%s" % (self.env.tgt_cont,
target_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_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, '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)
expected_location_hdr = "%s/%s" % (self.env.tgt_cont, target_obj)
self.assertIn(expected_location_hdr,
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.assertIn(expected_location_hdr,
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, '')
# 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,
'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,
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,
'<html><h1>Conflict</h1><p>There was a conflict when trying to'
' complete your request.</p></html>')
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,
'X-Symlink-Target header must be of the form'
' <container name>/<object name>')
def test_symlink_put_non_zero_length(self):
link_obj = uuid4().hex
headers = {'X-Symlink-Target':
'%s/%s' % (self.env.tgt_cont, self.env.tgt_obj)}
resp = retry(
self._make_request, method='PUT', container=self.env.link_cont,
obj=link_obj, body='non-zero-length', headers=headers)
self.assertEqual(resp.status, 400)
self.assertEqual(resp.content,
'Symlink requests require a zero byte body')
def test_symlink_target_itself(self):
link_obj = uuid4().hex
headers = {
'X-Symlink-Target': '%s/%s' % (self.env.link_cont, link_obj)}
resp = retry(self._make_request, method='PUT',
container=self.env.link_cont, obj=link_obj,
headers=headers)
self.assertEqual(resp.status, 400)
self.assertEqual(resp.content, 'Symlink cannot target itself')
def test_symlink_target_each_other(self):
symloop_max = cluster_info['symlink']['symloop_max']
link_obj1 = uuid4().hex
link_obj2 = uuid4().hex
# PUT two links which targets each other
self._test_put_symlink(
link_cont=self.env.link_cont, link_obj=link_obj1,
tgt_cont=self.env.link_cont, tgt_obj=link_obj2)
self._test_put_symlink(
link_cont=self.env.link_cont, link_obj=link_obj2,
tgt_cont=self.env.link_cont, tgt_obj=link_obj1)
for obj in (link_obj1, link_obj2):
# sanity: HEAD/GET on the link itself is ok
self._assertLinkObject(self.env.link_cont, obj)
for obj in (link_obj1, link_obj2):
resp = retry(self._make_request, method='HEAD',
container=self.env.link_cont, obj=obj)
self.assertEqual(resp.status, 409)
resp = retry(self._make_request, method='GET',
container=self.env.link_cont, obj=obj)
self.assertEqual(resp.status, 409)
self.assertEqual(
resp.content,
'Too many levels of symbolic links, maximum allowed is %d' %
symloop_max)
def test_symlink_put_copy_from(self):
link_obj1 = uuid4().hex
link_obj2 = uuid4().hex
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=link_obj1,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
copy_src = '%s/%s' % (self.env.link_cont, link_obj1)
# copy symlink
headers = {'X-Copy-From': copy_src}
resp = retry(self._make_request_with_symlink_get,
method='PUT',
container=self.env.link_cont, obj=link_obj2,
headers=headers)
self.assertEqual(resp.status, 201)
self._assertSymlink(link_cont=self.env.link_cont, link_obj=link_obj2)
@requires_acls
def test_symlink_put_copy_from_cross_account(self):
link_obj1 = uuid4().hex
link_obj2 = uuid4().hex
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=link_obj1,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
copy_src = '%s/%s' % (self.env.link_cont, link_obj1)
account_one = tf.parsed[0].path.split('/', 2)[2]
perm_two = tf.swift_test_perm[1]
# add X-Content-Read to account 1 link_cont and tgt_cont
# permit account 2 to read account 1 link_cont to perform copy_src
# and tgt_cont so that link_obj2 can refer to tgt_object
# this ACL allows the copy to succeed
headers = {'X-Container-Read': perm_two}
resp = retry(
self._make_request, method='POST',
container=self.env.link_cont, headers=headers)
self.assertEqual(resp.status, 204)
# this ACL allows link_obj in account 2 to target object in account 1
resp = retry(self._make_request, method='POST',
container=self.env.tgt_cont, headers=headers)
self.assertEqual(resp.status, 204)
# copy symlink itself to a different account w/o
# X-Symlink-Target-Account. This operation will result in copying
# symlink to the account 2 container that points to the
# container/object in the account 2.
# (the container/object is not prepared)
headers = {'X-Copy-From-Account': account_one,
'X-Copy-From': copy_src}
resp = retry(self._make_request_with_symlink_get, method='PUT',
container=self.env.link_cont, obj=link_obj2,
headers=headers, use_account=2)
self.assertEqual(resp.status, 201)
# sanity: HEAD/GET on link_obj itself
self._assertLinkObject(self.env.link_cont, link_obj2, use_account=2)
# no target object in the account 2
for method in ('HEAD', 'GET'):
resp = retry(
self._make_request, method=method,
container=self.env.link_cont, obj=link_obj2, use_account=2)
self.assertEqual(resp.status, 404)
self.assertIn('content-location', resp.headers)
self.assertIn(self.env.target_content_location(),
resp.getheader('content-location'))
# copy symlink itself to a different account with target account
# the target path will be in account 1
# the target path will have an object
headers = {'X-Symlink-target-Account': account_one,
'X-Copy-From-Account': account_one,
'X-Copy-From': copy_src}
resp = retry(
self._make_request_with_symlink_get, method='PUT',
container=self.env.link_cont, obj=link_obj2,
headers=headers, use_account=2)
self.assertEqual(resp.status, 201)
self._assertSymlink(link_cont=self.env.link_cont, link_obj=link_obj2,
use_account=2)
def test_symlink_copy_from_target(self):
link_obj1 = uuid4().hex
obj2 = uuid4().hex
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=link_obj1,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
copy_src = '%s/%s' % (self.env.link_cont, link_obj1)
# issuing a COPY request to a symlink w/o symlink=get, should copy
# the target object, not the symlink itself
headers = {'X-Copy-From': copy_src}
resp = retry(self._make_request, method='PUT',
container=self.env.tgt_cont, obj=obj2,
headers=headers)
self.assertEqual(resp.status, 201)
# HEAD to the copied object
resp = retry(self._make_request, method='HEAD',
container=self.env.tgt_cont, obj=obj2)
self.assertEqual(200, resp.status)
self.assertNotIn('Content-Location', resp.headers)
# GET to the copied object
resp = retry(self._make_request, method='GET',
container=self.env.tgt_cont, obj=obj2)
# But... this is a raw object (not a symlink)
self.assertEqual(200, resp.status)
self.assertNotIn('Content-Location', resp.headers)
self.assertEqual(TARGET_BODY, resp.content)
def test_symlink_copy(self):
link_obj1 = uuid4().hex
link_obj2 = uuid4().hex
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=link_obj1,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
copy_dst = '%s/%s' % (self.env.link_cont, link_obj2)
# copy symlink
headers = {'Destination': copy_dst}
resp = retry(
self._make_request_with_symlink_get, method='COPY',
container=self.env.link_cont, obj=link_obj1, headers=headers)
self.assertEqual(resp.status, 201)
self._assertSymlink(link_cont=self.env.link_cont, link_obj=link_obj2)
def test_symlink_copy_target(self):
link_obj1 = uuid4().hex
obj2 = uuid4().hex
self._test_put_symlink(link_cont=self.env.link_cont,
link_obj=link_obj1,
tgt_cont=self.env.tgt_cont,
tgt_obj=self.env.tgt_obj)
copy_dst = '%s/%s' % (self.env.tgt_cont, obj2)
# copy target object
headers = {'Destination': copy_dst}
resp = retry(self._make_request, method='COPY',
container=self.env.link_cont, obj=link_obj1,
headers=headers)
self.assertEqual(resp.status, 201)
# HEAD to target object via symlink
resp = retry(self._make_request, method='HEAD',
container=self.env.tgt_cont, obj=obj2)
self.assertEqual(resp.status, 200)
self.assertNotIn('Content-Location', resp.headers)
# GET to the copied object that should be a raw object (not symlink)
resp = retry(self._make_request, method='GET',
container=self.env.tgt_cont, obj=obj2)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, TARGET_BODY)
self.assertNotIn('Content-Location', resp.headers)
def test_post_symlink(self):
link_obj = uuid4().hex
value1 = uuid4().hex
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)
# POSTing to a symlink is not allowed and should return a 307
headers = {'X-Object-Meta-Alpha': 'apple'}
resp = retry(
self._make_request, method='POST', container=self.env.link_cont,
obj=link_obj, headers=headers, allow_redirects=False)
self.assertEqual(resp.status, 307)
# we are using account 0 in this test
expected_location_hdr = "%s/%s/%s" % (
tf.parsed[0].path, self.env.tgt_cont, self.env.tgt_obj)
self.assertEqual(resp.getheader('Location'), expected_location_hdr)
# Read header from symlink itself. The metadata is applied to symlink
resp = retry(self._make_request_with_symlink_get, method='GET',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.getheader('X-Object-Meta-Alpha'), 'apple')
# Post the target object directly
headers = {'x-object-meta-test': value1}
resp = retry(
self._make_request, method='POST', container=self.env.tgt_cont,
obj=self.env.tgt_obj, headers=headers)
self.assertEqual(resp.status, 202)
resp = retry(self._make_request, method='GET',
container=self.env.tgt_cont, obj=self.env.tgt_obj)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.getheader('X-Object-Meta-Test'), value1)
# Read header from target object via symlink, should exist now.
resp = retry(
self._make_request, method='GET', container=self.env.link_cont,
obj=link_obj)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.getheader('X-Object-Meta-Test'), value1)
# sanity: no X-Object-Meta-Alpha exists in the response header
self.assertNotIn('X-Object-Meta-Alpha', resp.headers)
def test_post_with_symlink_header(self):
# POSTing to a symlink is not allowed and should return a 307
# updating the symlink target with a POST should always fail
headers = {'X-Symlink-Target': 'container/new_target'}
resp = retry(
self._make_request, method='POST', container=self.env.tgt_cont,
obj=self.env.tgt_obj, headers=headers, allow_redirects=False)
self.assertEqual(resp.status, 400)
self.assertEqual(resp.content,
'A PUT request is required to set a symlink target')
def test_overwrite_symlink(self):
link_obj = uuid4().hex
new_tgt_obj = "new_target_object_name"
new_tgt = '%s/%s' % (self.env.tgt_cont, new_tgt_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)
# sanity
self._assertSymlink(self.env.link_cont, link_obj)
# Overwrite symlink with PUT
self._test_put_symlink(link_cont=self.env.link_cont, link_obj=link_obj,
tgt_cont=self.env.tgt_cont,
tgt_obj=new_tgt_obj)
# head symlink to check X-Symlink-Target header
resp = retry(self._make_request_with_symlink_get, method='HEAD',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 200)
# target should remain with old target
self.assertEqual(resp.getheader('X-Symlink-Target'), new_tgt)
def test_delete_symlink(self):
link_obj = uuid4().hex
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)
resp = retry(self._make_request, method='DELETE',
container=self.env.link_cont, obj=link_obj)
self.assertEqual(resp.status, 204)
# make sure target object was not deleted and is still reachable
resp = retry(self._make_request, method='GET',
container=self.env.tgt_cont, obj=self.env.tgt_obj)
self.assertEqual(resp.status, 200)
self.assertEqual(resp.content, TARGET_BODY)
@requires_acls
def test_symlink_put_target_account(self):
if tf.skip or tf.skip2:
raise SkipTest
link_obj = uuid4().hex
account_one = tf.parsed[0].path.split('/', 2)[2]
# create symlink in account 2
# pointing to account 1
headers = {'X-Symlink-Target-Account': account_one,
'X-Symlink-Target':
'%s/%s' % (self.env.tgt_cont, self.env.tgt_obj)}
resp = retry(self._make_request, method='PUT',
container=self.env.link_cont, obj=link_obj,
headers=headers, use_account=2)
self.assertEqual(resp.status, 201)
perm_two = tf.swift_test_perm[1]
# sanity test:
# it should be ok to get the symlink itself, but not the target object
# because the read acl has not been configured yet
self._assertLinkObject(self.env.link_cont, link_obj, use_account=2)
resp = retry(
self._make_request, method='GET',
container=self.env.link_cont, obj=link_obj, use_account=2)
self.assertEqual(resp.status, 403)
# add X-Content-Read to account 1 tgt_cont
# permit account 2 to read account 1 tgt_cont
# add acl to allow reading from source
headers = {'X-Container-Read': perm_two}
resp = retry(self._make_request, method='POST',
container=self.env.tgt_cont, headers=headers)
self.assertEqual(resp.status, 204)
# GET on link_obj itself
self._assertLinkObject(self.env.link_cont, link_obj, use_account=2)
# GET to target object via symlink
resp = self._test_get_as_target_object(
self.env.link_cont, link_obj,
expected_content_location=self.env.target_content_location(),
use_account=2)
self.assertIn(account_one, resp.getheader('content-location'))
def test_symlink_object_listing(self):
link_obj = uuid4().hex
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)
# sanity
self._assertSymlink(self.env.link_cont, link_obj)
resp = retry(self._make_request, method='GET',
container=self.env.link_cont,
query_args='format=json')
self.assertEqual(resp.status, 200)
object_list = json.loads(resp.content)
self.assertEqual(len(object_list), 1)
self.assertIn('symlink_path', object_list[0])
self.assertIn(self.env.target_content_location(),
object_list[0]['symlink_path'])
class TestCrossPolicySymlinkEnv(TestSymlinkEnv):
multiple_policies_enabled = None
@classmethod
def setUp(cls):
if tf.skip or tf.skip2:
raise SkipTest
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
return
link_policy = cls.policies.select()
tgt_policy = cls.policies.exclude(name=link_policy['name']).select()
link_header = {'X-Storage-Policy': link_policy['name']}
tgt_header = {'X-Storage-Policy': tgt_policy['name']}
cls._create_container(cls.link_cont, headers=link_header)
cls._create_container(cls.tgt_cont, headers=tgt_header)
# container in account 2
cls._create_container(cls.link_cont, headers=link_header,
use_account=2)
cls._create_tgt_object()
class TestCrossPolicySymlink(TestSymlink):
env = TestCrossPolicySymlinkEnv
def setUp(self):
super(TestCrossPolicySymlink, 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.multiple_policies_enabled,))
def tearDown(self):
self.env.tearDown()
class TestSymlinkSlo(Base):
"""
Just some sanity testing of SLO + symlinks.
It is basically a copy of SLO tests in test_slo, but the tested object is
a symlink to the manifest (instead of the manifest itself)
"""
env = TestSloEnv
def setUp(self):
super(TestSymlinkSlo, self).setUp()
if self.env.slo_enabled is False:
raise SkipTest("SLO not enabled")
elif self.env.slo_enabled is not True:
# just some sanity checking
raise Exception(
"Expected slo_enabled to be True/False, got %r" %
(self.env.slo_enabled,))
self.file_symlink = self.env.container.file(uuid4().hex)
def test_symlink_target_slo_manifest(self):
self.file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'manifest-abcde')})
file_contents = self.file_symlink.read()
self.assertEqual(4 * 1024 * 1024 + 1, len(file_contents))
self.assertEqual('a', file_contents[0])
self.assertEqual('a', file_contents[1024 * 1024 - 1])
self.assertEqual('b', file_contents[1024 * 1024])
self.assertEqual('d', file_contents[-2])
self.assertEqual('e', file_contents[-1])
def test_symlink_target_slo_nested_manifest(self):
self.file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'manifest-abcde-submanifest')})
file_contents = self.file_symlink.read()
self.assertEqual(4 * 1024 * 1024 + 1, len(file_contents))
self.assertEqual('a', file_contents[0])
self.assertEqual('a', file_contents[1024 * 1024 - 1])
self.assertEqual('b', file_contents[1024 * 1024])
self.assertEqual('d', file_contents[-2])
self.assertEqual('e', file_contents[-1])
def test_slo_get_ranged_manifest(self):
self.file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'ranged-manifest')})
grouped_file_contents = [
(char, sum(1 for _char in grp))
for char, grp in itertools.groupby(self.file_symlink.read())]
self.assertEqual([
('c', 1),
('d', 1024 * 1024),
('e', 1),
('a', 512 * 1024),
('b', 512 * 1024),
('c', 1),
('d', 1)], grouped_file_contents)
def test_slo_ranged_get(self):
self.file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'manifest-abcde')})
file_contents = self.file_symlink.read(size=1024 * 1024 + 2,
offset=1024 * 1024 - 1)
self.assertEqual('a', file_contents[0])
self.assertEqual('b', file_contents[1])
self.assertEqual('b', file_contents[-2])
self.assertEqual('c', file_contents[-1])
class TestSymlinkSloEnv(TestSloEnv):
@classmethod
def create_links_to_segments(cls, container):
seg_info = {}
for letter in ('a', 'b'):
seg_name = "linkto_seg_%s" % letter
file_item = container.file(seg_name)
sym_hdr = {'X-Symlink-Target': '%s/seg_%s' % (container.name,
letter)}
file_item.write(hdrs=sym_hdr)
seg_info[seg_name] = {
'path': '/%s/%s' % (container.name, seg_name)}
return seg_info
@classmethod
def setUp(cls):
super(TestSymlinkSloEnv, cls).setUp()
cls.link_seg_info = cls.create_links_to_segments(cls.container)
file_item = cls.container.file("manifest-linkto-ab")
file_item.write(
json.dumps([cls.link_seg_info['linkto_seg_a'],
cls.link_seg_info['linkto_seg_b']]),
parms={'multipart-manifest': 'put'})
class TestSymlinkToSloSegments(Base):
"""
This test class will contain various tests where the segments of the SLO
manifest are symlinks to the actual segments. Again the tests are basicaly
a copy/paste of the tests in test_slo, only the manifest has been modified
to contain symlinks as the segments.
"""
env = TestSymlinkSloEnv
def setUp(self):
super(TestSymlinkToSloSegments, self).setUp()
if self.env.slo_enabled is False:
raise SkipTest("SLO not enabled")
elif self.env.slo_enabled is not True:
# just some sanity checking
raise Exception(
"Expected slo_enabled to be True/False, got %r" %
(self.env.slo_enabled,))
def test_slo_get_simple_manifest_with_links(self):
file_item = self.env.container.file("manifest-linkto-ab")
file_contents = file_item.read()
self.assertEqual(2 * 1024 * 1024, len(file_contents))
self.assertEqual('a', file_contents[0])
self.assertEqual('a', file_contents[1024 * 1024 - 1])
self.assertEqual('b', file_contents[1024 * 1024])
def test_slo_container_listing(self):
# the listing object size should equal the sum of the size of the
# segments, not the size of the manifest body
file_item = self.env.container.file(Utils.create_name())
file_item.write(
json.dumps([self.env.link_seg_info['linkto_seg_a']]),
parms={'multipart-manifest': 'put'})
# The container listing has the etag of the actual manifest object
# contents which we get using multipart-manifest=get. Arguably this
# should be the etag that we get when NOT using multipart-manifest=get,
# to be consistent with size and content-type. But here we at least
# verify that it remains consistent when the object is updated with a
# POST.
file_item.initialize(parms={'multipart-manifest': 'get'})
expected_etag = file_item.etag
listing = self.env.container.files(parms={'format': 'json'})
for f_dict in listing:
if f_dict['name'] == file_item.name:
self.assertEqual(1024 * 1024, f_dict['bytes'])
self.assertEqual('application/octet-stream',
f_dict['content_type'])
self.assertEqual(expected_etag, f_dict['hash'])
break
else:
self.fail('Failed to find manifest file in container listing')
# now POST updated content-type file
file_item.content_type = 'image/jpeg'
file_item.sync_metadata({'X-Object-Meta-Test': 'blah'})
file_item.initialize()
self.assertEqual('image/jpeg', file_item.content_type) # sanity
# verify that the container listing is consistent with the file
listing = self.env.container.files(parms={'format': 'json'})
for f_dict in listing:
if f_dict['name'] == file_item.name:
self.assertEqual(1024 * 1024, f_dict['bytes'])
self.assertEqual(file_item.content_type,
f_dict['content_type'])
self.assertEqual(expected_etag, f_dict['hash'])
break
else:
self.fail('Failed to find manifest file in container listing')
# now POST with no change to content-type
file_item.sync_metadata({'X-Object-Meta-Test': 'blah'},
cfg={'no_content_type': True})
file_item.initialize()
self.assertEqual('image/jpeg', file_item.content_type) # sanity
# verify that the container listing is consistent with the file
listing = self.env.container.files(parms={'format': 'json'})
for f_dict in listing:
if f_dict['name'] == file_item.name:
self.assertEqual(1024 * 1024, f_dict['bytes'])
self.assertEqual(file_item.content_type,
f_dict['content_type'])
self.assertEqual(expected_etag, f_dict['hash'])
break
else:
self.fail('Failed to find manifest file in container listing')
def test_slo_etag_is_hash_of_etags(self):
expected_hash = hashlib.md5()
expected_hash.update(hashlib.md5('a' * 1024 * 1024).hexdigest())
expected_hash.update(hashlib.md5('b' * 1024 * 1024).hexdigest())
expected_etag = expected_hash.hexdigest()
file_item = self.env.container.file('manifest-linkto-ab')
self.assertEqual(expected_etag, file_item.info()['etag'])
def test_slo_copy(self):
file_item = self.env.container.file("manifest-linkto-ab")
file_item.copy(self.env.container.name, "copied-abcde")
copied = self.env.container.file("copied-abcde")
copied_contents = copied.read(parms={'multipart-manifest': 'get'})
self.assertEqual(2 * 1024 * 1024, len(copied_contents))
def test_slo_copy_the_manifest(self):
# first just perform some tests of the contents of the manifest itself
source = self.env.container.file("manifest-linkto-ab")
source_contents = source.read(parms={'multipart-manifest': 'get'})
source_json = json.loads(source_contents)
source.initialize()
self.assertEqual('application/octet-stream', source.content_type)
source.initialize(parms={'multipart-manifest': 'get'})
source_hash = hashlib.md5()
source_hash.update(source_contents)
self.assertEqual(source_hash.hexdigest(), source.etag)
# now, copy the manifest
self.assertTrue(source.copy(self.env.container.name,
"copied-ab-manifest-only",
parms={'multipart-manifest': 'get'}))
copied = self.env.container.file("copied-ab-manifest-only")
copied_contents = copied.read(parms={'multipart-manifest': 'get'})
try:
copied_json = json.loads(copied_contents)
except ValueError:
self.fail("COPY didn't copy the manifest (invalid json on GET)")
# make sure content of copied manifest is the same as original man.
self.assertEqual(source_json, copied_json)
copied.initialize()
self.assertEqual('application/octet-stream', copied.content_type)
copied.initialize(parms={'multipart-manifest': 'get'})
copied_hash = hashlib.md5()
copied_hash.update(copied_contents)
self.assertEqual(copied_hash.hexdigest(), copied.etag)
self.assertEqual(copied_hash.hexdigest(), source.etag)
# verify the listing metadata
listing = self.env.container.files(parms={'format': 'json'})
names = {}
for f_dict in listing:
if f_dict['name'] in ('manifest-linkto-ab',
'copied-ab-manifest-only'):
names[f_dict['name']] = f_dict
self.assertIn('manifest-linkto-ab', names)
actual = names['manifest-linkto-ab']
self.assertEqual(2 * 1024 * 1024, actual['bytes'])
self.assertEqual('application/octet-stream', actual['content_type'])
self.assertEqual(source.etag, actual['hash'])
self.assertIn('copied-ab-manifest-only', names)
actual = names['copied-ab-manifest-only']
self.assertEqual(2 * 1024 * 1024, actual['bytes'])
self.assertEqual('application/octet-stream', actual['content_type'])
self.assertEqual(copied.etag, actual['hash'])
class TestSymlinkDlo(Base):
env = TestDloEnv
def test_get_manifest(self):
link_obj = uuid4().hex
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'man1')})
file_contents = file_symlink.read()
self.assertEqual(
file_contents,
"aaaaaaaaaabbbbbbbbbbccccccccccddddddddddeeeeeeeeee")
link_obj = uuid4().hex
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'man2')})
file_contents = file_symlink.read()
self.assertEqual(
file_contents,
"AAAAAAAAAABBBBBBBBBBCCCCCCCCCCDDDDDDDDDDEEEEEEEEEE")
link_obj = uuid4().hex
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'manall')})
file_contents = file_symlink.read()
self.assertEqual(
file_contents,
("aaaaaaaaaabbbbbbbbbbccccccccccddddddddddeeeeeeeeee" +
"AAAAAAAAAABBBBBBBBBBCCCCCCCCCCDDDDDDDDDDEEEEEEEEEE"))
def test_get_manifest_document_itself(self):
link_obj = uuid4().hex
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'man1')})
file_contents = file_symlink.read(parms={'multipart-manifest': 'get'})
self.assertEqual(file_contents, "man1-contents")
self.assertEqual(file_symlink.info()['x_object_manifest'],
"%s/%s/seg_lower" %
(self.env.container.name, self.env.segment_prefix))
def test_get_range(self):
link_obj = uuid4().hex + "_symlink"
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'man1')})
file_contents = file_symlink.read(size=25, offset=8)
self.assertEqual(file_contents, "aabbbbbbbbbbccccccccccddd")
file_contents = file_symlink.read(size=1, offset=47)
self.assertEqual(file_contents, "e")
def test_get_range_out_of_range(self):
link_obj = uuid4().hex
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
'man1')})
self.assertRaises(ResponseError, file_symlink.read, size=7, offset=50)
self.assert_status(416)
class TestSymlinkTargetObjectComparisonEnv(TestFileComparisonEnv):
@classmethod
def setUp(cls):
super(TestSymlinkTargetObjectComparisonEnv, cls).setUp()
cls.parms = None
cls.expect_empty_etag = False
cls.expect_body = True
class TestSymlinkComparisonEnv(TestFileComparisonEnv):
@classmethod
def setUp(cls):
super(TestSymlinkComparisonEnv, cls).setUp()
cls.parms = {'symlink': 'get'}
cls.expect_empty_etag = True
cls.expect_body = False
class TestSymlinkTargetObjectComparison(Base):
env = TestSymlinkTargetObjectComparisonEnv
def setUp(self):
super(TestSymlinkTargetObjectComparison, self).setUp()
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
file_item.name)})
def testIfMatch(self):
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
md5 = MD5_OF_EMPTY_STRING if self.env.expect_empty_etag else \
file_item.md5
hdrs = {'If-Match': md5}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
if self.env.expect_body:
self.assertTrue(body)
else:
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
hdrs = {'If-Match': 'bogus'}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(412)
self.assert_header('etag', md5)
def testIfMatchMultipleEtags(self):
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
md5 = MD5_OF_EMPTY_STRING if self.env.expect_empty_etag else \
file_item.md5
hdrs = {'If-Match': '"bogus1", "%s", "bogus2"' % md5}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
if self.env.expect_body:
self.assertTrue(body)
else:
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
hdrs = {'If-Match': '"bogus1", "bogus2", "bogus3"'}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(412)
self.assert_header('etag', md5)
def testIfNoneMatch(self):
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
md5 = MD5_OF_EMPTY_STRING if self.env.expect_empty_etag else \
file_item.md5
hdrs = {'If-None-Match': 'bogus'}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
if self.env.expect_body:
self.assertTrue(body)
else:
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
hdrs = {'If-None-Match': md5}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(304)
self.assert_header('etag', md5)
self.assert_header('accept-ranges', 'bytes')
def testIfNoneMatchMultipleEtags(self):
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
md5 = MD5_OF_EMPTY_STRING if self.env.expect_empty_etag else \
file_item.md5
hdrs = {'If-None-Match': '"bogus1", "bogus2", "bogus3"'}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
if self.env.expect_body:
self.assertTrue(body)
else:
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
hdrs = {'If-None-Match':
'"bogus1", "bogus2", "%s"' % md5}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(304)
self.assert_header('etag', md5)
self.assert_header('accept-ranges', 'bytes')
def testIfModifiedSince(self):
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
md5 = MD5_OF_EMPTY_STRING if self.env.expect_empty_etag else \
file_item.md5
hdrs = {'If-Modified-Since': self.env.time_old_f1}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
if self.env.expect_body:
self.assertTrue(body)
else:
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
self.assertTrue(file_symlink.info(hdrs=hdrs, parms=self.env.parms))
hdrs = {'If-Modified-Since': self.env.time_new}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(304)
self.assert_header('etag', md5)
self.assert_header('accept-ranges', 'bytes')
self.assertRaises(ResponseError, file_symlink.info, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(304)
self.assert_header('etag', md5)
self.assert_header('accept-ranges', 'bytes')
def testIfUnmodifiedSince(self):
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
md5 = MD5_OF_EMPTY_STRING if self.env.expect_empty_etag else \
file_item.md5
hdrs = {'If-Unmodified-Since': self.env.time_new}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
if self.env.expect_body:
self.assertTrue(body)
else:
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
self.assertTrue(file_symlink.info(hdrs=hdrs, parms=self.env.parms))
hdrs = {'If-Unmodified-Since': self.env.time_old_f2}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(412)
self.assert_header('etag', md5)
self.assertRaises(ResponseError, file_symlink.info, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(412)
self.assert_header('etag', md5)
def testIfMatchAndUnmodified(self):
for file_item in self.env.files:
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
md5 = MD5_OF_EMPTY_STRING if self.env.expect_empty_etag else \
file_item.md5
hdrs = {'If-Match': md5,
'If-Unmodified-Since': self.env.time_new}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
if self.env.expect_body:
self.assertTrue(body)
else:
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
hdrs = {'If-Match': 'bogus',
'If-Unmodified-Since': self.env.time_new}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(412)
self.assert_header('etag', md5)
hdrs = {'If-Match': md5,
'If-Unmodified-Since': self.env.time_old_f3}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(412)
self.assert_header('etag', md5)
def testLastModified(self):
file_item = self.env.container.file(Utils.create_name())
file_item.content_type = Utils.create_name()
resp = file_item.write_random_return_resp(self.env.file_size)
put_last_modified = resp.getheader('last-modified')
md5 = file_item.md5
# create symlink
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
file_symlink.write(hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
file_item.name)})
info = file_symlink.info()
self.assertIn('last_modified', info)
last_modified = info['last_modified']
self.assertEqual(put_last_modified, info['last_modified'])
hdrs = {'If-Modified-Since': last_modified}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs)
self.assert_status(304)
self.assert_header('etag', md5)
self.assert_header('accept-ranges', 'bytes')
hdrs = {'If-Unmodified-Since': last_modified}
self.assertTrue(file_symlink.read(hdrs=hdrs))
class TestSymlinkComparison(TestSymlinkTargetObjectComparison):
env = TestSymlinkComparisonEnv
def setUp(self):
super(TestSymlinkComparison, self).setUp()
def testLastModified(self):
file_item = self.env.container.file(Utils.create_name())
file_item.content_type = Utils.create_name()
resp = file_item.write_random_return_resp(self.env.file_size)
put_target_last_modified = resp.getheader('last-modified')
md5 = MD5_OF_EMPTY_STRING
# get different last-modified between file and symlink
time.sleep(1)
# create symlink
link_obj = file_item.name + '_symlink'
file_symlink = self.env.container.file(link_obj)
resp = file_symlink.write(return_resp=True,
hdrs={'X-Symlink-Target':
'%s/%s' % (self.env.container.name,
file_item.name)})
put_sym_last_modified = resp.getheader('last-modified')
info = file_symlink.info(parms=self.env.parms)
self.assertIn('last_modified', info)
last_modified = info['last_modified']
self.assertEqual(put_sym_last_modified, info['last_modified'])
hdrs = {'If-Modified-Since': put_target_last_modified}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
hdrs = {'If-Modified-Since': last_modified}
self.assertRaises(ResponseError, file_symlink.read, hdrs=hdrs,
parms=self.env.parms)
self.assert_status(304)
self.assert_header('etag', md5)
self.assert_header('accept-ranges', 'bytes')
hdrs = {'If-Unmodified-Since': last_modified}
body = file_symlink.read(hdrs=hdrs, parms=self.env.parms)
self.assertEqual('', body)
self.assert_status(200)
self.assert_header('etag', md5)
class TestSymlinkAccountTempurl(Base):
env = TestTempurlEnv
def setUp(self):
super(TestSymlinkAccountTempurl, self).setUp()
if self.env.tempurl_enabled is False:
raise SkipTest("TempURL not enabled")
elif self.env.tempurl_enabled is not True:
# just some sanity checking
raise Exception(
"Expected tempurl_enabled to be True/False, got %r" %
(self.env.tempurl_enabled,))
self.expires = int(time.time()) + 86400
self.obj_tempurl_parms = self.tempurl_parms(
'GET', self.expires, self.env.conn.make_path(self.env.obj.path),
self.env.tempurl_key)
def tempurl_parms(self, method, expires, path, key):
sig = hmac.new(
key,
'%s\n%s\n%s' % (method, expires, urllib.parse.unquote(path)),
hashlib.sha1).hexdigest()
return {'temp_url_sig': sig, 'temp_url_expires': str(expires)}
def test_PUT_symlink(self):
new_sym = self.env.container.file(Utils.create_name())
# give out a signature which allows a PUT to new_obj
expires = int(time.time()) + 86400
put_parms = self.tempurl_parms(
'PUT', expires, self.env.conn.make_path(new_sym.path),
self.env.tempurl_key)
# try to create symlink object
try:
new_sym.write(
'', {'x-symlink-target': 'cont/foo'}, parms=put_parms,
cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 400)
else:
self.fail('request did not error')
def test_GET_symlink_inside_container(self):
tgt_obj = self.env.container.file(Utils.create_name())
sym = self.env.container.file(Utils.create_name())
tgt_obj.write("target object body")
sym.write(
'',
{'x-symlink-target': '%s/%s' % (self.env.container.name, tgt_obj)})
expires = int(time.time()) + 86400
get_parms = self.tempurl_parms(
'GET', expires, self.env.conn.make_path(sym.path),
self.env.tempurl_key)
contents = sym.read(parms=get_parms, cfg={'no_auth_token': True})
self.assert_status([200])
self.assertEqual(contents, "target object body")
def test_GET_symlink_outside_container(self):
tgt_obj = self.env.container.file(Utils.create_name())
tgt_obj.write("target object body")
container2 = self.env.account.container(Utils.create_name())
container2.create()
sym = container2.file(Utils.create_name())
sym.write(
'',
{'x-symlink-target': '%s/%s' % (self.env.container.name, tgt_obj)})
expires = int(time.time()) + 86400
get_parms = self.tempurl_parms(
'GET', expires, self.env.conn.make_path(sym.path),
self.env.tempurl_key)
# cross container tempurl works fine for account tempurl key
contents = sym.read(parms=get_parms, cfg={'no_auth_token': True})
self.assert_status([200])
self.assertEqual(contents, "target object body")
class TestSymlinkContainerTempurl(Base):
env = TestContainerTempurlEnv
def setUp(self):
super(TestSymlinkContainerTempurl, self).setUp()
if self.env.tempurl_enabled is False:
raise SkipTest("TempURL not enabled")
elif self.env.tempurl_enabled is not True:
# just some sanity checking
raise Exception(
"Expected tempurl_enabled to be True/False, got %r" %
(self.env.tempurl_enabled,))
expires = int(time.time()) + 86400
sig = self.tempurl_sig(
'GET', expires, self.env.conn.make_path(self.env.obj.path),
self.env.tempurl_key)
self.obj_tempurl_parms = {'temp_url_sig': sig,
'temp_url_expires': str(expires)}
def tempurl_sig(self, method, expires, path, key):
return hmac.new(
key,
'%s\n%s\n%s' % (method, expires, urllib.parse.unquote(path)),
hashlib.sha1).hexdigest()
def test_PUT_symlink(self):
new_sym = self.env.container.file(Utils.create_name())
# give out a signature which allows a PUT to new_obj
expires = int(time.time()) + 86400
sig = self.tempurl_sig(
'PUT', expires, self.env.conn.make_path(new_sym.path),
self.env.tempurl_key)
put_parms = {'temp_url_sig': sig,
'temp_url_expires': str(expires)}
# try to create symlink object, should fail
try:
new_sym.write(
'', {'x-symlink-target': 'cont/foo'}, parms=put_parms,
cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 400)
else:
self.fail('request did not error')
def test_GET_symlink_inside_container(self):
tgt_obj = self.env.container.file(Utils.create_name())
sym = self.env.container.file(Utils.create_name())
tgt_obj.write("target object body")
sym.write(
'',
{'x-symlink-target': '%s/%s' % (self.env.container.name, tgt_obj)})
expires = int(time.time()) + 86400
sig = self.tempurl_sig(
'GET', expires, self.env.conn.make_path(sym.path),
self.env.tempurl_key)
parms = {'temp_url_sig': sig,
'temp_url_expires': str(expires)}
contents = sym.read(parms=parms, cfg={'no_auth_token': True})
self.assert_status([200])
self.assertEqual(contents, "target object body")
def test_GET_symlink_outside_container(self):
tgt_obj = self.env.container.file(Utils.create_name())
tgt_obj.write("target object body")
container2 = self.env.account.container(Utils.create_name())
container2.create()
sym = container2.file(Utils.create_name())
sym.write(
'',
{'x-symlink-target': '%s/%s' % (self.env.container.name, tgt_obj)})
expires = int(time.time()) + 86400
sig = self.tempurl_sig(
'GET', expires, self.env.conn.make_path(sym.path),
self.env.tempurl_key)
parms = {'temp_url_sig': sig,
'temp_url_expires': str(expires)}
# cross container tempurl does not work for container tempurl key
try:
sym.read(parms=parms, cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 401)
else:
self.fail('request did not error')
try:
sym.info(parms=parms, cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 401)
else:
self.fail('request did not error')
if __name__ == '__main__':
unittest2.main()