Unit test S3 Bucket resource with associated fixes.

Change-Id: Ib23ed500385c299247bf80062a7a4342f5afe1d2
This commit is contained in:
Steve Baker 2012-09-17 12:03:05 +12:00
parent 77700c10bf
commit 8df32a93d2
3 changed files with 260 additions and 12 deletions

View File

@ -59,19 +59,24 @@ class S3Bucket(Resource):
return {'Error':
'S3 services unavaialble because of missing swiftclient.'}
@staticmethod
def _create_container_name(resource_name):
return 'heat-%s-%s' % (resource_name,
binascii.hexlify(os.urandom(10)))
def handle_create(self):
"""Create a bucket."""
container = 'heat-%s-%s' % (self.resource_physical_name(),
binascii.hexlify(os.urandom(10)))
container = S3Bucket._create_container_name(
self.physical_resource_name())
headers = {}
logger.debug('S3Bucket create container %s with headers %s' %
(container, headers))
if 'WebsiteConfiguration' in self.properties:
site_cfg = self.properties['WebsiteConfiguration']
if self.properties['WebsiteConfiguration'] is not None:
sc = self.properties['WebsiteConfiguration']
# we will assume that swift is configured for the staticweb
# wsgi middleware
headers['X-Container-Meta-Web-Index'] = site_cfg['IndexDocument']
headers['X-Container-Meta-Web-Error'] = site_cfg['ErrorDocument']
headers['X-Container-Meta-Web-Index'] = sc['IndexDocument']
headers['X-Container-Meta-Web-Error'] = sc['ErrorDocument']
con = self.context
ac = self.properties['AccessControl']

237
heat/tests/test_s3.py Normal file
View File

@ -0,0 +1,237 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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 sys
import os
import re
import nose
import unittest
import mox
import json
from nose.plugins.attrib import attr
from heat.engine import s3
from heat.engine import parser
from nose.exc import SkipTest
try:
from swiftclient import client as swiftclient
except:
raise SkipTest("unable to import swiftclient, skipping")
@attr(tag=['unit', 'resource'])
@attr(speed='fast')
class s3Test(unittest.TestCase):
def setUp(self):
self.m = mox.Mox()
self.m.CreateMock(swiftclient.Connection)
self.m.StubOutWithMock(swiftclient.Connection, 'put_container')
self.m.StubOutWithMock(swiftclient.Connection, 'delete_container')
self.m.StubOutWithMock(swiftclient.Connection, 'get_auth')
self.container_pattern = 'heat-test_stack.test_resource-[0-9a-f]+'
def tearDown(self):
self.m.UnsetStubs()
print "s3Test teardown complete"
def load_template(self):
self.path = os.path.dirname(os.path.realpath(__file__)).\
replace('heat/tests', 'templates')
f = open("%s/S3_Single_Instance.template" % self.path)
t = json.loads(f.read())
f.close()
return t
def parse_stack(self, t):
class DummyContext():
tenant = 'test_tenant'
username = 'test_username'
password = 'password'
auth_url = 'http://localhost:5000/v2.0'
stack = parser.Stack(DummyContext(), 'test_stack', parser.Template(t),
stack_id=-1)
return stack
def create_resource(self, t, stack, resource_name):
resource = s3.S3Bucket('test_resource',
t['Resources'][resource_name],
stack)
self.assertEqual(None, resource.create())
self.assertEqual(s3.S3Bucket.CREATE_COMPLETE, resource.state)
return resource
def test_create_container_name(self):
self.m.UnsetStubs()
self.assertTrue(re.match(self.container_pattern,
s3.S3Bucket._create_container_name('test_stack.test_resource')))
def test_attributes(self):
swiftclient.Connection.put_container(
mox.Regex(self.container_pattern),
{'X-Container-Write': 'test_tenant:test_username',
'X-Container-Read': 'test_tenant:test_username'}).AndReturn(None)
swiftclient.Connection.get_auth().MultipleTimes().AndReturn(
('http://localhost:8080/v_2', None))
swiftclient.Connection.delete_container(
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
stack = self.parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
ref_id = resource.FnGetRefId()
self.assertTrue(re.match(self.container_pattern,
ref_id))
self.assertEquals('localhost', resource.FnGetAtt('DomainName'))
url = 'http://localhost:8080/v_2/%s' % ref_id
self.assertEquals(url, resource.FnGetAtt('WebsiteURL'))
try:
resource.FnGetAtt('Foo')
raise Exception('Expected InvalidTemplateAttribute')
except s3.exception.InvalidTemplateAttribute:
pass
self.assertEquals(s3.S3Bucket.UPDATE_REPLACE, resource.handle_update())
resource.delete()
self.m.VerifyAll()
def test_public_read(self):
swiftclient.Connection.put_container(
mox.Regex(self.container_pattern),
{'X-Container-Write': 'test_tenant:test_username',
'X-Container-Read': '.r:*'}).AndReturn(None)
swiftclient.Connection.delete_container(
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
properties = t['Resources']['S3Bucket']['Properties']
properties['AccessControl'] = 'PublicRead'
stack = self.parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
self.m.VerifyAll()
def test_public_read_write(self):
swiftclient.Connection.put_container(
mox.Regex(self.container_pattern),
{'X-Container-Write': '.r:*',
'X-Container-Read': '.r:*'}).AndReturn(None)
swiftclient.Connection.delete_container(
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
properties = t['Resources']['S3Bucket']['Properties']
properties['AccessControl'] = 'PublicReadWrite'
stack = self.parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
self.m.VerifyAll()
def test_authenticated_read(self):
swiftclient.Connection.put_container(
mox.Regex(self.container_pattern),
{'X-Container-Write': 'test_tenant:test_username',
'X-Container-Read': 'test_tenant'}).AndReturn(None)
swiftclient.Connection.delete_container(
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
properties = t['Resources']['S3Bucket']['Properties']
properties['AccessControl'] = 'AuthenticatedRead'
stack = self.parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
self.m.VerifyAll()
def test_website(self):
swiftclient.Connection.put_container(
mox.Regex(self.container_pattern),
{'X-Container-Meta-Web-Error': 'error.html',
'X-Container-Meta-Web-Index': 'index.html',
'X-Container-Write': 'test_tenant:test_username',
'X-Container-Read': '.r:*'}).AndReturn(None)
swiftclient.Connection.delete_container(
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
stack = self.parse_stack(t)
resource = self.create_resource(t, stack, 'S3BucketWebsite')
resource.delete()
self.m.VerifyAll()
def test_delete_exception(self):
swiftclient.Connection.put_container(
mox.Regex(self.container_pattern),
{'X-Container-Write': 'test_tenant:test_username',
'X-Container-Read': 'test_tenant:test_username'}).AndReturn(None)
swiftclient.Connection.delete_container(
mox.Regex(self.container_pattern)).AndRaise(
swiftclient.ClientException('Test delete failure'))
self.m.ReplayAll()
t = self.load_template()
stack = self.parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
self.m.VerifyAll()
def test_delete_retain(self):
# first run, with retain policy
swiftclient.Connection.put_container(
mox.Regex(self.container_pattern),
{'X-Container-Write': 'test_tenant:test_username',
'X-Container-Read': 'test_tenant:test_username'}).AndReturn(None)
# This should not be called
swiftclient.Connection.delete_container(
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
t = self.load_template()
properties = t['Resources']['S3Bucket']['Properties']
properties['DeletionPolicy'] = 'Retain'
stack = self.parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
# if delete_container is called, mox verify will succeed
resource.delete()
try:
self.m.VerifyAll()
except mox.ExpectedMethodCallsError:
return
raise Exception('delete_container was called despite Retain policy')
# allows testing of the test directly, shown below
if __name__ == '__main__':
sys.argv.append(__file__)
nose.main()

View File

@ -4,7 +4,7 @@
"Description" : "Template to test S3 Bucket resources",
"Resources" : {
"S3Bucket" : {
"S3BucketWebsite" : {
"Type" : "AWS::S3::Bucket",
"Properties" : {
"AccessControl" : "PublicRead",
@ -12,19 +12,25 @@
"IndexDocument" : "index.html",
"ErrorDocument" : "error.html"
},
"DeletionPolicy" : "Delete"
"DeletionPolicy" : "Delete"
}
},
"S3Bucket" : {
"Type" : "AWS::S3::Bucket",
"Properties" : {
"AccessControl" : "Private"
}
}
},
"Outputs" : {
"WebsiteURL" : {
"WebsiteURL" : {
"Value" : { "Fn::GetAtt" : [ "S3Bucket", "WebsiteURL" ] },
"Description" : "URL for website hosted on S3"
},
"DomainName" : {
"Value" : { "Fn::GetAtt" : [ "S3Bucket", "DomainName" ] },
"Description" : "Domain of S3 host"
"DomainName" : {
"Value" : { "Fn::GetAtt" : [ "S3Bucket", "DomainName" ] },
"Description" : "Domain of S3 host"
}
}
}