<commit_before># 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 nose
import unittest
import mox
import json
import sqlalchemy
from nose.plugins.attrib import attr
from nose import with_setup

from heat.common import context
from heat.tests.v1_1 import fakes
import heat.engine.api as engine_api
from heat.engine import instance as instances
import heat.db as db_api
from heat.engine import parser
from heat.engine import manager
from heat.engine import auth
from heat.engine import watchrule


tests_dir = os.path.dirname(os.path.realpath(__file__))
templates_dir = os.path.normpath(os.path.join(tests_dir,
                                              os.path.pardir, os.path.pardir,
                                              'templates'))


def create_context(mocks, user='stacks_test_user',
                   tenant='test_admin', ctx=None):
    ctx = ctx or context.get_admin_context()
    mocks.StubOutWithMock(ctx, 'username')
    mocks.StubOutWithMock(ctx, 'tenant')
    ctx.username = user
    ctx.tenant = tenant
    mocks.StubOutWithMock(auth, 'authenticate')
    return ctx


def get_wordpress_stack(stack_name, ctx):
    tmpl_path = os.path.join(templates_dir,
                             'WordPress_Single_Instance_gold.template')
    with open(tmpl_path) as f:
        t = json.load(f)

    template = parser.Template(t)
    parameters = parser.Parameters(stack_name, template,
                                   {'KeyName': 'test'})

    stack = parser.Stack(ctx, stack_name, template, parameters)

    return stack


def setup_mocks(mocks, stack):
    fc = fakes.FakeClient()
    mocks.StubOutWithMock(instances.Instance, 'nova')
    instances.Instance.nova().MultipleTimes().AndReturn(fc)

    instance = stack.resources['WebServer']
    instance.calculate_properties()
    server_userdata = instance._build_userdata(instance.properties['UserData'])
    mocks.StubOutWithMock(fc.servers, 'create')
    fc.servers.create(image=744, flavor=3, key_name='test',
                      name='WebServer', security_groups=None,
                      userdata=server_userdata, scheduler_hints=None,
                      meta=None).AndReturn(fc.servers.list()[-1])


@attr(tag=['unit', 'stack'])
@attr(speed='slow')
class stackCreateTest(unittest.TestCase):
    def setUp(self):
        self.m = mox.Mox()

    def tearDown(self):
        self.m.UnsetStubs()
        print "stackTest teardown complete"

    def test_wordpress_single_instance_stack_create(self):
        stack = get_wordpress_stack('test_stack', create_context(self.m))
        setup_mocks(self.m, stack)
        self.m.ReplayAll()
        stack.store()
        stack.create()

        self.assertNotEqual(stack.resources['WebServer'], None)
        self.assertTrue(stack.resources['WebServer'].instance_id > 0)
        self.assertNotEqual(stack.resources['WebServer'].ipaddress, '0.0.0.0')

    def test_wordpress_single_instance_stack_delete(self):
        ctx = create_context(self.m, tenant='test_delete_tenant')
        stack = get_wordpress_stack('test_stack', ctx)
        setup_mocks(self.m, stack)
        self.m.ReplayAll()
        stack_id = stack.store()
        stack.create()

        db_s = db_api.stack_get(ctx, stack_id)
        self.assertNotEqual(db_s, None)

        self.assertNotEqual(stack.resources['WebServer'], None)
        self.assertTrue(stack.resources['WebServer'].instance_id > 0)

        stack.delete()

        self.assertEqual(stack.resources['WebServer'].state, 'DELETE_COMPLETE')
        self.assertEqual(db_api.stack_get(ctx, stack_id), None)
        self.assertEqual(db_s.status, 'DELETE_COMPLETE')


@attr(tag=['unit', 'engine-api', 'engine-manager'])
@attr(speed='fast')
class stackManagerTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        m = mox.Mox()
        cls.username = 'stack_manager_test_user'
        cls.tenant = 'stack_manager_test_tenant'
        ctx = create_context(m, cls.username, cls.tenant)
        cls.stack_name = 'manager_test_stack'

        stack = get_wordpress_stack(cls.stack_name, ctx)
        setup_mocks(m, stack)
        m.ReplayAll()

        stack.store()
        stack.create()
        cls.stack = stack
        cls.stack_identity = stack.identifier()

        m.UnsetStubs()

    @classmethod
    def tearDownClass(cls):
        cls = cls
        m = mox.Mox()
        create_context(m, cls.username, cls.tenant, ctx=cls.stack.context)
        setup_mocks(m, cls.stack)
        m.ReplayAll()

        cls.stack.delete()

        m.UnsetStubs()

    def setUp(self):
        self.m = mox.Mox()
        self.ctx = create_context(self.m, self.username, self.tenant)
        auth.authenticate(self.ctx).AndReturn(True)
        setup_mocks(self.m, self.stack)
        self.m.ReplayAll()

        self.man = manager.EngineManager()

    def tearDown(self):
        self.m.UnsetStubs()

    def test_stack_identify(self):
        identity = self.man.identify_stack(self.ctx, self.stack_name)
        self.assertEqual(identity, self.stack_identity)

    def test_stack_identify_nonexist(self):
        self.assertRaises(AttributeError, self.man.identify_stack,
                          self.ctx, 'wibble')
<commit_msg>Add unit tests for EngineManager create/update/delete_stack<commit_after># 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 nose
import unittest
import mox
import json
import sqlalchemy
from nose.plugins.attrib import attr
from nose import with_setup

from heat.common import context
from heat.tests.v1_1 import fakes
import heat.engine.api as engine_api
from heat.engine import instance as instances
import heat.db as db_api
from heat.engine import parser
from heat.engine import manager
from heat.engine import auth
from heat.engine import watchrule


tests_dir = os.path.dirname(os.path.realpath(__file__))
templates_dir = os.path.normpath(os.path.join(tests_dir,
                                              os.path.pardir, os.path.pardir,
                                              'templates'))


def create_context(mocks, user='stacks_test_user',
                   tenant='test_admin', ctx=None):
    ctx = ctx or context.get_admin_context()
    mocks.StubOutWithMock(ctx, 'username')
    mocks.StubOutWithMock(ctx, 'tenant')
    ctx.username = user
    ctx.tenant = tenant
    mocks.StubOutWithMock(auth, 'authenticate')
    return ctx


def get_wordpress_stack(stack_name, ctx):
    tmpl_path = os.path.join(templates_dir,
                             'WordPress_Single_Instance_gold.template')
    with open(tmpl_path) as f:
        t = json.load(f)

    template = parser.Template(t)
    parameters = parser.Parameters(stack_name, template,
                                   {'KeyName': 'test'})

    stack = parser.Stack(ctx, stack_name, template, parameters)

    return stack


def setup_mocks(mocks, stack):
    fc = fakes.FakeClient()
    mocks.StubOutWithMock(instances.Instance, 'nova')
    instances.Instance.nova().MultipleTimes().AndReturn(fc)

    instance = stack.resources['WebServer']
    instance.calculate_properties()
    server_userdata = instance._build_userdata(instance.properties['UserData'])
    mocks.StubOutWithMock(fc.servers, 'create')
    fc.servers.create(image=744, flavor=3, key_name='test',
                      name='WebServer', security_groups=None,
                      userdata=server_userdata, scheduler_hints=None,
                      meta=None).AndReturn(fc.servers.list()[-1])


@attr(tag=['unit', 'stack'])
@attr(speed='slow')
class stackCreateTest(unittest.TestCase):
    def setUp(self):
        self.m = mox.Mox()

    def tearDown(self):
        self.m.UnsetStubs()
        print "stackTest teardown complete"

    def test_wordpress_single_instance_stack_create(self):
        stack = get_wordpress_stack('test_stack', create_context(self.m))
        setup_mocks(self.m, stack)
        self.m.ReplayAll()
        stack.store()
        stack.create()

        self.assertNotEqual(stack.resources['WebServer'], None)
        self.assertTrue(stack.resources['WebServer'].instance_id > 0)
        self.assertNotEqual(stack.resources['WebServer'].ipaddress, '0.0.0.0')

    def test_wordpress_single_instance_stack_delete(self):
        ctx = create_context(self.m, tenant='test_delete_tenant')
        stack = get_wordpress_stack('test_stack', ctx)
        setup_mocks(self.m, stack)
        self.m.ReplayAll()
        stack_id = stack.store()
        stack.create()

        db_s = db_api.stack_get(ctx, stack_id)
        self.assertNotEqual(db_s, None)

        self.assertNotEqual(stack.resources['WebServer'], None)
        self.assertTrue(stack.resources['WebServer'].instance_id > 0)

        stack.delete()

        self.assertEqual(stack.resources['WebServer'].state, 'DELETE_COMPLETE')
        self.assertEqual(db_api.stack_get(ctx, stack_id), None)
        self.assertEqual(db_s.status, 'DELETE_COMPLETE')


@attr(tag=['unit', 'engine-api', 'engine-manager'])
@attr(speed='fast')
class stackManagerCreateUpdateDeleteTest(unittest.TestCase):

    def setUp(self):
        self.m = mox.Mox()
        self.username = 'stack_manager_create_test_user'
        self.tenant = 'stack_manager_create_test_tenant'
        self.ctx = create_context(self.m, self.username, self.tenant)
        auth.authenticate(self.ctx).AndReturn(True)

        self.man = manager.EngineManager()

    def tearDown(self):
        self.m.UnsetStubs()

    def test_stack_create(self):
        stack_name = 'manager_create_test_stack'
        params = {'foo': 'bar'}
        template = '{ "Template": "data" }'

        stack = get_wordpress_stack(stack_name, self.ctx)

        self.m.StubOutWithMock(parser, 'Template')
        self.m.StubOutWithMock(parser, 'Parameters')
        self.m.StubOutWithMock(parser, 'Stack')

        parser.Template(template).AndReturn(stack.t)
        parser.Parameters(stack_name,
                          stack.t,
                          params).AndReturn(stack.parameters)
        parser.Stack(self.ctx, stack.name,
                     stack.t, stack.parameters).AndReturn(stack)

        self.m.StubOutWithMock(stack, 'validate')
        stack.validate().AndReturn({'Description': 'Successfully validated'})

        self.m.StubOutWithMock(manager.greenpool, 'spawn_n')
        manager.greenpool.spawn_n(stack.create)

        self.m.ReplayAll()

        result = self.man.create_stack(self.ctx, stack_name,
                                       template, params, {})
        self.assertEqual(result, stack.identifier())
        self.assertTrue(isinstance(result, dict))
        self.assertTrue(result['stack_id'])
        self.m.VerifyAll()

    def test_stack_create_verify_err(self):
        stack_name = 'manager_create_verify_err_test_stack'
        params = {'foo': 'bar'}
        template = '{ "Template": "data" }'

        stack = get_wordpress_stack(stack_name, self.ctx)

        self.m.StubOutWithMock(parser, 'Template')
        self.m.StubOutWithMock(parser, 'Parameters')
        self.m.StubOutWithMock(parser, 'Stack')

        parser.Template(template).AndReturn(stack.t)
        parser.Parameters(stack_name,
                          stack.t,
                          params).AndReturn(stack.parameters)
        parser.Stack(self.ctx, stack.name,
                     stack.t, stack.parameters).AndReturn(stack)

        self.m.StubOutWithMock(stack, 'validate')
        error = {'Description': 'fubar'}
        stack.validate().AndReturn(error)

        self.m.StubOutWithMock(manager.greenpool, 'spawn_n')

        self.m.ReplayAll()

        result = self.man.create_stack(self.ctx, stack_name,
                                       template, params, {})
        self.assertEqual(result, error)
        self.m.VerifyAll()

    def test_stack_delete(self):
        stack_name = 'manager_delete_test_stack'
        stack = get_wordpress_stack(stack_name, self.ctx)
        stack.store()

        self.m.StubOutWithMock(parser.Stack, 'load')
        self.m.StubOutWithMock(manager.greenpool, 'spawn_n')

        parser.Stack.load(self.ctx, stack.id).AndReturn(stack)
        manager.greenpool.spawn_n(stack.delete)

        self.m.ReplayAll()

        self.assertEqual(self.man.delete_stack(self.ctx,
                                               stack.identifier(), {}),
                         None)
        self.m.VerifyAll()

    def test_stack_delete_nonexist(self):
        stack_name = 'manager_delete_nonexist_test_stack'
        stack = get_wordpress_stack(stack_name, self.ctx)

        self.m.ReplayAll()

        self.assertRaises(AttributeError,
                          self.man.delete_stack,
                          self.ctx, stack.identifier(), {})
        self.m.VerifyAll()

    def test_stack_update(self):
        stack_name = 'manager_update_test_stack'
        params = {'foo': 'bar'}
        template = '{ "Template": "data" }'

        old_stack = get_wordpress_stack(stack_name, self.ctx)
        old_stack.store()

        stack = get_wordpress_stack(stack_name, self.ctx)

        self.m.StubOutWithMock(parser, 'Stack')
        self.m.StubOutWithMock(parser.Stack, 'load')
        parser.Stack.load(self.ctx, old_stack.id).AndReturn(old_stack)

        self.m.StubOutWithMock(parser, 'Template')
        self.m.StubOutWithMock(parser, 'Parameters')

        parser.Template(template).AndReturn(stack.t)
        parser.Parameters(stack_name,
                          stack.t,
                          params).AndReturn(stack.parameters)
        parser.Stack(self.ctx, stack.name,
                     stack.t, stack.parameters).AndReturn(stack)

        self.m.StubOutWithMock(stack, 'validate')
        stack.validate().AndReturn({'Description': 'Successfully validated'})

        self.m.StubOutWithMock(manager.greenpool, 'spawn_n')
        manager.greenpool.spawn_n(old_stack.update, stack)

        self.m.ReplayAll()

        result = self.man.update_stack(self.ctx, old_stack.identifier(),
                                       template, params, {})
        self.assertEqual(result, old_stack.identifier())
        self.assertTrue(isinstance(result, dict))
        self.assertTrue(result['stack_id'])
        self.m.VerifyAll()

    def test_stack_update_verify_err(self):
        stack_name = 'manager_update_verify_err_test_stack'
        params = {'foo': 'bar'}
        template = '{ "Template": "data" }'

        old_stack = get_wordpress_stack(stack_name, self.ctx)
        old_stack.store()

        stack = get_wordpress_stack(stack_name, self.ctx)

        self.m.StubOutWithMock(parser, 'Stack')
        self.m.StubOutWithMock(parser.Stack, 'load')
        parser.Stack.load(self.ctx, old_stack.id).AndReturn(old_stack)

        self.m.StubOutWithMock(parser, 'Template')
        self.m.StubOutWithMock(parser, 'Parameters')

        parser.Template(template).AndReturn(stack.t)
        parser.Parameters(stack_name,
                          stack.t,
                          params).AndReturn(stack.parameters)
        parser.Stack(self.ctx, stack.name,
                     stack.t, stack.parameters).AndReturn(stack)

        self.m.StubOutWithMock(stack, 'validate')
        error = {'Description': 'fubar'}
        stack.validate().AndReturn(error)

        self.m.StubOutWithMock(manager.greenpool, 'spawn_n')

        self.m.ReplayAll()

        result = self.man.update_stack(self.ctx, old_stack.identifier(),
                                       template, params, {})
        self.assertEqual(result, error)
        self.m.VerifyAll()

    def test_stack_update_nonexist(self):
        stack_name = 'manager_update_nonexist_test_stack'
        params = {'foo': 'bar'}
        template = '{ "Template": "data" }'
        stack = get_wordpress_stack(stack_name, self.ctx)

        self.m.ReplayAll()

        self.assertRaises(AttributeError,
                          self.man.update_stack,
                          self.ctx, stack.identifier(), template, params, {})
        self.m.VerifyAll()


@attr(tag=['unit', 'engine-api', 'engine-manager'])
@attr(speed='fast')
class stackManagerTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        m = mox.Mox()
        cls.username = 'stack_manager_test_user'
        cls.tenant = 'stack_manager_test_tenant'
        ctx = create_context(m, cls.username, cls.tenant)
        cls.stack_name = 'manager_test_stack'

        stack = get_wordpress_stack(cls.stack_name, ctx)
        setup_mocks(m, stack)
        m.ReplayAll()

        stack.store()
        stack.create()
        cls.stack = stack
        cls.stack_identity = stack.identifier()

        m.UnsetStubs()

    @classmethod
    def tearDownClass(cls):
        cls = cls
        m = mox.Mox()
        create_context(m, cls.username, cls.tenant, ctx=cls.stack.context)
        setup_mocks(m, cls.stack)
        m.ReplayAll()

        cls.stack.delete()

        m.UnsetStubs()

    def setUp(self):
        self.m = mox.Mox()
        self.ctx = create_context(self.m, self.username, self.tenant)
        auth.authenticate(self.ctx).AndReturn(True)
        setup_mocks(self.m, self.stack)
        self.m.ReplayAll()

        self.man = manager.EngineManager()

    def tearDown(self):
        self.m.UnsetStubs()

    def test_stack_identify(self):
        identity = self.man.identify_stack(self.ctx, self.stack_name)
        self.assertEqual(identity, self.stack_identity)

    def test_stack_identify_nonexist(self):
        self.assertRaises(AttributeError, self.man.identify_stack,
                          self.ctx, 'wibble')

    def test_stack_create_existing(self):
        self.assertRaises(AttributeError, self.man.create_stack,
                          self.ctx, self.stack_name, self.stack.t, {}, {})

{'fraction_non_alphanumeric': 0.11281020159796815, 'fraction_numerical': 0.0021165141012752, 'mean_word_length': 2.6103151862464182, 'pattern_counts': {'":': 5, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6323937', 'n_tokens_mistral': 5743, 'n_tokens_neox': 5668, 'n_words': 1015}