# License for the specific language governing permissions and limitations
# under the License.
-
-import os
-
from heat.tests.v1_1 import fakes
from heat.engine.resources import instance as instances
from heat.common import template_format
-from heat.engine import parser
from heat.engine import scheduler
from heat.openstack.common import uuidutils
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
+
+
+nokey_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "NoKey Test",
+ "Parameters" : {},
+ "Resources" : {
+ "WebServer": {
+ "Type": "AWS::EC2::Instance",
+ "Properties": {
+ "ImageId" : "foo",
+ "InstanceType" : "m1.large",
+ "UserData" : "some data"
+ }
+ }
+ }
+}
+'''
class nokeyTest(HeatTestCase):
def setUp(self):
super(nokeyTest, self).setUp()
self.fc = fakes.FakeClient()
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
setup_dummy_db()
def test_nokey_create(self):
- f = open("%s/WordPress_NoKey.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
stack_name = 'instance_create_test_nokey_stack'
- template = parser.Template(t)
- params = parser.Parameters(stack_name, template, {})
- stack = parser.Stack(None, stack_name, template, params,
- stack_id=uuidutils.generate_uuid())
+ t = template_format.parse(nokey_template)
+ stack = parse_stack(t, stack_name=stack_name,
+ stack_id=uuidutils.generate_uuid())
t['Resources']['WebServer']['Properties']['ImageId'] = 'CentOS 5.2'
t['Resources']['WebServer']['Properties']['InstanceType'] = \
# under the License.
-import os
-
from testtools import skipIf
-from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.engine import properties
from heat.engine.resources.quantum import floatingip
from heat.engine.resources.quantum import port
from heat.engine.resources.quantum.quantum import QuantumResource as qr
-from heat.engine import parser
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
+
+quantum_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Template to test Quantum resources",
+ "Parameters" : {},
+ "Resources" : {
+ "network": {
+ "Type": "OS::Quantum::Net",
+ "Properties": {
+ "name": "the_network"
+ }
+ },
+ "unnamed_network": {
+ "Type": "OS::Quantum::Net"
+ },
+ "admin_down_network": {
+ "Type": "OS::Quantum::Net",
+ "Properties": {
+ "admin_state_up": false
+ }
+ },
+ "subnet": {
+ "Type": "OS::Quantum::Subnet",
+ "Properties": {
+ "network_id": { "Ref" : "network" },
+ "ip_version": 4,
+ "cidr": "10.0.3.0/24",
+ "allocation_pools": [{"start": "10.0.3.20", "end": "10.0.3.150"}]
+ }
+ },
+ "port": {
+ "Type": "OS::Quantum::Port",
+ "Properties": {
+ "device_id": "d6b4d3a5-c700-476f-b609-1493dd9dadc0",
+ "name": "port1",
+ "network_id": { "Ref" : "network" },
+ "fixed_ips": [{
+ "subnet_id": { "Ref" : "subnet" },
+ "ip_address": "10.0.3.21"
+ }]
+ }
+ },
+ "router": {
+ "Type": "OS::Quantum::Router"
+ },
+ "router_interface": {
+ "Type": "OS::Quantum::RouterInterface",
+ "Properties": {
+ "router_id": { "Ref" : "router" },
+ "subnet_id": { "Ref" : "subnet" }
+ }
+ }
+ }
+}
+'''
+
+quantum_floating_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Template to test Quantum resources",
+ "Parameters" : {},
+ "Resources" : {
+ "port_floating": {
+ "Type": "OS::Quantum::Port",
+ "Properties": {
+ "network_id": "xyz1234",
+ "fixed_ips": [{
+ "subnet_id": "12.12.12.0",
+ "ip_address": "10.0.0.10"
+ }]
+ }
+ },
+ "floating_ip": {
+ "Type": "OS::Quantum::FloatingIP",
+ "Properties": {
+ "floating_network_id": "abcd1234",
+ }
+ },
+ "floating_ip_assoc": {
+ "Type": "OS::Quantum::FloatingIPAssociation",
+ "Properties": {
+ "floatingip_id": { "Ref" : "floating_ip" },
+ "port_id": { "Ref" : "port_floating" }
+ }
+ }
+ }
+}
+'''
class FakeQuantum():
self.m.StubOutWithMock(net.Net, 'quantum')
setup_dummy_db()
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/Quantum.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- stack_name = 'test_stack'
- tmpl = parser.Template(t)
- params = parser.Parameters(stack_name, tmpl,
- {'external_network': 'abcd1234'})
- stack = parser.Stack(ctx, stack_name, tmpl, params)
-
- return stack
-
def create_net(self, t, stack, resource_name):
resource = net.Net('test_net', t['Resources'][resource_name], stack)
scheduler.TaskRunner(resource.create)()
net.Net.quantum().MultipleTimes().AndReturn(fq)
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(quantum_template)
+ stack = parse_stack(t)
resource = self.create_net(t, stack, 'network')
resource.validate()
self.m.StubOutWithMock(port.Port, 'quantum')
setup_dummy_db()
- def load_template(self, name='Quantum'):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/%s.template" % (self.path, name))
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- stack_name = 'test_stack'
- tmpl = parser.Template(t)
- params = parser.Parameters(stack_name, tmpl,
- {'external_network': 'abcd1234',
- 'internal_network': 'xyz1234',
- 'internal_subnet': '12.12.12.0'})
- stack = parser.Stack(ctx, stack_name, tmpl, params)
-
- return stack
-
def test_floating_ip(self):
if net.clients.quantumclient is None:
raise SkipTest
self.m.ReplayAll()
- t = self.load_template('Quantum_floating')
- stack = self.parse_stack(t)
+ t = template_format.parse(quantum_floating_template)
+ stack = parse_stack(t)
fip = stack['floating_ip']
scheduler.TaskRunner(fip.create)()
self.m.ReplayAll()
- t = self.load_template('Quantum_floating')
- stack = self.parse_stack(t)
+ t = template_format.parse(quantum_floating_template)
+ stack = parse_stack(t)
p = stack['port_floating']
scheduler.TaskRunner(p.create)()
self.m.ReplayAll()
- t = self.load_template('Quantum_floating')
- stack = self.parse_stack(t)
+ t = template_format.parse(quantum_floating_template)
+ stack = parse_stack(t)
fip = stack['floating_ip']
scheduler.TaskRunner(fip.create)()
# under the License.
-import os
import re
import mox
from testtools import skipIf
-from heat.common import context
from heat.common import template_format
from heat.openstack.common.importutils import try_import
from heat.engine.resources import s3
-from heat.engine import parser
from heat.engine import scheduler
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
swiftclient = try_import('swiftclient.client')
+swift_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Template to test S3 Bucket resources",
+ "Resources" : {
+ "S3BucketWebsite" : {
+ "Type" : "AWS::S3::Bucket",
+ "DeletionPolicy" : "Delete",
+ "Properties" : {
+ "AccessControl" : "PublicRead",
+ "WebsiteConfiguration" : {
+ "IndexDocument" : "index.html",
+ "ErrorDocument" : "error.html"
+ }
+ }
+ },
+ "S3Bucket" : {
+ "Type" : "AWS::S3::Bucket",
+ "Properties" : {
+ "AccessControl" : "Private"
+ }
+ }
+ }
+}
+'''
+
class s3Test(HeatTestCase):
@skipIf(swiftclient is None, 'unable to import swiftclient')
self.container_pattern = 'test_stack-test_resource-[0-9a-z]+'
setup_dummy_db()
- 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 = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- stack = parser.Stack(ctx, 'test_stack', parser.Template(t))
-
- return stack
-
def create_resource(self, t, stack, resource_name):
resource = s3.S3Bucket('test_resource',
t['Resources'][resource_name],
@skipIf(swiftclient is None, 'unable to import swiftclient')
def test_create_container_name(self):
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = s3.S3Bucket('test_resource',
t['Resources']['S3Bucket'],
stack)
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
ref_id = resource.FnGetRefId()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(swift_template)
properties = t['Resources']['S3Bucket']['Properties']
properties['AccessControl'] = 'PublicRead'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
self.m.VerifyAll()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(swift_template)
properties = t['Resources']['S3Bucket']['Properties']
properties['AccessControl'] = 'PublicReadWrite'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
self.m.VerifyAll()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(swift_template)
properties = t['Resources']['S3Bucket']['Properties']
properties['AccessControl'] = 'AuthenticatedRead'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
self.m.VerifyAll()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'S3BucketWebsite')
resource.delete()
self.m.VerifyAll()
swiftclient.ClientException('Test delete failure'))
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
resource.delete()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(swift_template)
bucket = t['Resources']['S3Bucket']
bucket['DeletionPolicy'] = 'Retain'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'S3Bucket')
# if delete_container is called, mox verify will succeed
resource.delete()
# under the License.
-import os
import re
import mox
from testtools import skipIf
-from heat.common import context
from heat.common import template_format
from heat.openstack.common.importutils import try_import
from heat.engine.resources import swift
-from heat.engine import parser
from heat.engine import scheduler
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
swiftclient = try_import('swiftclient.client')
+swift_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Template to test OS::Swift::Container resources",
+ "Resources" : {
+ "SwiftContainerWebsite" : {
+ "Type" : "OS::Swift::Container",
+ "DeletionPolicy" : "Delete",
+ "Properties" : {
+ "X-Container-Read" : ".r:*",
+ "X-Container-Meta" : {
+ "Web-Index" : "index.html",
+ "Web-Error" : "error.html"
+ }
+ }
+ },
+ "SwiftContainer" : {
+ "Type" : "OS::Swift::Container",
+ "Properties" : {
+ }
+ }
+ }
+}
+'''
+
class swiftTest(HeatTestCase):
@skipIf(swiftclient is None, 'unable to import swiftclient')
self.container_pattern = 'test_stack-test_resource-[0-9a-z]+'
setup_dummy_db()
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/Swift.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- stack = parser.Stack(ctx, 'test_stack', parser.Template(t))
-
- return stack
-
def create_resource(self, t, stack, resource_name):
resource = swift.SwiftContainer(
'test_resource',
@skipIf(swiftclient is None, 'unable to import swiftclient')
def test_create_container_name(self):
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = swift.SwiftContainer(
'test_resource',
t['Resources']['SwiftContainer'],
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'SwiftContainer')
ref_id = resource.FnGetRefId()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(swift_template)
properties = t['Resources']['SwiftContainer']['Properties']
properties['X-Container-Read'] = '.r:*'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'SwiftContainer')
resource.delete()
self.m.VerifyAll()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(swift_template)
properties = t['Resources']['SwiftContainer']['Properties']
properties['X-Container-Read'] = '.r:*'
properties['X-Container-Write'] = '.r:*'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'SwiftContainer')
resource.delete()
self.m.VerifyAll()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'SwiftContainerWebsite')
resource.delete()
self.m.VerifyAll()
swiftclient.ClientException('Test delete failure'))
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(swift_template)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'SwiftContainer')
resource.delete()
mox.Regex(self.container_pattern)).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(swift_template)
container = t['Resources']['SwiftContainer']
container['DeletionPolicy'] = 'Retain'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_resource(t, stack, 'SwiftContainer')
# if delete_container is called, mox verify will succeed
resource.delete()
# under the License.
-import os
-
import eventlet
from testtools import skipIf
-from heat.common import context
from heat.common import exception
from heat.common import template_format
-from heat.engine import parser
from heat.engine import scheduler
from heat.engine.resources import volume as vol
from heat.engine import clients
from heat.tests.common import HeatTestCase
from heat.tests.v1_1 import fakes
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
from cinderclient.v1 import client as cinderclient
volume_backups = try_import('cinderclient.v1.volume_backups')
+volume_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Volume Test",
+ "Parameters" : {},
+ "Resources" : {
+ "WikiDatabase": {
+ "Type": "AWS::EC2::Instance",
+ "Properties": {
+ "ImageId" : "foo",
+ "InstanceType" : "m1.large",
+ "KeyName" : "test",
+ "UserData" : "some data"
+ }
+ },
+ "DataVolume" : {
+ "Type" : "AWS::EC2::Volume",
+ "Properties" : {
+ "Size" : "1",
+ "AvailabilityZone" : "nova",
+ "Tags" : [{ "Key" : "Usage", "Value" : "Wiki Data Volume" }]
+ }
+ },
+ "MountPoint" : {
+ "Type" : "AWS::EC2::VolumeAttachment",
+ "Properties" : {
+ "InstanceId" : { "Ref" : "WikiDatabase" },
+ "VolumeId" : { "Ref" : "DataVolume" },
+ "Device" : "/dev/vdc"
+ }
+ }
+ }
+}
+'''
+
class VolumeTest(HeatTestCase):
def setUp(self):
self.m.StubOutWithMock(eventlet, 'sleep')
setup_dummy_db()
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/WordPress_2_Instances_With_EBS.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t, stack_name):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters(stack_name, template, {'KeyName': 'test'})
- stack = parser.Stack(ctx, stack_name, template, params)
-
- return stack
-
def create_volume(self, t, stack, resource_name):
resource = vol.Volume(resource_name,
t['Resources'][resource_name],
clients.cinder_exceptions.NotFound('Not found'))
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t, stack_name)
+ t = template_format.parse(volume_template)
+ stack = parse_stack(t, stack_name=stack_name)
resource = self.create_volume(t, stack, 'DataVolume')
self.assertEqual(fv.status, 'available')
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t, stack_name)
+ t = template_format.parse(volume_template)
+ stack = parse_stack(t, stack_name=stack_name)
resource = vol.Volume('DataVolume',
t['Resources']['DataVolume'],
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t, stack_name)
+ t = template_format.parse(volume_template)
+ stack = parse_stack(t, stack_name=stack_name)
scheduler.TaskRunner(stack['DataVolume'].create)()
self.assertEqual(fv.status, 'available')
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t, stack_name)
+ t = template_format.parse(volume_template)
+ stack = parse_stack(t, stack_name=stack_name)
scheduler.TaskRunner(stack['DataVolume'].create)()
self.assertEqual(fv.status, 'available')
self.cinder_fc.volumes.delete('vol-123').AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(volume_template)
t['Resources']['DataVolume']['DeletionPolicy'] = 'Snapshot'
- stack = self.parse_stack(t, stack_name)
+ stack = parse_stack(t, stack_name=stack_name)
resource = self.create_volume(t, stack, 'DataVolume')
eventlet.sleep(1).AndReturn(None)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(volume_template)
t['Resources']['DataVolume']['DeletionPolicy'] = 'Snapshot'
- stack = self.parse_stack(t, stack_name)
+ stack = parse_stack(t, stack_name=stack_name)
resource = self.create_volume(t, stack, 'DataVolume')
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(volume_template)
t['Resources']['DataVolume']['DeletionPolicy'] = 'Snapshot'
- stack = self.parse_stack(t, stack_name)
+ stack = parse_stack(t, stack_name=stack_name)
resource = vol.Volume('DataVolume',
t['Resources']['DataVolume'],
stack)
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(volume_template)
t['Resources']['DataVolume']['Properties']['SnapshotId'] = 'backup-123'
- stack = self.parse_stack(t, stack_name)
+ stack = parse_stack(t, stack_name=stack_name)
self.create_volume(t, stack, 'DataVolume')
self.assertEqual(fv.status, 'available')
Code Review / openstack-build / heat-build.git / commitdiff