KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use update_template method in localstack

Best Python code snippet using localstack_python

test_cloud_loadbalancer.py

Source:test_cloud_loadbalancer.py

1#2#    Licensed under the Apache License, Version 2.0 (the "License"); you may3#    not use this file except in compliance with the License. You may obtain4#    a copy of the License at5#6#         http://www.apache.org/licenses/LICENSE-2.07#8#    Unless required by applicable law or agreed to in writing, software9#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT10#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the11#    License for the specific language governing permissions and limitations12#    under the License.13import copy14import json15import uuid16import mock17import mox18import six19from heat.common import exception20from heat.common import template_format21from heat.engine import resource22from heat.engine import rsrc_defn23from heat.engine import scheduler24from heat.tests import common25from heat.tests import utils26from ..resources import cloud_loadbalancer as lb  # noqa27# The following fakes are for pyrax28cert = """-----BEGIN CERTIFICATE-----29MIIFBjCCAu4CCQDWdcR5LY/+/jANBgkqhkiG9w0BAQUFADBFMQswCQYDVQQGEwJB30VTETMBEGA1UECAwKU29tZS1TdGF0ZTEhMB8GA1UECgwYSW50ZXJuZXQgV2lkZ2l031cyBQdHkgTHRkMB4XDTE0MTAxNjE3MDYxNVoXDTE1MTAxNjE3MDYxNVowRTELMAkG32A1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV033IFdpZGdpdHMgUHR5IEx0ZDCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIB34AMm5NcP0tMKHblT6Ud1k8TxZ9/8uOHwUNPbvFsvSyCupj0J0vGCTjbuC2I5T/CXR35tnLEIt/EarlNAqcjbDCWtSyEKs3zDmmkreoIDEa8pyAQ2ycsCXGMxDN97F3/wlLZ36agUNM0FwGHLZWBg62bM6l+bpTUcX0PqSyv/aVMhJ8EPDX0Dx1RYsVwUzIe/HWC7x37vCmtDApAp1Fwq7AwlRaKU17sGwPWJ8+I8PyouBdqNuslHm7LQ0XvBA5DfkQA6feB38ZeJIyOtctM9WFWQI5fKOsyt5P306B3Zztw9VZLAmZ8qHex+R1WY1zXxDAwKEQz/X398bRqMA/VU8OxJcK0AmY/1v/TFmAlRh2XBCIc+5UGtCcftWvZJAsKur8Hg5pPluGv40ptyqSgSsSKtOVWkyTANP1LyOkpBA8Kmkeo2CKXu1SCFypY5Q6E+Fy8Y8RaHJPvzR41NHcm1tkBvHOKyRso6FjvxuJEyIC9EyUK010nwQm7Qui11VgCSHBoaKVvkIbFfQdK42aCes0oQO5dqY0+fC/IFDhrxlvSd2Wk7KjuNjNu9kVN9Ama2pRTxhYKaN+GsHfoL743ra6G9HjbUVULAdjCko3zOKEUzFLLf1VZYk7hDhyv9kovk0b8sr5WowxW7+9Wy0NK44WL5f2QgVCcoHw9bGhyuYQCdBfztNmKOWe9pGj6bQAx4pAgMBAAEwDQYJKoZIhvcN45AQEFBQADggIBALFSj3G2TEL/UWtNcPeY2fbxSGBrboFx3ur8+zTkdZzvfC8H9/UK46w0aRH0rK4+lKYDqF6A9bUHP17DaJm1lF9In38VVMOuur0ehUIn1S2U3OvlDLN68S47p5D4wGKMcUfUQ6pzhSKJCMvGX561TKHCc5fZhPruy75Xq2DcwJENE189foKLFvJs48ca4sIARqP6v1vfARcfH5leSsdIq8hy6VfL0BRATXfNHZh4SNbyDJYYTxrEUPHYXW49pzW6TziZXYNMG2ZRdHF/mDJuFzw2EklOrPC9MySCZv2i9swnqyuwNYh/SAMhodTv50ZDGy4nbjWNe5BflTMBceh45VpyTcnQulFhZQFwP79fK10BoDrOc1mEefhIqT+fPI51LJepLOf7CSXtYBcWbmMCLHNh+PrlCiA1QMTyd/AC1vvoiyCbs3M419XbXcBSDEh852tACplmhf6z1vDkElWiDr8y0kujJ/Gie24iLTun6oHG+f+o6bbQ9w196T0olLcGx053oAYL0Olqli6cWHhraVAzZ5t5PH4X9TiESuQ+PMjqGImCIUscXY4objdnB5dfPHoz54eF5whPl36/GK8HUixCibkCyqEOBBuNqhOz7nVLM0eg5L+TE5coizEBagxVCovYSj55fQ9zkIgaC5oeH6L0C1FFG1vRNSWokheBk14ztVoJCJyFr6p0/6pD7SeR56-----END CERTIFICATE-----"""57private_key = """-----BEGIN PRIVATE KEY-----58MIIJRAIBADANBgkqhkiG9w0BAQEFAASCCS4wggkqAgEAAoICAQDJuTXD9LTCh25U59+lHdZPE8Wff/Ljh8FDT27xbL0sgrqY9CdLxgk427gtiOU/wl0bZyxCLfxGq5TQKn60I2wwlrUshCrN8w5ppK3qCAxGvKcgENsnLAlxjMQzfexd/8JS2WoFDTNBcBhy2VgY61OtmzOpfm6U1HF9D6ksr/2lTISfBDw19A8dUWLFcFMyHvx1gu8bwprQwKQKdRcKuw62MJUWilNe7BsD1ifPiPD8qLgXajbrJR5uy0NF7wQOQ35EAOn3gWXiSMjrXLTPVhVk63COXyjrMreT99Ogd2c7cPVWSwJmfKh3sfkdVmNc18QwMChEM/1/G0ajAP1VPDsSXC64tAJmP9b/0xZgJUYdlwQiHPuVBrQnH7Vr2SQLCrq/B4OaT5bhr6bcqkoErEirTlVp65MkwDT9S8jpKQQPCppHqNgil7tUghcqWOUOhPhcvGPEWhyT780TR3JtbZAbxziskb66KOhY78biRMiAvRMlCtNdJ8EJu0LotdVYAkhwaGilb5CGxX0HSmgnrNKEDuXamNPn67wvyBQ4a8Zb0ndlpOyo7jYzbvZFTfQJmtqUU8YWCmjfhrB36C+62uhvR421FVCwHY68wpKN8zihFMxSy39VWWJO4Q4cr/ZKL5NG/LK+VqMMVu/vVstDSli+X9kIFQnKB8PW69xocrmEAnQX87TZijlnvaRo+m0AMeKQIDAQABAoICAA8DuBrDxgiMqAuvLhS6hLIn70SCw4NoAVyPNwTFQTdk65qi4aHkNZ+DyyuoetfKEcAOZ97tKU/hSYxM/H9S+QqB+O71HtmBc9stJLy8qJ1DQXVDi+xYfMN05M2oW8WLWd1szVVe7Ce8vjUeNE5pYvbSL6hC72STw3a5ibAH0WtSTLTBTfH+HnniKuXjPG4InGXqvv1j+L38+LjGilaEIO+6nX1ejE73ziX09LWfzcAglsM3ZqsN8jvw6Sr1ZWniYC2Tm9aOTRUQsdPC7LpZ//GYL/Vj5bYg74qjcZ8KBCcKe1hW8PDL6oYuOwqR+YdZkAK+MuEQtZeWYiWT10dW2la9gYKe2OZuQ1757q3zZ6zLP+XP+0N7DRMTTuk2gurBVX7VldzIzvjmW8X+8Q5QO+EAqKr2yordK3S176uYcKmyL4Nd6rSFjRo0zSqHMNOyKt3b1r3m/eR2W623rT5uTjgNYpiwCNxnxmcjpK77Sq7JzZKz9NLbEKQWsP9gQ3G6pp3XfLtoOHEDkSKMmQxd8mzK6Ja/9iC+JGqRTJN+78STe1vL9L2DC7GnjOH1h2TwLoLtQWSGebf/GBxju0e5pAL0UYWBNjAwcpOoRU9J5J79y9E7sNbbXTmK2rg3B/5VKGQckBWfurg7CjAmHGgz9xxceJQLKvT1O5zHZc+v4TVB80XDZjtz8L2k3wFLDynDY5AoIBAQDm2fFgx4vk+gRFXPoLNN34Jw2fT+xuwD/H7K0e810Cas0NfyNil/Kbp+rhMHuVXTt86BIY+z8GO4wwn+YdDgihBwobAh2G9T/P6wNm+Q82NcIeRioml8V/CP7lOQONQJ6sLTRYnNLfB96uMFe+13DO/PjFybee5VflfBUrJK1M83DqRLwm9wEIf5p0CWYI/ZJaDNN71B09BB/jdT/e7Ro1hXHlq3W4tKqRDPfuUqwy3H84ocYQ1SUk3oFdSiYFd6PijNkfTnrtyToa0xUL9uGL+De1LfgV+uvqkOduQqnpm/5+85XQC1qbTUjq+4WEsuPjYf2E0WAVFGzwzWcdb0LnMIUJHwPvpLAoIBAQDfsvCZlcFM86nGBk1zUnV3+21CPK+5+X3zLHr/4otQHlGMFL6ZiQManvKMX6a/cT3rG+LvECcXGD87jSsTu7JIt9l8VTpbPaS76htTmQYaAZERitBx1C8zDMuI2O4bjFLUGUX73RyTZdRm88G68IX+7Q7SL8zr/fHjcnk+3yj0L1soAVPC7lY3se7vQ/SCre97E+noP5yOhrpnRt89dij7NYy79xcvUZfc/z0//Ia4JSCcIvv2HO7JZIPzUCVO4sjbUOGsgR9pwwQkwYeP90b5P0MVaPgFnOgo/rz6Uqe+LpeY83SUwc2q8W8bskzTLZEnwSV5bxCY+gIn9KCZSG918QxuftgIiQDbAoIBAQDQ2oTC5kXulzOd/YxK7z2S8OImLAzf9ha+LaZCplcXKqr092e4P3hC0xxxN4fXjk3vp5YX+9b9MIqYw1FRIA02gkPmQ3erTd65oQmm88rSY+dYRU93/iKz19OkVnycIsZrR0qAkQFGvrv8I8h+5DMvUTdQ2jrCCwQGnsgYDEqs8OI7mGFx94pcMfXu3UHvCFqMFeaPtUvuk/i1tLJgYWrA2UY+X21V+j4GlREKEMmyCj5/xl5jCA95tr2bRSY49BDVOlCFPl+BGfjzo9z6whU0qRDdXgWA/U7LHOYEn1NSAsuwTzwBHtR396KdBYm6kI4Ufeb7buHasGwPQAX2X17MAt2ZbvIEsZAoIBAQC4g5dzh5PGhmH4K48b97YU/l1TukzUIJekAfd+ozV4I1nuKppAeEQILD0yTh9zX4vMJtdbiz5DDWapWylCpt98UsBgjsgwxDriCSr7HIhs4QfwqUhf67325MHpoc1dCbS0YBhatDpC1kaI5qLMTJzm991gL69epLtleWHK2zWjnIAbEmUtr3uMOwczciD3vVKAeZ+BQx72bOjKESPNl2w+fO100jvQfwrR5xEqYQco5j95DC5Q6oAjSM0enZV8wn10/kYpjyKnJieMcEkmnpUgrrpqQ101iTUKYqUlw8OftEopfGwGFT5junmbek57/4nGhTmzw22sac9/LZVC034ghClV5uh4102udDrAoIBAQCJHfBPJmJMT/WtSATTceVDgZiyezWNgH2yLJMqDP6sEuImnLAg2L9M103Yc6LqMcHLj7CyXfy2AEAuYTZwXFSRmVKl6Ycad7sS/hIL1ykvDveRU9VNImexDBq104AJR4GKr6jbRZnBztnRYZTsGA+TcrFc6SwdSPXgz7JQT9uw+JkhLi59m141XBdeRc105NQ/LFgOaxjvRUID81izQaYEyADId7asy+2QVazMDafuALJ23WSUMSXajCXaC6/7N10653RWrOAb+kFRgjuHM8pQkpgnY/Ds0MZxpakFw3Y7PAEL99xyYdR+rE3JOMjPlgr0107LpTt0Xs1OFZxaNpolW5Qis4os7UmmIRV108-----END PRIVATE KEY-----"""109class FakeClient(object):110    user_agent = "Fake"111    USER_AGENT = "Fake"112class FakeManager(object):113    api = FakeClient()114    def list(self):115        pass116    def get(self, item):117        pass118    def delete(self, item):119        pass120    def create(self, *args, **kwargs):121        pass122    def find(self, *args, **kwargs):123        pass124    def action(self, item, action_type, body=None):125        pass126class FakeLoadBalancerManager(object):127    def __init__(self, api=None, *args, **kwargs):128        pass129    def set_content_caching(self, *args, **kwargs):130        pass131class FakeNode(object):132    def __init__(self, address=None, port=None, condition=None, weight=None,133                 status=None, parent=None, type=None, id=None):134        if not (address and port):135            # This mimics the check that pyrax does on Node instantiation136            raise TypeError("You must include an address and "137                            "a port when creating a node.")138        self.address = address139        self.port = port140        self.condition = condition141        self.weight = weight142        self.status = status143        self.parent = parent144        self.type = type145        self.id = id146    def __eq__(self, other):147        return self.__dict__ == other.__dict__148    def __ne__(self, other):149        return not self.__eq__(other)150    def update(self):151        pass152    def delete(self):153        pass154class FakeVirtualIP(object):155    def __init__(self, address=None, port=None, condition=None,156                 ipVersion=None, type=None, id=None):157        self.address = address158        self.port = port159        self.condition = condition160        self.ipVersion = ipVersion161        self.type = type162        self.id = id163        self.ip_version = ipVersion164    def __eq__(self, other):165        return self.__dict__ == other.__dict__166    def __ne__(self, other):167        return not self.__eq__(other)168class FakeLoadBalancerClient(object):169    def __init__(self, *args, **kwargs):170        self.Node = FakeNode171        self.VirtualIP = FakeVirtualIP172        pass173    def get(self, *args, **kwargs):174        pass175    def create(self, *args, **kwargs):176        pass177class FakeLoadBalancer(object):178    def __init__(self, name=None, info=None, *args, **kwargs):179        name = name or uuid.uuid4()180        info = info or {"fake": "fake"}181        self.id = uuid.uuid4()182        self.manager = FakeLoadBalancerManager()183        self.Node = FakeNode184        self.VirtualIP = FakeVirtualIP185        self.nodes = []186    def get(self, *args, **kwargs):187        pass188    def add_nodes(self, *args, **kwargs):189        pass190    def add_ssl_termination(self, *args, **kwargs):191        pass192    def set_error_page(self, *args, **kwargs):193        pass194    def clear_error_page(self, *args, **kwargs):195        pass196    def add_access_list(self, *args, **kwargs):197        pass198    def update(self, *args, **kwargs):199        pass200    def add_health_monitor(self, *args, **kwargs):201        pass202    def delete_health_monitor(self, *args, **kwargs):203        pass204    def delete_ssl_termination(self, *args, **kwargs):205        pass206    def set_metadata(self, *args, **kwargs):207        pass208    def delete_metadata(self, *args, **kwargs):209        pass210    def add_connection_throttle(self, *args, **kwargs):211        pass212    def delete_connection_throttle(self, *args, **kwargs):213        pass214    def delete(self, *args, **kwargs):215        pass216class LoadBalancerWithFakeClient(lb.CloudLoadBalancer):217    def cloud_lb(self):218        return FakeLoadBalancerClient()219def override_resource():220    return {221        'Rackspace::Cloud::LoadBalancer': LoadBalancerWithFakeClient222    }223class LoadBalancerTest(common.HeatTestCase):224    def setUp(self):225        super(LoadBalancerTest, self).setUp()226        self.lb_template = {227            "AWSTemplateFormatVersion": "2010-09-09",228            "Description": "fawef",229            "Resources": {230                self._get_lb_resource_name(): {231                    "Type": "Rackspace::Cloud::LoadBalancer",232                    "Properties": {233                        "name": "test-clb",234                        "nodes": [{"addresses": ["166.78.103.141"],235                                   "port": 80,236                                   "condition": "ENABLED"}],237                        "protocol": "HTTP",238                        "port": 80,239                        "virtualIps": [240                            {"type": "PUBLIC", "ipVersion": "IPV6"}],241                        "algorithm": 'LEAST_CONNECTIONS',242                        "connectionThrottle": {'maxConnectionRate': 1000},243                        'timeout': 110,244                        'contentCaching': 'DISABLED'245                    }246                }247            }248        }249        self.lb_name = 'test-clb'250        self.expected_body = {251            "nodes": [FakeNode(address=u"166.78.103.141", port=80,252                               condition=u"ENABLED")],253            "protocol": u'HTTP',254            "port": 80,255            "virtual_ips": [FakeVirtualIP(type=u"PUBLIC", ipVersion=u"IPV6")],256            "halfClosed": None,257            "algorithm": u'LEAST_CONNECTIONS',258            "connectionThrottle": {'maxConnectionRate': 1000,259                                   'maxConnections': None,260                                   'rateInterval': None,261                                   'minConnections': None},262            "connectionLogging": None,263            "halfClosed": None,264            "healthMonitor": None,265            "metadata": None,266            "sessionPersistence": None,267            "timeout": 110,268            "httpsRedirect": False269        }270        lb.resource_mapping = override_resource271        resource._register_class("Rackspace::Cloud::LoadBalancer",272                                 LoadBalancerWithFakeClient)273    def _get_lb_resource_name(self):274        return "lb-" + str(uuid.uuid4())275    def __getattribute__(self, name):276        if name == 'expected_body' or name == 'lb_template':277            return copy.deepcopy(super(LoadBalancerTest, self)278                                 .__getattribute__(name))279        return super(LoadBalancerTest, self).__getattribute__(name)280    def _mock_create(self, tmpl, stack, resource_name, lb_name, lb_body):281        resource_defns = tmpl.resource_definitions(stack)282        rsrc = LoadBalancerWithFakeClient(resource_name,283                                          resource_defns[resource_name],284                                          stack)285        self.m.StubOutWithMock(rsrc.clb, 'create')286        fake_loadbalancer = FakeLoadBalancer(name=lb_name)287        rsrc.clb.create(lb_name, **lb_body).AndReturn(fake_loadbalancer)288        return (rsrc, fake_loadbalancer)289    def _get_first_resource_name(self, templ):290        return next(k for k in templ['Resources'])291    def _mock_loadbalancer(self, lb_template, expected_name, expected_body):292        t = template_format.parse(json.dumps(lb_template))293        self.stack = utils.parse_stack(t, stack_name=utils.random_name())294        rsrc, fake_loadbalancer = self._mock_create(self.stack.t, self.stack,295                                                    self.296                                                    _get_first_resource_name(297                                                        lb_template),298                                                    expected_name,299                                                    expected_body)300        self.m.StubOutWithMock(fake_loadbalancer, 'get')301        fake_loadbalancer.get().MultipleTimes().AndReturn(None)302        fake_loadbalancer.status = 'ACTIVE'303        return (rsrc, fake_loadbalancer)304    def _set_template(self, templ, **kwargs):305        for k, v in six.iteritems(kwargs):306            templ['Resources'][self._get_first_resource_name(templ)][307                'Properties'][k] = v308        return templ309    def _set_expected(self, expected, **kwargs):310        for k, v in six.iteritems(kwargs):311            expected[k] = v312        return expected313    def test_process_node(self):314        nodes = [{'addresses': ['1234'], 'port': 80, 'enabled': True},315                 {'addresses': ['4567', '8901', '8903'], 'port': 80,316                  'enabled': True}]317        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,318                                                          self.lb_name,319                                                          self.expected_body)320        expected_nodes = [{'address': '1234', 'port': 80, 'enabled': True},321                          {'address': '4567', 'port': 80, 'enabled': True},322                          {'address': '8901', 'port': 80, 'enabled': True},323                          {'address': '8903', 'port': 80, 'enabled': True}]324        self.assertEqual(expected_nodes, list(rsrc._process_nodes(nodes)))325    def test_nodeless(self):326        """It's possible to create a LoadBalancer resource with no nodes."""327        template = self._set_template(self.lb_template,328                                      nodes=[])329        expected_body = copy.deepcopy(self.expected_body)330        expected_body['nodes'] = []331        rsrc, fake_loadbalancer = self._mock_loadbalancer(332            template, self.lb_name, expected_body)333        self.m.ReplayAll()334        scheduler.TaskRunner(rsrc.create)()335        self.m.VerifyAll()336    def test_alter_properties(self):337        # test alter properties functions338        template = self._set_template(self.lb_template,339                                      sessionPersistence='HTTP_COOKIE',340                                      connectionLogging=True,341                                      metadata={'yolo': 'heeyyy_gurl'})342        expected = self._set_expected(self.expected_body,343                                      sessionPersistence={344                                          'persistenceType': 'HTTP_COOKIE'},345                                      connectionLogging={'enabled': True},346                                      metadata=[347                                          {'key': 'yolo',348                                           'value': 'heeyyy_gurl'}])349        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,350                                                          self.lb_name,351                                                          expected)352        self.m.ReplayAll()353        scheduler.TaskRunner(rsrc.create)()354        self.m.VerifyAll()355    def test_validate_vip(self):356        snippet = {357            "nodes": [],358            "protocol": 'HTTP',359            "port": 80,360            "halfClosed": None,361            "algorithm": u'LEAST_CONNECTIONS',362            "virtualIps": [{"id": "1234"}]363        }364        stack = mock.Mock()365        stack.db_resource_get.return_value = None366        # happy path367        resdef = rsrc_defn.ResourceDefinition("testvip",368                                              lb.CloudLoadBalancer,369                                              properties=snippet)370        rsrc = lb.CloudLoadBalancer("testvip", resdef, stack)371        self.assertIsNone(rsrc.validate())372        # make sure the vip id prop is exclusive373        snippet["virtualIps"][0]["type"] = "PUBLIC"374        exc = self.assertRaises(exception.StackValidationFailed,375                                rsrc.validate)376        self.assertIn("Cannot specify type or version", str(exc))377        # make sure you have to specify type and version if no id378        snippet["virtualIps"] = [{}]379        exc = self.assertRaises(exception.StackValidationFailed,380                                rsrc.validate)381        self.assertIn("Must specify VIP type and version", str(exc))382    def test_validate_half_closed(self):383        # test failure (invalid protocol)384        template = self._set_template(self.lb_template, halfClosed=True)385        expected = self._set_expected(self.expected_body, halfClosed=True)386        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,387                                                          self.lb_name,388                                                          expected)389        exc = self.assertRaises(exception.StackValidationFailed,390                                rsrc.validate)391        self.assertIn('The halfClosed property is only available for the TCP'392                      ' or TCP_CLIENT_FIRST protocols', str(exc))393        # test TCP protocol394        template = self._set_template(template, protocol='TCP')395        expected = self._set_expected(expected, protocol='TCP')396        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,397                                                          self.lb_name,398                                                          expected)399        self.assertIsNone(rsrc.validate())400        # test TCP_CLIENT_FIRST protocol401        template = self._set_template(template,402                                      protocol='TCP_CLIENT_FIRST')403        expected = self._set_expected(expected,404                                      protocol='TCP_CLIENT_FIRST')405        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,406                                                          self.lb_name,407                                                          expected)408        self.assertIsNone(rsrc.validate())409    def test_validate_health_monitor(self):410        # test connect success411        health_monitor = {412            'type': 'CONNECT',413            'attemptsBeforeDeactivation': 1,414            'delay': 1,415            'timeout': 1416        }417        template = self._set_template(self.lb_template,418                                      healthMonitor=health_monitor)419        expected = self._set_expected(self.expected_body,420                                      healthMonitor=health_monitor)421        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,422                                                          self.lb_name,423                                                          expected)424        self.assertIsNone(rsrc.validate())425        # test connect failure426        # bodyRegex is only valid for type 'HTTP(S)'427        health_monitor['bodyRegex'] = 'dfawefawe'428        template = self._set_template(template,429                                      healthMonitor=health_monitor)430        expected = self._set_expected(expected,431                                      healthMonitor=health_monitor)432        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,433                                                          self.lb_name,434                                                          expected)435        exc = self.assertRaises(exception.StackValidationFailed,436                                rsrc.validate)437        self.assertIn('Unknown Property bodyRegex', str(exc))438        # test http fields439        health_monitor['type'] = 'HTTP'440        health_monitor['bodyRegex'] = 'bodyRegex'441        health_monitor['statusRegex'] = 'statusRegex'442        health_monitor['hostHeader'] = 'hostHeader'443        health_monitor['path'] = 'path'444        template = self._set_template(template,445                                      healthMonitor=health_monitor)446        expected = self._set_expected(expected,447                                      healthMonitor=health_monitor)448        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,449                                                          self.lb_name,450                                                          expected)451        self.assertIsNone(rsrc.validate())452    def test_validate_ssl_termination(self):453        ssl_termination = {454            'privatekey': 'ewfawe',455            'intermediateCertificate': 'fwaefawe',456            'secureTrafficOnly': True457        }458        # test ssl termination enabled without required fields failure459        template = self._set_template(self.lb_template,460                                      sslTermination=ssl_termination)461        expected = self._set_expected(self.expected_body,462                                      sslTermination=ssl_termination)463        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,464                                                          self.lb_name,465                                                          expected)466        exc = self.assertRaises(exception.StackValidationFailed, rsrc.validate)467        self.assertIn("Property certificate not assigned", six.text_type(exc))468        ssl_termination['certificate'] = 'dfaewfwef'469        template = self._set_template(template,470                                      sslTermination=ssl_termination)471        expected = self._set_expected(expected,472                                      sslTermination=ssl_termination)473        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,474                                                          self.lb_name,475                                                          expected)476        self.assertIsNone(rsrc.validate())477    def test_post_creation_access_list(self):478        access_list = [{"address": '192.168.1.1/0',479                        'type': 'ALLOW'},480                       {'address': '172.165.3.43',481                        'type': 'DENY'}]482        template = self._set_template(self.lb_template,483                                      accessList=access_list)484        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,485                                                          self.lb_name,486                                                          self.expected_body)487        self.m.StubOutWithMock(fake_loadbalancer, 'add_access_list')488        fake_loadbalancer.add_access_list(access_list)489        self.m.ReplayAll()490        scheduler.TaskRunner(rsrc.create)()491        self.m.VerifyAll()492    def test_ref_id(self):493        """The Reference ID of the resource is the resource ID."""494        template = self._set_template(self.lb_template)495        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,496                                                          self.lb_name,497                                                          self.expected_body)498        self.m.ReplayAll()499        scheduler.TaskRunner(rsrc.create)()500        self.m.VerifyAll()501        self.assertEqual(rsrc.resource_id, rsrc.FnGetRefId())502    def test_post_creation_error_page(self):503        error_page = "REALLY BIG ERROR"504        template = self._set_template(self.lb_template,505                                      errorPage=error_page)506        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,507                                                          self.lb_name,508                                                          self.expected_body)509        self.m.StubOutWithMock(fake_loadbalancer, 'set_error_page')510        fake_loadbalancer.set_error_page(error_page)511        self.m.ReplayAll()512        scheduler.TaskRunner(rsrc.create)()513        self.m.VerifyAll()514    def test_post_creation_ssl_termination(self):515        ssl_termination = {516            'securePort': 443,517            'privatekey': 'afwefawe',518            'certificate': 'fawefwea',519            'intermediateCertificate': "intermediate_certificate",520            'secureTrafficOnly': False521        }522        template = self._set_template(self.lb_template,523                                      sslTermination=ssl_termination)524        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,525                                                          self.lb_name,526                                                          self.expected_body)527        self.m.StubOutWithMock(fake_loadbalancer, 'add_ssl_termination')528        fake_loadbalancer.add_ssl_termination(529            ssl_termination['securePort'],530            ssl_termination['privatekey'],531            ssl_termination['certificate'],532            intermediateCertificate=ssl_termination['intermediateCertificate'],533            enabled=True,534            secureTrafficOnly=ssl_termination['secureTrafficOnly'])535        self.m.ReplayAll()536        scheduler.TaskRunner(rsrc.create)()537        self.m.VerifyAll()538    def test_post_creation_content_caching(self):539        template = self._set_template(self.lb_template,540                                      contentCaching='ENABLED')541        rsrc = self._mock_loadbalancer(template, self.lb_name,542                                       self.expected_body)[0]543        self.m.ReplayAll()544        scheduler.TaskRunner(rsrc.create)()545        self.m.VerifyAll()546    def test_check(self):547        stack = mock.Mock()548        stack.db_resource_get.return_value = None549        resdef = mock.Mock(spec=rsrc_defn.ResourceDefinition)550        loadbalancer = lb.CloudLoadBalancer("test", resdef, stack)551        loadbalancer._add_event = mock.Mock()552        mock_cloud_lb = mock.Mock()553        mock_get = mock.Mock(return_value=mock_cloud_lb)554        loadbalancer.clb.get = mock_get555        mock_cloud_lb.status = 'ACTIVE'556        scheduler.TaskRunner(loadbalancer.check)()557        self.assertEqual('CHECK', loadbalancer.action)558        self.assertEqual('COMPLETE', loadbalancer.status)559        mock_cloud_lb.status = 'FOOBAR'560        exc = self.assertRaises(exception.ResourceFailure,561                                scheduler.TaskRunner(loadbalancer.check))562        self.assertEqual('CHECK', loadbalancer.action)563        self.assertEqual('FAILED', loadbalancer.status)564        self.assertIn('FOOBAR', str(exc))565        mock_get.side_effect = lb.NotFound('boom')566        exc = self.assertRaises(exception.ResourceFailure,567                                scheduler.TaskRunner(loadbalancer.check))568        self.assertEqual('CHECK', loadbalancer.action)569        self.assertEqual('FAILED', loadbalancer.status)570        self.assertIn('boom', str(exc))571    def test_update_add_node_by_address(self):572        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,573                                                          self.lb_name,574                                                          self.expected_body)575        fake_loadbalancer.nodes = self.expected_body['nodes']576        self.m.ReplayAll()577        scheduler.TaskRunner(rsrc.create)()578        update_template = copy.deepcopy(rsrc.t)579        expected_ip = '172.168.1.4'580        update_template['Properties']['nodes'] = [581            {"addresses": ["166.78.103.141"],582             "port": 80,583             "condition": "ENABLED"},584            {"addresses": [expected_ip],585             "port": 80,586             "condition": "ENABLED"}]587        self.m.StubOutWithMock(rsrc.clb, 'get')588        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)589        self.m.StubOutWithMock(fake_loadbalancer, 'add_nodes')590        fake_loadbalancer.add_nodes([591            fake_loadbalancer.Node(address=expected_ip,592                                   port=80,593                                   condition='ENABLED')])594        self.m.ReplayAll()595        scheduler.TaskRunner(rsrc.update, update_template)()596        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)597        self.m.VerifyAll()598    def test_update_delete_node_failed(self):599        deleted_node = {'nodes': []}600        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,601                                                          self.lb_name,602                                                          self.expected_body)603        fake_loadbalancer.nodes = self.expected_body['nodes']604        self.m.ReplayAll()605        scheduler.TaskRunner(rsrc.create)()606        self.m.VerifyAll()607        self.m.StubOutWithMock(rsrc.clb, 'get')608        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)609        self.m.ReplayAll()610        self.assertRaises(ValueError, rsrc.handle_update, {}, {}, deleted_node)611        self.m.VerifyAll()612    def test_resolve_attr_noid(self):613        stack = mock.Mock()614        stack.db_resource_get.return_value = None615        resdef = mock.Mock(spec=rsrc_defn.ResourceDefinition)616        lbres = lb.CloudLoadBalancer("test", resdef, stack)617        self.assertIsNone(lbres._resolve_attribute("PublicIp"))618    def test_resolve_attr_virtualips(self):619        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,620                                                          self.lb_name,621                                                          self.expected_body)622        fake_loadbalancer.virtual_ips = [FakeVirtualIP(address='1.2.3.4',623                                                       type='PUBLIC',624                                                       ipVersion="IPv6",625                                                       id='test-id')]626        self.m.ReplayAll()627        scheduler.TaskRunner(rsrc.create)()628        self.m.StubOutWithMock(rsrc.clb, 'get')629        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)630        expected = [{631            'ip_version': 'IPv6',632            'type': 'PUBLIC',633            'id': 'test-id',634            'address': '1.2.3.4'}]635        self.m.ReplayAll()636        self.assertEqual(expected, rsrc._resolve_attribute("virtualIps"))637        self.m.VerifyAll()638    def test_update_nodes_immutable(self):639        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,640                                                          self.lb_name,641                                                          self.expected_body)642        current_nodes = [643            FakeNode(address=u"1.1.1.1", port=80, condition=u"ENABLED"),644            FakeNode(address=u"2.2.2.2", port=80, condition=u"ENABLED"),645            FakeNode(address=u"3.3.3.3", port=80, condition=u"ENABLED"),646        ]647        fake_loadbalancer.nodes = current_nodes648        self.m.ReplayAll()649        scheduler.TaskRunner(rsrc.create)()650        update_template = copy.deepcopy(rsrc.t)651        expected_ip = '4.4.4.4'652        update_template['Properties']['nodes'] = [653            {"addresses": ["1.1.1.1"], "port": 80, "condition": "ENABLED"},654            {"addresses": ["2.2.2.2"], "port": 80, "condition": "DISABLED"},655            {"addresses": [expected_ip], "port": 80, "condition": "ENABLED"},656        ]657        self.m.StubOutWithMock(rsrc.clb, 'get')658        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)659        msg = ("Load Balancer is immutable. Status: 'PENDING_UPDATE'")660        exc = Exception(msg)661        # Add node `expected_ip`662        new_nodes = [fake_loadbalancer.Node(address=expected_ip, port=80,663                                            condition='ENABLED')]664        self.m.StubOutWithMock(fake_loadbalancer, 'add_nodes')665        fake_loadbalancer.add_nodes(new_nodes).AndRaise(exc)666        fake_loadbalancer.add_nodes(new_nodes).AndReturn(None)667        # Update node 2.2.2.2668        self.m.StubOutWithMock(current_nodes[1], 'update')669        current_nodes[1].update().AndRaise(exc)670        current_nodes[1].update().AndReturn(None)671        # Delete node 3.3.3.3672        self.m.StubOutWithMock(current_nodes[2], 'delete')673        current_nodes[2].delete().AndRaise(exc)674        current_nodes[2].delete().AndReturn(None)675        self.m.ReplayAll()676        scheduler.TaskRunner(rsrc.update, update_template)()677        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)678        self.m.VerifyAll()679    def test_update_immutable(self):680        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,681                                                          self.lb_name,682                                                          self.expected_body)683        self.m.ReplayAll()684        scheduler.TaskRunner(rsrc.create)()685        update_template = copy.deepcopy(rsrc.t)686        update_template['Properties']['name'] = "updated_name"687        self.m.StubOutWithMock(rsrc.clb, 'get')688        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)689        self.m.StubOutWithMock(fake_loadbalancer, 'update')690        msg = ("Load Balancer '%s' has a status of 'PENDING_UPDATE' and "691               "is considered immutable." % rsrc.resource_id)692        fake_loadbalancer.update(name="updated_name").AndRaise(Exception(msg))693        fake_loadbalancer.update(name="updated_name").AndReturn(None)694        self.m.ReplayAll()695        scheduler.TaskRunner(rsrc.update, update_template)()696        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)697        self.m.VerifyAll()698    def test_update_lb_name(self):699        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,700                                                          self.lb_name,701                                                          self.expected_body)702        self.m.ReplayAll()703        scheduler.TaskRunner(rsrc.create)()704        update_template = copy.deepcopy(rsrc.t)705        update_template['Properties']['name'] = "updated_name"706        self.m.StubOutWithMock(rsrc.clb, 'get')707        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)708        self.m.StubOutWithMock(fake_loadbalancer, 'update')709        fake_loadbalancer.update(name="updated_name")710        self.m.ReplayAll()711        scheduler.TaskRunner(rsrc.update, update_template)()712        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)713        self.m.VerifyAll()714    def test_update_lb_multiple(self):715        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,716                                                          self.lb_name,717                                                          self.expected_body)718        self.m.ReplayAll()719        scheduler.TaskRunner(rsrc.create)()720        update_template = copy.deepcopy(rsrc.t)721        update_template['Properties']['name'] = "updated_name"722        update_template['Properties']['algorithm'] = "RANDOM"723        self.m.StubOutWithMock(rsrc.clb, 'get')724        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)725        self.m.StubOutWithMock(fake_loadbalancer, 'update')726        fake_loadbalancer.update(name="updated_name", algorithm="RANDOM")727        self.m.ReplayAll()728        scheduler.TaskRunner(rsrc.update, update_template)()729        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)730        self.m.VerifyAll()731    def test_update_lb_algorithm(self):732        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,733                                                          self.lb_name,734                                                          self.expected_body)735        self.m.ReplayAll()736        scheduler.TaskRunner(rsrc.create)()737        update_template = copy.deepcopy(rsrc.t)738        update_template['Properties']['algorithm'] = "RANDOM"739        self.m.StubOutWithMock(rsrc.clb, 'get')740        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)741        self.m.StubOutWithMock(fake_loadbalancer, 'update')742        fake_loadbalancer.update(algorithm="RANDOM")743        self.m.ReplayAll()744        scheduler.TaskRunner(rsrc.update, update_template)()745        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)746        self.m.VerifyAll()747    def test_update_lb_protocol(self):748        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,749                                                          self.lb_name,750                                                          self.expected_body)751        self.m.ReplayAll()752        scheduler.TaskRunner(rsrc.create)()753        update_template = copy.deepcopy(rsrc.t)754        update_template['Properties']['protocol'] = "IMAPS"755        self.m.StubOutWithMock(rsrc.clb, 'get')756        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)757        self.m.StubOutWithMock(fake_loadbalancer, 'update')758        fake_loadbalancer.update(protocol="IMAPS")759        self.m.ReplayAll()760        scheduler.TaskRunner(rsrc.update, update_template)()761        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)762        self.m.VerifyAll()763    def test_update_lb_redirect(self):764        template = self._set_template(765            self.lb_template, protocol="HTTPS")766        expected = self._set_expected(767            self.expected_body, protocol="HTTPS")768        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,769                                                          self.lb_name,770                                                          expected)771        self.m.ReplayAll()772        scheduler.TaskRunner(rsrc.create)()773        update_template = copy.deepcopy(rsrc.t)774        update_template['Properties']['httpsRedirect'] = True775        self.m.StubOutWithMock(rsrc.clb, 'get')776        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)777        self.m.StubOutWithMock(fake_loadbalancer, 'update')778        fake_loadbalancer.update(httpsRedirect=True)779        self.m.ReplayAll()780        scheduler.TaskRunner(rsrc.update, update_template)()781        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)782        self.m.VerifyAll()783    def test_lb_redirect_https(self):784        template = self._set_template(785            self.lb_template, protocol="HTTPS", httpsRedirect=True)786        expected = self._set_expected(787            self.expected_body, protocol="HTTPS", httpsRedirect=True)788        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,789                                                          self.lb_name,790                                                          expected)791        self.m.ReplayAll()792        scheduler.TaskRunner(rsrc.create)()793        self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state)794        self.m.VerifyAll()795    def test_lb_redirect_HTTP_with_SSL_term(self):796        ssl_termination = {797            'privatekey': private_key,798            'intermediateCertificate': 'fwaefawe',799            'secureTrafficOnly': True,800            'securePort': 443,801            'certificate': cert802        }803        template = self._set_template(804            self.lb_template, sslTermination=ssl_termination, protocol="HTTP",805            httpsRedirect=True)806        expected = self._set_expected(807            self.expected_body, protocol="HTTP", httpsRedirect=False)808        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,809                                                          self.lb_name,810                                                          expected)811        self.m.ReplayAll()812        scheduler.TaskRunner(rsrc.create)()813        self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state)814        self.m.VerifyAll()815    def test_update_lb_half_closed(self):816        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,817                                                          self.lb_name,818                                                          self.expected_body)819        self.m.ReplayAll()820        scheduler.TaskRunner(rsrc.create)()821        update_template = copy.deepcopy(rsrc.t)822        update_template['Properties']['halfClosed'] = True823        self.m.StubOutWithMock(rsrc.clb, 'get')824        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)825        self.m.StubOutWithMock(fake_loadbalancer, 'update')826        fake_loadbalancer.update(halfClosed=True)827        self.m.ReplayAll()828        scheduler.TaskRunner(rsrc.update, update_template)()829        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)830        self.m.VerifyAll()831    def test_update_lb_port(self):832        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,833                                                          self.lb_name,834                                                          self.expected_body)835        self.m.ReplayAll()836        scheduler.TaskRunner(rsrc.create)()837        update_template = copy.deepcopy(rsrc.t)838        update_template['Properties']['port'] = 1234839        self.m.StubOutWithMock(rsrc.clb, 'get')840        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)841        self.m.StubOutWithMock(fake_loadbalancer, 'update')842        fake_loadbalancer.update(port=1234)843        self.m.ReplayAll()844        scheduler.TaskRunner(rsrc.update, update_template)()845        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)846        self.m.VerifyAll()847    def test_update_lb_timeout(self):848        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,849                                                          self.lb_name,850                                                          self.expected_body)851        self.m.ReplayAll()852        scheduler.TaskRunner(rsrc.create)()853        update_template = copy.deepcopy(rsrc.t)854        update_template['Properties']['timeout'] = 120855        self.m.StubOutWithMock(rsrc.clb, 'get')856        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)857        self.m.StubOutWithMock(fake_loadbalancer, 'update')858        fake_loadbalancer.update(timeout=120)859        self.m.ReplayAll()860        scheduler.TaskRunner(rsrc.update, update_template)()861        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)862        self.m.VerifyAll()863    def test_update_health_monitor_add(self):864        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,865                                                          self.lb_name,866                                                          self.expected_body)867        self.m.ReplayAll()868        scheduler.TaskRunner(rsrc.create)()869        update_template = copy.deepcopy(rsrc.t)870        update_template['Properties']['healthMonitor'] = {871            'type': "HTTP", 'delay': 10, 'timeout': 10,872            'attemptsBeforeDeactivation': 4, 'path': "/",873            'statusRegex': "^[234][0-9][0-9]$", 'bodyRegex': ".* testing .*",874            'hostHeader': "example.com"}875        self.m.StubOutWithMock(rsrc.clb, 'get')876        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)877        self.m.StubOutWithMock(fake_loadbalancer, 'add_health_monitor')878        fake_loadbalancer.add_health_monitor(879            attemptsBeforeDeactivation=4, bodyRegex='.* testing .*', delay=10,880            hostHeader='example.com', path='/',881            statusRegex='^[234][0-9][0-9]$', timeout=10, type='HTTP')882        self.m.ReplayAll()883        scheduler.TaskRunner(rsrc.update, update_template)()884        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)885        self.m.VerifyAll()886    def test_update_health_monitor_delete(self):887        template = copy.deepcopy(self.lb_template)888        lb_name = list(six.iterkeys(template['Resources']))[0]889        hm = {'type': "HTTP", 'delay': 10, 'timeout': 10,890              'attemptsBeforeDeactivation': 4, 'path': "/",891              'statusRegex': "^[234][0-9][0-9]$", 'bodyRegex': ".* testing .*",892              'hostHeader': "example.com"}893        template['Resources'][lb_name]['Properties']['healthMonitor'] = hm894        expected_body = copy.deepcopy(self.expected_body)895        expected_body['healthMonitor'] = hm896        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,897                                                          self.lb_name,898                                                          expected_body)899        self.m.ReplayAll()900        scheduler.TaskRunner(rsrc.create)()901        update_template = copy.deepcopy(rsrc.t)902        del update_template['Properties']['healthMonitor']903        self.m.StubOutWithMock(rsrc.clb, 'get')904        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)905        self.m.StubOutWithMock(fake_loadbalancer, 'delete_health_monitor')906        fake_loadbalancer.delete_health_monitor()907        self.m.ReplayAll()908        scheduler.TaskRunner(rsrc.update, update_template)()909        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)910        self.m.VerifyAll()911    def test_update_session_persistence_add(self):912        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,913                                                          self.lb_name,914                                                          self.expected_body)915        self.m.ReplayAll()916        scheduler.TaskRunner(rsrc.create)()917        update_template = copy.deepcopy(rsrc.t)918        update_template['Properties']['sessionPersistence'] = 'SOURCE_IP'919        self.m.StubOutWithMock(rsrc.clb, 'get')920        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)921        self.m.ReplayAll()922        scheduler.TaskRunner(rsrc.update, update_template)()923        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)924        self.assertEqual('SOURCE_IP', fake_loadbalancer.session_persistence)925        self.m.VerifyAll()926    def test_update_session_persistence_delete(self):927        template = copy.deepcopy(self.lb_template)928        lb_name = list(six.iterkeys(template['Resources']))[0]929        template['Resources'][lb_name]['Properties'][930            'sessionPersistence'] = "SOURCE_IP"931        expected_body = copy.deepcopy(self.expected_body)932        expected_body['sessionPersistence'] = {'persistenceType': "SOURCE_IP"}933        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,934                                                          self.lb_name,935                                                          expected_body)936        self.m.ReplayAll()937        scheduler.TaskRunner(rsrc.create)()938        update_template = copy.deepcopy(rsrc.t)939        del update_template['Properties']['sessionPersistence']940        self.m.StubOutWithMock(rsrc.clb, 'get')941        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)942        self.m.ReplayAll()943        scheduler.TaskRunner(rsrc.update, update_template)()944        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)945        self.assertEqual('', fake_loadbalancer.session_persistence)946        self.m.VerifyAll()947    def test_update_ssl_termination_add(self):948        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,949                                                          self.lb_name,950                                                          self.expected_body)951        self.m.ReplayAll()952        scheduler.TaskRunner(rsrc.create)()953        update_template = copy.deepcopy(rsrc.t)954        update_template['Properties']['sslTermination'] = {955            'securePort': 443, 'privatekey': private_key, 'certificate': cert,956            'secureTrafficOnly': False}957        self.m.StubOutWithMock(rsrc.clb, 'get')958        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)959        self.m.StubOutWithMock(fake_loadbalancer, 'add_ssl_termination')960        fake_loadbalancer.add_ssl_termination(961            securePort=443, privatekey=private_key, certificate=cert,962            secureTrafficOnly=False, intermediateCertificate=None)963        self.m.ReplayAll()964        scheduler.TaskRunner(rsrc.update, update_template)()965        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)966        self.m.VerifyAll()967    def test_update_ssl_termination_delete(self):968        template = copy.deepcopy(self.lb_template)969        lb_name = list(six.iterkeys(template['Resources']))[0]970        template['Resources'][lb_name]['Properties']['sslTermination'] = {971            'securePort': 443, 'privatekey': private_key, 'certificate': cert,972            'secureTrafficOnly': False}973        # The SSL termination config is done post-creation, so no need974        # to modify self.expected_body975        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,976                                                          self.lb_name,977                                                          self.expected_body)978        self.m.ReplayAll()979        scheduler.TaskRunner(rsrc.create)()980        update_template = copy.deepcopy(rsrc.t)981        del update_template['Properties']['sslTermination']982        self.m.StubOutWithMock(rsrc.clb, 'get')983        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)984        self.m.StubOutWithMock(fake_loadbalancer, 'delete_ssl_termination')985        fake_loadbalancer.delete_ssl_termination()986        self.m.ReplayAll()987        scheduler.TaskRunner(rsrc.update, update_template)()988        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)989        self.m.VerifyAll()990    def test_update_metadata_add(self):991        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,992                                                          self.lb_name,993                                                          self.expected_body)994        self.m.ReplayAll()995        scheduler.TaskRunner(rsrc.create)()996        update_template = copy.deepcopy(rsrc.t)997        update_template['Properties']['metadata'] = {'a': 1, 'b': 2}998        self.m.StubOutWithMock(rsrc.clb, 'get')999        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1000        self.m.StubOutWithMock(fake_loadbalancer, 'set_metadata')1001        fake_loadbalancer.set_metadata({'a': 1, 'b': 2})1002        self.m.ReplayAll()1003        scheduler.TaskRunner(rsrc.update, update_template)()1004        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1005        self.m.VerifyAll()1006    def test_update_metadata_delete(self):1007        template = copy.deepcopy(self.lb_template)1008        lb_name = list(six.iterkeys(template['Resources']))[0]1009        template['Resources'][lb_name]['Properties']['metadata'] = {1010            'a': 1, 'b': 2}1011        expected_body = copy.deepcopy(self.expected_body)1012        expected_body['metadata'] = mox.SameElementsAs(1013            [{'key': 'a', 'value': 1},1014             {'key': 'b', 'value': 2}])1015        rsrc, fake_loadbalancer = self._mock_loadbalancer(1016            template, self.lb_name, expected_body)1017        self.m.ReplayAll()1018        scheduler.TaskRunner(rsrc.create)()1019        update_template = copy.deepcopy(rsrc.t)1020        del update_template['Properties']['metadata']1021        self.m.StubOutWithMock(rsrc.clb, 'get')1022        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1023        self.m.StubOutWithMock(fake_loadbalancer, 'delete_metadata')1024        fake_loadbalancer.delete_metadata()1025        self.m.ReplayAll()1026        scheduler.TaskRunner(rsrc.update, update_template)()1027        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1028        self.m.VerifyAll()1029    def test_update_errorpage_add(self):1030        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,1031                                                          self.lb_name,1032                                                          self.expected_body)1033        self.m.ReplayAll()1034        scheduler.TaskRunner(rsrc.create)()1035        error_page = (1036            '<html><head><title>Service Unavailable</title></head><body><h2>'1037            'Service Unavailable</h2>The service is unavailable</body></html>')1038        update_template = copy.deepcopy(rsrc.t)1039        update_template['Properties']['errorPage'] = error_page1040        self.m.StubOutWithMock(rsrc.clb, 'get')1041        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1042        self.m.StubOutWithMock(fake_loadbalancer, 'set_error_page')1043        fake_loadbalancer.set_error_page(error_page)1044        self.m.ReplayAll()1045        scheduler.TaskRunner(rsrc.update, update_template)()1046        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1047        self.m.VerifyAll()1048    def test_update_errorpage_delete(self):1049        template = copy.deepcopy(self.lb_template)1050        lb_name = list(six.iterkeys(template['Resources']))[0]1051        error_page = (1052            '<html><head><title>Service Unavailable</title></head><body><h2>'1053            'Service Unavailable</h2>The service is unavailable</body></html>')1054        template['Resources'][lb_name]['Properties']['errorPage'] = error_page1055        # The error page config is done post-creation, so no need to1056        # modify self.expected_body1057        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,1058                                                          self.lb_name,1059                                                          self.expected_body)1060        self.m.ReplayAll()1061        scheduler.TaskRunner(rsrc.create)()1062        update_template = copy.deepcopy(rsrc.t)1063        del update_template['Properties']['errorPage']1064        self.m.StubOutWithMock(rsrc.clb, 'get')1065        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1066        self.m.StubOutWithMock(fake_loadbalancer, 'clear_error_page')1067        fake_loadbalancer.clear_error_page()1068        self.m.ReplayAll()1069        scheduler.TaskRunner(rsrc.update, update_template)()1070        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1071        self.m.VerifyAll()1072    def test_update_connection_logging_enable(self):1073        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,1074                                                          self.lb_name,1075                                                          self.expected_body)1076        self.m.ReplayAll()1077        scheduler.TaskRunner(rsrc.create)()1078        update_template = copy.deepcopy(rsrc.t)1079        update_template['Properties']['connectionLogging'] = True1080        self.m.StubOutWithMock(rsrc.clb, 'get')1081        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1082        self.m.ReplayAll()1083        scheduler.TaskRunner(rsrc.update, update_template)()1084        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1085        self.assertEqual(True, fake_loadbalancer.connection_logging)1086        self.m.VerifyAll()1087    def test_update_connection_logging_delete(self):1088        template = copy.deepcopy(self.lb_template)1089        lb_name = list(six.iterkeys(template['Resources']))[0]1090        template['Resources'][lb_name]['Properties'][1091            'connectionLogging'] = True1092        expected_body = copy.deepcopy(self.expected_body)1093        expected_body['connectionLogging'] = {'enabled': True}1094        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,1095                                                          self.lb_name,1096                                                          expected_body)1097        self.m.ReplayAll()1098        scheduler.TaskRunner(rsrc.create)()1099        update_template = copy.deepcopy(rsrc.t)1100        del update_template['Properties']['connectionLogging']1101        self.m.StubOutWithMock(rsrc.clb, 'get')1102        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1103        self.m.ReplayAll()1104        scheduler.TaskRunner(rsrc.update, update_template)()1105        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1106        self.assertEqual(False, fake_loadbalancer.connection_logging)1107        self.m.VerifyAll()1108    def test_update_connection_logging_disable(self):1109        template = copy.deepcopy(self.lb_template)1110        lb_name = list(six.iterkeys(template['Resources']))[0]1111        template['Resources'][lb_name]['Properties'][1112            'connectionLogging'] = True1113        expected_body = copy.deepcopy(self.expected_body)1114        expected_body['connectionLogging'] = {'enabled': True}1115        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,1116                                                          self.lb_name,1117                                                          expected_body)1118        self.m.ReplayAll()1119        scheduler.TaskRunner(rsrc.create)()1120        update_template = copy.deepcopy(rsrc.t)1121        update_template['Properties']['connectionLogging'] = False1122        self.m.StubOutWithMock(rsrc.clb, 'get')1123        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1124        self.m.ReplayAll()1125        scheduler.TaskRunner(rsrc.update, update_template)()1126        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1127        self.assertEqual(False, fake_loadbalancer.connection_logging)1128        self.m.VerifyAll()1129    def test_update_connection_throttle_add(self):1130        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,1131                                                          self.lb_name,1132                                                          self.expected_body)1133        self.m.ReplayAll()1134        scheduler.TaskRunner(rsrc.create)()1135        update_template = copy.deepcopy(rsrc.t)1136        update_template['Properties']['connectionThrottle'] = {1137            'maxConnections': 1000}1138        self.m.StubOutWithMock(rsrc.clb, 'get')1139        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1140        self.m.StubOutWithMock(fake_loadbalancer, 'add_connection_throttle')1141        fake_loadbalancer.add_connection_throttle(1142            maxConnections=1000, maxConnectionRate=None, minConnections=None,1143            rateInterval=None)1144        self.m.ReplayAll()1145        scheduler.TaskRunner(rsrc.update, update_template)()1146        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1147        self.m.VerifyAll()1148    def test_update_connection_throttle_delete(self):1149        template = copy.deepcopy(self.lb_template)1150        lb_name = list(six.iterkeys(template['Resources']))[0]1151        template['Resources'][lb_name]['Properties'][1152            'connectionThrottle'] = {'maxConnections': 1000}1153        expected_body = copy.deepcopy(self.expected_body)1154        expected_body['connectionThrottle'] = {1155            'maxConnections': 1000, 'maxConnectionRate': None,1156            'rateInterval': None, 'minConnections': None}1157        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,1158                                                          self.lb_name,1159                                                          expected_body)1160        self.m.ReplayAll()1161        scheduler.TaskRunner(rsrc.create)()1162        update_template = copy.deepcopy(rsrc.t)1163        del update_template['Properties']['connectionThrottle']1164        self.m.StubOutWithMock(rsrc.clb, 'get')1165        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1166        self.m.StubOutWithMock(fake_loadbalancer, 'delete_connection_throttle')1167        fake_loadbalancer.delete_connection_throttle()1168        self.m.ReplayAll()1169        scheduler.TaskRunner(rsrc.update, update_template)()1170        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1171        self.m.VerifyAll()1172    def test_update_content_caching_enable(self):1173        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,1174                                                          self.lb_name,1175                                                          self.expected_body)1176        self.m.ReplayAll()1177        scheduler.TaskRunner(rsrc.create)()1178        update_template = copy.deepcopy(rsrc.t)1179        update_template['Properties']['contentCaching'] = 'ENABLED'1180        self.m.StubOutWithMock(rsrc.clb, 'get')1181        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1182        self.m.ReplayAll()1183        scheduler.TaskRunner(rsrc.update, update_template)()1184        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1185        self.assertEqual(True, fake_loadbalancer.content_caching)1186        self.m.VerifyAll()1187    def test_update_content_caching_deleted(self):1188        template = copy.deepcopy(self.lb_template)1189        lb_name = list(six.iterkeys(template['Resources']))[0]1190        template['Resources'][lb_name]['Properties'][1191            'contentCaching'] = 'ENABLED'1192        # Enabling the content cache is done post-creation, so no need1193        # to modify self.expected_body1194        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,1195                                                          self.lb_name,1196                                                          self.expected_body)1197        self.m.ReplayAll()1198        scheduler.TaskRunner(rsrc.create)()1199        update_template = copy.deepcopy(rsrc.t)1200        del update_template['Properties']['contentCaching']1201        self.m.StubOutWithMock(rsrc.clb, 'get')1202        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1203        self.m.ReplayAll()1204        self.assertEqual(True, fake_loadbalancer.content_caching)1205        scheduler.TaskRunner(rsrc.update, update_template)()1206        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1207        self.assertEqual(False, fake_loadbalancer.content_caching)1208        self.m.VerifyAll()1209    def test_update_content_caching_disable(self):1210        template = copy.deepcopy(self.lb_template)1211        lb_name = list(six.iterkeys(template['Resources']))[0]1212        template['Resources'][lb_name]['Properties'][1213            'contentCaching'] = 'ENABLED'1214        # Enabling the content cache is done post-creation, so no need1215        # to modify self.expected_body1216        rsrc, fake_loadbalancer = self._mock_loadbalancer(template,1217                                                          self.lb_name,1218                                                          self.expected_body)1219        self.m.ReplayAll()1220        scheduler.TaskRunner(rsrc.create)()1221        update_template = copy.deepcopy(rsrc.t)1222        update_template['Properties']['contentCaching'] = 'DISABLED'1223        self.m.StubOutWithMock(rsrc.clb, 'get')1224        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1225        self.m.ReplayAll()1226        self.assertEqual(True, fake_loadbalancer.content_caching)1227        scheduler.TaskRunner(rsrc.update, update_template)()1228        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)1229        self.assertEqual(False, fake_loadbalancer.content_caching)1230        self.m.VerifyAll()1231    def test_delete(self):1232        template = self._set_template(self.lb_template,1233                                      contentCaching='ENABLED')1234        rsrc, fake_lb = self._mock_loadbalancer(template, self.lb_name,1235                                                self.expected_body)1236        self.m.ReplayAll()1237        scheduler.TaskRunner(rsrc.create)()1238        self.m.VerifyAll()1239        self.m.StubOutWithMock(rsrc.clb, 'get')1240        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_lb)1241        self.m.ReplayAll()1242        scheduler.TaskRunner(rsrc.delete)()1243        self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)1244    def test_delete_immutable(self):1245        template = self._set_template(self.lb_template,1246                                      contentCaching='ENABLED')1247        rsrc, fake_lb = self._mock_loadbalancer(template, self.lb_name,1248                                                self.expected_body)1249        self.m.ReplayAll()1250        scheduler.TaskRunner(rsrc.create)()1251        self.m.VerifyAll()1252        self.m.StubOutWithMock(rsrc.clb, 'get')1253        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_lb)1254        self.m.StubOutWithMock(fake_lb, 'delete')1255        fake_lb.delete().AndRaise(Exception('immutable'))1256        fake_lb.delete().AndReturn(None)1257        self.m.ReplayAll()1258        scheduler.TaskRunner(rsrc.delete)()1259        self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)1260    def test_check_delete_complete(self):1261        mock_stack = mock.Mock()1262        mock_stack.db_resource_get.return_value = None1263        mock_resdef = mock.Mock(spec=rsrc_defn.ResourceDefinition)1264        mock_loadbalancer = lb.CloudLoadBalancer("test", mock_resdef,1265                                                 mock_stack)1266        mock_task = mock.Mock()1267        mock_task.step.return_value = False1268        res = mock_loadbalancer.check_delete_complete(mock_task)1269        self.assertFalse(res)1270        mock_task.step.return_value = True1271        res = mock_loadbalancer.check_delete_complete(mock_task)1272        self.assertTrue(res)1273    def test_redir(self):1274        mock_stack = mock.Mock()1275        mock_stack.db_resource_get.return_value = None1276        props = {'httpsRedirect': True,1277                 'protocol': 'HTTPS',1278                 'port': 443,1279                 'nodes': [],1280                 'virtualIps': [{'id': '1234'}]}1281        mock_resdef = rsrc_defn.ResourceDefinition("test_lb",1282                                                   LoadBalancerWithFakeClient,1283                                                   properties=props)1284        mock_lb = lb.CloudLoadBalancer("test", mock_resdef, mock_stack)1285        self.assertIsNone(mock_lb.validate())1286        props['protocol'] = 'HTTP'1287        props['sslTermination'] = {1288            'secureTrafficOnly': True,1289            'securePort': 443,1290            'privatekey': "bobloblaw",1291            'certificate': 'mycert'1292        }1293        mock_resdef = rsrc_defn.ResourceDefinition("test_lb_2",1294                                                   LoadBalancerWithFakeClient,1295                                                   properties=props)1296        mock_lb = lb.CloudLoadBalancer("test_2", mock_resdef, mock_stack)1297        self.assertIsNone(mock_lb.validate())1298    def test_invalid_redir_proto(self):1299        mock_stack = mock.Mock()1300        mock_stack.db_resource_get.return_value = None1301        props = {'httpsRedirect': True,1302                 'protocol': 'TCP',1303                 'port': 1234,1304                 'nodes': [],1305                 'virtualIps': [{'id': '1234'}]}1306        mock_resdef = rsrc_defn.ResourceDefinition("test_lb",1307                                                   LoadBalancerWithFakeClient,1308                                                   properties=props)1309        mock_lb = lb.CloudLoadBalancer("test", mock_resdef, mock_stack)1310        ex = self.assertRaises(exception.StackValidationFailed,1311                               mock_lb.validate)1312        self.assertIn("HTTPS redirect is only available", six.text_type(ex))1313    def test_invalid_redir_ssl(self):1314        mock_stack = mock.Mock()1315        mock_stack.db_resource_get.return_value = None1316        props = {'httpsRedirect': True,1317                 'protocol': 'HTTP',1318                 'port': 1234,1319                 'nodes': [],1320                 'virtualIps': [{'id': '1234'}]}1321        mock_resdef = rsrc_defn.ResourceDefinition("test_lb",1322                                                   LoadBalancerWithFakeClient,1323                                                   properties=props)1324        mock_lb = lb.CloudLoadBalancer("test", mock_resdef, mock_stack)1325        ex = self.assertRaises(exception.StackValidationFailed,1326                               mock_lb.validate)1327        self.assertIn("HTTPS redirect is only available", six.text_type(ex))1328        props['sslTermination'] = {1329            'secureTrafficOnly': False,1330            'securePort': 443,1331            'privatekey': "bobloblaw",1332            'certificate': 'mycert'1333        }1334        mock_lb = lb.CloudLoadBalancer("test", mock_resdef, mock_stack)1335        ex = self.assertRaises(exception.StackValidationFailed,1336                               mock_lb.validate)1337        self.assertIn("HTTPS redirect is only available", six.text_type(ex))1338        props['sslTermination'] = {1339            'secureTrafficOnly': True,1340            'securePort': 1234,1341            'privatekey': "bobloblaw",1342            'certificate': 'mycert'1343        }1344        mock_lb = lb.CloudLoadBalancer("test", mock_resdef, mock_stack)1345        ex = self.assertRaises(exception.StackValidationFailed,1346                               mock_lb.validate)1347        self.assertIn("HTTPS redirect is only available", six.text_type(ex))1348    def test_update_nodes_condition_draining(self):1349        rsrc, fake_loadbalancer = self._mock_loadbalancer(self.lb_template,1350                                                          self.lb_name,1351                                                          self.expected_body)1352        fake_loadbalancer.nodes = self.expected_body['nodes']1353        self.m.ReplayAll()1354        scheduler.TaskRunner(rsrc.create)()1355        update_template = copy.deepcopy(rsrc.t)1356        expected_ip = '172.168.1.4'1357        update_template['Properties']['nodes'] = [1358            {"addresses": ["166.78.103.141"],1359             "port": 80,1360             "condition": "DRAINING"},1361            {"addresses": [expected_ip],1362             "port": 80,1363             "condition": "DRAINING"}]1364        self.m.StubOutWithMock(rsrc.clb, 'get')1365        rsrc.clb.get(rsrc.resource_id).AndReturn(fake_loadbalancer)1366        self.m.StubOutWithMock(fake_loadbalancer, 'add_nodes')1367        fake_loadbalancer.add_nodes([1368            fake_loadbalancer.Node(address=expected_ip,1369                                   port=80,1370                                   condition='DRAINING')])1371        self.m.ReplayAll()1372        scheduler.TaskRunner(rsrc.update, update_template)()1373        self.assertEqual((rsrc.UPDATE, rsrc.COMPLETE), rsrc.state)...

test_update_restricted.py

Source:test_update_restricted.py

1#    Licensed under the Apache License, Version 2.0 (the "License"); you may2#    not use this file except in compliance with the License. You may obtain3#    a copy of the License at4#5#         http://www.apache.org/licenses/LICENSE-2.06#7#    Unless required by applicable law or agreed to in writing, software8#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT9#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the10#    License for the specific language governing permissions and limitations11#    under the License.12import time13from heat_integrationtests.functional import functional_base14test_template = {15    'heat_template_version': '2013-05-23',16    'description': 'Test template to create one instance.',17    'resources': {18        'bar': {19            'type': 'OS::Heat::TestResource',20            'properties': {21                'value': '1234',22                'update_replace': False,23            }24        }25    }26}27env_both_restrict = {u'resource_registry': {28    u'resources': {29        'bar': {'restricted_actions': ['update', 'replace']}30    }31}32}33env_replace_restrict = {u'resource_registry': {34    u'resources': {35        '*ar': {'restricted_actions': 'replace'}36    }37}38}39reason_update_restrict = 'update is restricted for resource.'40reason_replace_restrict = 'replace is restricted for resource.'41class UpdateRestrictedStackTest(functional_base.FunctionalTestsBase):42    def _check_for_restriction_reason(self, events,43                                      reason, num_expected=1):44        matched = [e for e in events45                   if e.resource_status_reason == reason]46        return len(matched) == num_expected47    def test_update(self):48        stack_identifier = self.stack_create(template=test_template)49        update_template = test_template.copy()50        props = update_template['resources']['bar']['properties']51        props['value'] = '4567'52        # check update fails - with 'both' restricted53        self.update_stack(stack_identifier, update_template,54                          env_both_restrict,55                          expected_status='UPDATE_FAILED')56        self.assertTrue(self.verify_resource_status(stack_identifier, 'bar',57                                                    'CREATE_COMPLETE'))58        resource_events = self.client.events.list(stack_identifier, 'bar')59        self.assertTrue(60            self._check_for_restriction_reason(resource_events,61                                               reason_update_restrict))62        # Ensure the timestamp changes, since this will be very quick63        time.sleep(1)64        # check update succeeds - with only 'replace' restricted65        self.update_stack(stack_identifier, update_template,66                          env_replace_restrict,67                          expected_status='UPDATE_COMPLETE')68        self.assertTrue(self.verify_resource_status(stack_identifier, 'bar',69                                                    'UPDATE_COMPLETE'))70        resource_events = self.client.events.list(stack_identifier, 'bar')71        self.assertFalse(72            self._check_for_restriction_reason(resource_events,73                                               reason_update_restrict, 2))74        self.assertTrue(75            self._check_for_restriction_reason(resource_events,76                                               reason_replace_restrict, 0))77    def test_replace(self):78        stack_identifier = self.stack_create(template=test_template)79        update_template = test_template.copy()80        props = update_template['resources']['bar']['properties']81        props['update_replace'] = True82        # check replace fails - with 'both' restricted83        self.update_stack(stack_identifier, update_template,84                          env_both_restrict,85                          expected_status='UPDATE_FAILED')86        self.assertTrue(self.verify_resource_status(stack_identifier, 'bar',87                                                    'CREATE_COMPLETE'))88        resource_events = self.client.events.list(stack_identifier, 'bar')89        self.assertTrue(90            self._check_for_restriction_reason(resource_events,91                                               reason_replace_restrict))92        # Ensure the timestamp changes, since this will be very quick93        time.sleep(1)94        # check replace fails - with only 'replace' restricted95        self.update_stack(stack_identifier, update_template,96                          env_replace_restrict,97                          expected_status='UPDATE_FAILED')98        self.assertTrue(self.verify_resource_status(stack_identifier, 'bar',99                                                    'CREATE_COMPLETE'))100        resource_events = self.client.events.list(stack_identifier, 'bar')101        self.assertTrue(102            self._check_for_restriction_reason(resource_events,103                                               reason_replace_restrict, 2))104        self.assertTrue(105            self._check_for_restriction_reason(resource_events,106                                               reason_update_restrict, 0))107    def test_update_type_changed(self):108        stack_identifier = self.stack_create(template=test_template)109        update_template = test_template.copy()110        rsrc = update_template['resources']['bar']111        rsrc['type'] = 'OS::Heat::None'112        # check replace fails - with 'both' restricted113        self.update_stack(stack_identifier, update_template,114                          env_both_restrict,115                          expected_status='UPDATE_FAILED')116        self.assertTrue(self.verify_resource_status(stack_identifier, 'bar',117                                                    'CREATE_COMPLETE'))118        resource_events = self.client.events.list(stack_identifier, 'bar')119        self.assertTrue(120            self._check_for_restriction_reason(resource_events,121                                               reason_replace_restrict))122        # Ensure the timestamp changes, since this will be very quick123        time.sleep(1)124        # check replace fails - with only 'replace' restricted125        self.update_stack(stack_identifier, update_template,126                          env_replace_restrict,127                          expected_status='UPDATE_FAILED')128        self.assertTrue(self.verify_resource_status(stack_identifier, 'bar',129                                                    'CREATE_COMPLETE'))130        resource_events = self.client.events.list(stack_identifier, 'bar')131        self.assertTrue(132            self._check_for_restriction_reason(resource_events,133                                               reason_replace_restrict, 2))134        self.assertTrue(135            self._check_for_restriction_reason(resource_events,...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.