How to use _create_container method in localstack

Best Python code snippet using localstack_python
test_container.py
Source:test_container.py
...62class SwiftTest(common.HeatTestCase):63    def setUp(self):64        super(SwiftTest, self).setUp()65        self.t = template_format.parse(SWIFT_TEMPLATE)66    def _create_container(self, stack, definition_name='SwiftContainer'):67        resource_defns = stack.t.resource_definitions(stack)68        container = swift_c.SwiftContainer('test_resource',69                                           resource_defns[definition_name],70                                           stack)71        runner = scheduler.TaskRunner(container.create)72        runner()73        self.assertEqual((container.CREATE, container.COMPLETE),74                         container.state)75        return container76    @mock.patch('swiftclient.client.Connection.put_container')77    def test_create_container_name(self, mock_put):78        # Setup79        res_prop = self.t['Resources']['SwiftContainer']['Properties']80        res_prop['name'] = 'the_name'81        stack = utils.parse_stack(self.t)82        # Test83        container = self._create_container(stack)84        container_name = container.physical_resource_name()85        # Verify86        self.assertEqual('the_name', container_name)87        mock_put.assert_called_once_with('the_name', {})88    def test_build_meta_headers(self):89        # Setup90        headers = {'Web-Index': 'index.html', 'Web-Error': 'error.html'}91        # Test92        self.assertEqual({}, swift_c.SwiftContainer._build_meta_headers(93            'container', {}))94        self.assertEqual({}, swift_c.SwiftContainer._build_meta_headers(95            'container', None))96        built = swift_c.SwiftContainer._build_meta_headers(97            'container', headers)98        # Verify99        expected = {100            'X-Container-Meta-Web-Index': 'index.html',101            'X-Container-Meta-Web-Error': 'error.html'102        }103        self.assertEqual(expected, built)104    @mock.patch('swiftclient.client.Connection.head_container')105    @mock.patch('swiftclient.client.Connection.put_container')106    def test_attributes(self, mock_put, mock_head):107        # Setup108        headers = {'content-length': '0',109                   'x-container-object-count': '82',110                   'accept-ranges': 'bytes',111                   'x-trans-id': 'tx08ea48ef2fa24e6da3d2f5c188fd938b',112                   'date': 'Wed, 23 Jan 2013 22:48:05 GMT',113                   'x-timestamp': '1358980499.84298',114                   'x-container-read': '.r:*',115                   'x-container-bytes-used': '17680980',116                   'content-type': 'text/plain; charset=utf-8'}117        mock_head.return_value = headers118        stack = utils.parse_stack(self.t)119        container_name = utils.PhysName(stack.name, 'test_resource')120        # Test121        container = self._create_container(stack)122        # call this to populate the url of swiftclient. This is actually123        # set in head_container/put_container, but we're patching them in124        # this test.125        container.client().get_auth()126        # Verify Attributes127        self.assertEqual(container_name, container.FnGetRefId())128        self.assertEqual('82', container.FnGetAtt('ObjectCount'))129        self.assertEqual('17680980', container.FnGetAtt('BytesUsed'))130        self.assertEqual('server.test', container.FnGetAtt('DomainName'))131        self.assertEqual(headers, container.FnGetAtt('HeadContainer'))132        self.assertEqual(headers, container.FnGetAtt('show'))133        expected_url = 'http://server.test:5000/v3/%s' % container.FnGetRefId()134        self.assertEqual(expected_url, container.FnGetAtt('WebsiteURL'))135        self.assertRaises(exception.InvalidTemplateAttribute,136                          container.FnGetAtt, 'Foo')137        # Verify Expected Calls138        mock_put.assert_called_once_with(container_name, {})139        self.assertGreater(mock_head.call_count, 0)140    @mock.patch('swiftclient.client.Connection.put_container')141    def test_public_read(self, mock_put):142        # Setup143        properties = self.t['Resources']['SwiftContainer']['Properties']144        properties['X-Container-Read'] = '.r:*'145        stack = utils.parse_stack(self.t)146        container_name = utils.PhysName(stack.name, 'test_resource')147        # Test148        self._create_container(stack)149        # Verify150        expected = {'X-Container-Read': '.r:*'}151        mock_put.assert_called_once_with(container_name, expected)152    @mock.patch('swiftclient.client.Connection.put_container')153    def test_public_read_write(self, mock_put):154        # Setup155        properties = self.t['Resources']['SwiftContainer']['Properties']156        properties['X-Container-Read'] = '.r:*'157        properties['X-Container-Write'] = '.r:*'158        stack = utils.parse_stack(self.t)159        container_name = utils.PhysName(stack.name, 'test_resource')160        # Test161        self._create_container(stack)162        # Verify163        expected = {'X-Container-Write': '.r:*', 'X-Container-Read': '.r:*'}164        mock_put.assert_called_once_with(container_name, expected)165    @mock.patch('swiftclient.client.Connection.put_container')166    def test_container_headers(self, mock_put):167        # Setup168        stack = utils.parse_stack(self.t)169        container_name = utils.PhysName(stack.name, 'test_resource')170        # Test171        self._create_container(stack,172                               definition_name='SwiftContainerWebsite')173        # Verify174        expected = {'X-Container-Meta-Web-Error': 'error.html',175                    'X-Container-Meta-Web-Index': 'index.html',176                    'X-Container-Read': '.r:*'}177        mock_put.assert_called_once_with(container_name, expected)178    @mock.patch('swiftclient.client.Connection.post_account')179    @mock.patch('swiftclient.client.Connection.put_container')180    def test_account_headers(self, mock_put, mock_post):181        # Setup182        stack = utils.parse_stack(self.t)183        container_name = utils.PhysName(stack.name, 'test_resource')184        # Test185        self._create_container(stack,186                               definition_name='SwiftAccountMetadata')187        # Verify188        mock_put.assert_called_once_with(container_name, {})189        expected = {'X-Account-Meta-Temp-Url-Key': 'secret'}190        mock_post.assert_called_once_with(expected)191    @mock.patch('swiftclient.client.Connection.put_container')192    def test_default_headers_not_none_empty_string(self, mock_put):193        # Setup194        stack = utils.parse_stack(self.t)195        container_name = utils.PhysName(stack.name, 'test_resource')196        # Test197        container = self._create_container(stack)198        # Verify199        mock_put.assert_called_once_with(container_name, {})200        self.assertEqual({}, container.metadata_get())201    @mock.patch('swiftclient.client.Connection.delete_container')202    @mock.patch('swiftclient.client.Connection.get_container')203    @mock.patch('swiftclient.client.Connection.put_container')204    def test_delete_exception(self, mock_put, mock_get, mock_delete):205        # Setup206        stack = utils.parse_stack(self.t)207        container_name = utils.PhysName(stack.name, 'test_resource')208        mock_delete.side_effect = sc.ClientException('test-delete-failure')209        mock_get.return_value = ({'name': container_name}, [])210        # Test211        container = self._create_container(stack)212        runner = scheduler.TaskRunner(container.delete)213        self.assertRaises(exception.ResourceFailure, runner)214        # Verify215        self.assertEqual((container.DELETE, container.FAILED),216                         container.state)217        mock_put.assert_called_once_with(container_name, {})218        mock_get.assert_called_once_with(container_name)219        mock_delete.assert_called_once_with(container_name)220    @mock.patch('swiftclient.client.Connection.delete_container')221    @mock.patch('swiftclient.client.Connection.get_container')222    @mock.patch('swiftclient.client.Connection.put_container')223    def test_delete_not_found(self, mock_put, mock_get, mock_delete):224        # Setup225        stack = utils.parse_stack(self.t)226        container_name = utils.PhysName(stack.name, 'test_resource')227        mock_delete.side_effect = sc.ClientException('missing',228                                                     http_status=404)229        mock_get.return_value = ({'name': container_name}, [])230        # Test231        container = self._create_container(stack)232        runner = scheduler.TaskRunner(container.delete)233        runner()234        # Verify235        self.assertEqual((container.DELETE, container.COMPLETE),236                         container.state)237        mock_put.assert_called_once_with(container_name, {})238        mock_get.assert_called_once_with(container_name)239        mock_delete.assert_called_once_with(container_name)240    @mock.patch('swiftclient.client.Connection.get_container')241    @mock.patch('swiftclient.client.Connection.put_container')242    def test_delete_non_empty_not_allowed(self, mock_put, mock_get):243        # Setup244        stack = utils.parse_stack(self.t)245        container_name = utils.PhysName(stack.name, 'test_resource')246        mock_get.return_value = ({'name': container_name},247                                 [{'name': 'test_object'}])248        # Test249        container = self._create_container(stack)250        runner = scheduler.TaskRunner(container.delete)251        ex = self.assertRaises(exception.ResourceFailure, runner)252        # Verify253        self.assertEqual((container.DELETE, container.FAILED),254                         container.state)255        self.assertIn('ResourceActionNotSupported: resources.test_resource: '256                      'Deleting non-empty container',257                      six.text_type(ex))258        mock_put.assert_called_once_with(container_name, {})259        mock_get.assert_called_once_with(container_name)260    @mock.patch('swiftclient.client.Connection.delete_container')261    @mock.patch('swiftclient.client.Connection.delete_object')262    @mock.patch('swiftclient.client.Connection.get_container')263    @mock.patch('swiftclient.client.Connection.put_container')264    def test_delete_non_empty_allowed(self, mock_put, mock_get,265                                      mock_delete_object,266                                      mock_delete_container):267        # Setup268        res_prop = self.t['Resources']['SwiftContainer']['Properties']269        res_prop['PurgeOnDelete'] = True270        stack = utils.parse_stack(self.t)271        container_name = utils.PhysName(stack.name, 'test_resource')272        get_return_values = [273            ({'name': container_name},274             [{'name': 'test_object1'},275              {'name': 'test_object2'}]),276            ({'name': container_name}, [{'name': 'test_object1'}]),277        ]278        mock_get.side_effect = get_return_values279        # Test280        container = self._create_container(stack)281        runner = scheduler.TaskRunner(container.delete)282        runner()283        # Verify284        self.assertEqual((container.DELETE, container.COMPLETE),285                         container.state)286        mock_put.assert_called_once_with(container_name, {})287        mock_delete_container.assert_called_once_with(container_name)288        self.assertEqual(2, mock_get.call_count)289        self.assertEqual(2, mock_delete_object.call_count)290    @mock.patch('swiftclient.client.Connection.delete_container')291    @mock.patch('swiftclient.client.Connection.delete_object')292    @mock.patch('swiftclient.client.Connection.get_container')293    @mock.patch('swiftclient.client.Connection.put_container')294    def test_delete_non_empty_allowed_not_found(self, mock_put, mock_get,295                                                mock_delete_object,296                                                mock_delete_container):297        # Setup298        res_prop = self.t['Resources']['SwiftContainer']['Properties']299        res_prop['PurgeOnDelete'] = True300        stack = utils.parse_stack(self.t)301        container_name = utils.PhysName(stack.name, 'test_resource')302        mock_get.return_value = ({'name': container_name},303                                 [{'name': 'test_object'}])304        mock_delete_object.side_effect = sc.ClientException('object-is-gone',305                                                            http_status=404)306        mock_delete_container.side_effect = sc.ClientException(307            'container-is-gone', http_status=404)308        # Test309        container = self._create_container(stack)310        runner = scheduler.TaskRunner(container.delete)311        runner()312        # Verify313        self.assertEqual((container.DELETE, container.COMPLETE),314                         container.state)315        mock_put.assert_called_once_with(container_name, {})316        mock_get.assert_called_once_with(container_name)317        mock_delete_object.assert_called_once_with(container_name,318                                                   'test_object')319        mock_delete_container.assert_called_once_with(container_name)320    @mock.patch('swiftclient.client.Connection.delete_object')321    @mock.patch('swiftclient.client.Connection.get_container')322    @mock.patch('swiftclient.client.Connection.put_container')323    def test_delete_non_empty_fails_delete_object(self, mock_put, mock_get,324                                                  mock_delete_object):325        # Setup326        res_prop = self.t['Resources']['SwiftContainer']['Properties']327        res_prop['PurgeOnDelete'] = True328        stack = utils.parse_stack(self.t)329        container_name = utils.PhysName(stack.name, 'test_resource')330        mock_get.return_value = ({'name': container_name},331                                 [{'name': 'test_object'}])332        mock_delete_object.side_effect = (333            sc.ClientException('object-delete-failure'))334        # Test335        container = self._create_container(stack)336        runner = scheduler.TaskRunner(container.delete)337        self.assertRaises(exception.ResourceFailure, runner)338        # Verify339        self.assertEqual((container.DELETE, container.FAILED),340                         container.state)341        mock_put.assert_called_once_with(container_name, {})342        mock_get.assert_called_once_with(container_name)343        mock_delete_object.assert_called_once_with(container_name,344                                                   'test_object')345    @mock.patch('swiftclient.client.Connection.put_container')346    def test_delete_retain(self, mock_put):347        # Setup348        self.t['Resources']['SwiftContainer']['DeletionPolicy'] = 'Retain'349        stack = utils.parse_stack(self.t)350        container_name = utils.PhysName(stack.name, 'test_resource')351        # Test352        container = self._create_container(stack)353        runner = scheduler.TaskRunner(container.delete)354        runner()355        # Verify356        self.assertEqual((container.DELETE, container.COMPLETE),357                         container.state)358        mock_put.assert_called_once_with(container_name, {})359    @mock.patch('swiftclient.client.Connection.get_container')360    @mock.patch('swiftclient.client.Connection.put_container')361    def test_check(self, mock_put, mock_get):362        # Setup363        res_prop = self.t['Resources']['SwiftContainer']['Properties']364        res_prop['PurgeOnDelete'] = True365        stack = utils.parse_stack(self.t)366        # Test367        container = self._create_container(stack)368        runner = scheduler.TaskRunner(container.check)369        runner()370        self.assertEqual((container.CHECK, container.COMPLETE),371                         container.state)372    @mock.patch('swiftclient.client.Connection.get_container')373    @mock.patch('swiftclient.client.Connection.put_container')374    def test_check_fail(self, mock_put, mock_get):375        # Setup376        res_prop = self.t['Resources']['SwiftContainer']['Properties']377        res_prop['PurgeOnDelete'] = True378        stack = utils.parse_stack(self.t)379        mock_get.side_effect = Exception('boom')380        # Test381        container = self._create_container(stack)382        runner = scheduler.TaskRunner(container.check)383        ex = self.assertRaises(exception.ResourceFailure, runner)384        # Verify385        self.assertIn('boom', six.text_type(ex))386        self.assertEqual((container.CHECK, container.FAILED),387                         container.state)388    def test_refid(self):389        stack = utils.parse_stack(self.t)390        rsrc = stack['SwiftContainer']391        rsrc.resource_id = 'xyz'392        self.assertEqual('xyz', rsrc.FnGetRefId())393    def test_refid_convergence_cache_data(self):394        cache_data = {'SwiftContainer': node_data.NodeData.from_dict({395            'uuid': mock.ANY,396            'id': mock.ANY,397            'action': 'CREATE',398            'status': 'COMPLETE',399            'reference_id': 'xyz_convg'400        })}401        stack = utils.parse_stack(self.t, cache_data=cache_data)402        rsrc = stack.defn['SwiftContainer']403        self.assertEqual('xyz_convg', rsrc.FnGetRefId())404    @mock.patch('swiftclient.client.Connection.head_account')405    @mock.patch('swiftclient.client.Connection.head_container')406    @mock.patch('swiftclient.client.Connection.put_container')407    def test_parse_live_resource_data(self, mock_put, mock_container,408                                      mock_account):409        stack = utils.parse_stack(self.t)410        container = self._create_container(411            stack, definition_name="SwiftContainerWebsite")412        mock_container.return_value = {413            'x-container-read': '.r:*',414            'x-container-meta-web-index': 'index.html',415            'x-container-meta-web-error': 'error.html',416            'x-container-meta-login': 'login.html'417        }418        mock_account.return_value = {}419        live_state = container.parse_live_resource_data(420            container.properties, container.get_live_resource_data())421        # live state properties values should be equal to current resource422        # properties values...
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.