KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use delete_container method in tempest
Best Python code snippet using tempest_python
test_s3.py
Source:test_s3.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 mock14import six15from oslo_config import cfg16import swiftclient.client as sc17from heat.common import exception18from heat.common import template_format19from heat.engine.resources.aws.s3 import s320from heat.engine import scheduler21from heat.tests import common22from heat.tests import utils23swift_template = '''24{25 "AWSTemplateFormatVersion" : "2010-09-09",26 "Description" : "Template to test S3 Bucket resources",27 "Resources" : {28 "S3BucketWebsite" : {29 "Type" : "AWS::S3::Bucket",30 "DeletionPolicy" : "Delete",31 "Properties" : {32 "AccessControl" : "PublicRead",33 "WebsiteConfiguration" : {34 "IndexDocument" : "index.html",35 "ErrorDocument" : "error.html"36 }37 }38 },39 "SwiftContainer": {40 "Type": "OS::Swift::Container",41 "Properties": {42 "S3Bucket": {"Ref" : "S3Bucket"},43 }44 },45 "S3Bucket" : {46 "Type" : "AWS::S3::Bucket",47 "Properties" : {48 "AccessControl" : "Private"49 }50 },51 "S3Bucket_with_tags" : {52 "Type" : "AWS::S3::Bucket",53 "Properties" : {54 "Tags" : [{"Key": "greeting", "Value": "hello"},55 {"Key": "location", "Value": "here"}]56 }57 }58 }59}60'''61class s3Test(common.HeatTestCase):62 def setUp(self):63 super(s3Test, self).setUp()64 self.mock_con = mock.Mock(spec=sc.Connection)65 self.patchobject(s3.S3Bucket, 'client',66 return_value=self.mock_con)67 def create_resource(self, t, stack, resource_name):68 resource_defns = stack.t.resource_definitions(stack)69 rsrc = s3.S3Bucket('test_resource',70 resource_defns[resource_name],71 stack)72 scheduler.TaskRunner(rsrc.create)()73 self.assertEqual((rsrc.CREATE, rsrc.COMPLETE), rsrc.state)74 return rsrc75 def test_attributes(self):76 t = template_format.parse(swift_template)77 stack = utils.parse_stack(t)78 container_name = utils.PhysName(stack.name, 'test_resource')79 self.mock_con.put_container.return_value = None80 self.mock_con.get_auth.return_value = (81 'http://server.test:8080/v_2', None)82 self.mock_con.delete_container.return_value = None83 rsrc = self.create_resource(t, stack, 'S3Bucket')84 ref_id = rsrc.FnGetRefId()85 self.assertEqual(container_name, ref_id)86 self.assertEqual('server.test', rsrc.FnGetAtt('DomainName'))87 url = 'http://server.test:8080/v_2/%s' % ref_id88 self.assertEqual(url, rsrc.FnGetAtt('WebsiteURL'))89 self.assertRaises(exception.InvalidTemplateAttribute,90 rsrc.FnGetAtt, 'Foo')91 scheduler.TaskRunner(rsrc.delete)()92 self.mock_con.put_container.assert_called_once_with(93 container_name,94 {'X-Container-Write': 'test_tenant:test_username',95 'X-Container-Read': 'test_tenant:test_username'}96 )97 self.mock_con.get_auth.assert_called_with()98 self.assertEqual(2, self.mock_con.get_auth.call_count)99 self.mock_con.delete_container.assert_called_once_with(container_name)100 def test_public_read(self):101 t = template_format.parse(swift_template)102 properties = t['Resources']['S3Bucket']['Properties']103 properties['AccessControl'] = 'PublicRead'104 stack = utils.parse_stack(t)105 container_name = utils.PhysName(stack.name, 'test_resource')106 self.mock_con.put_container.return_value = None107 self.mock_con.delete_container.return_value = None108 rsrc = self.create_resource(t, stack, 'S3Bucket')109 scheduler.TaskRunner(rsrc.delete)()110 self.mock_con.put_container.assert_called_once_with(111 utils.PhysName(stack.name, 'test_resource'),112 {'X-Container-Write': 'test_tenant:test_username',113 'X-Container-Read': '.r:*'})114 self.mock_con.delete_container.assert_called_once_with(container_name)115 def test_tags(self):116 t = template_format.parse(swift_template)117 stack = utils.parse_stack(t)118 container_name = utils.PhysName(stack.name, 'test_resource')119 self.mock_con.put_container.return_value = None120 self.mock_con.delete_container.return_value = None121 rsrc = self.create_resource(t, stack, 'S3Bucket_with_tags')122 scheduler.TaskRunner(rsrc.delete)()123 self.mock_con.put_container.assert_called_once_with(124 utils.PhysName(stack.name, 'test_resource'),125 {'X-Container-Write': 'test_tenant:test_username',126 'X-Container-Read': 'test_tenant:test_username',127 'X-Container-Meta-S3-Tag-greeting': 'hello',128 'X-Container-Meta-S3-Tag-location': 'here'})129 self.mock_con.delete_container.assert_called_once_with(container_name)130 def test_public_read_write(self):131 t = template_format.parse(swift_template)132 properties = t['Resources']['S3Bucket']['Properties']133 properties['AccessControl'] = 'PublicReadWrite'134 stack = utils.parse_stack(t)135 container_name = utils.PhysName(stack.name, 'test_resource')136 self.mock_con.put_container.return_value = None137 self.mock_con.delete_container.return_value = None138 rsrc = self.create_resource(t, stack, 'S3Bucket')139 scheduler.TaskRunner(rsrc.delete)()140 self.mock_con.put_container.assert_called_once_with(141 container_name,142 {'X-Container-Write': '.r:*',143 'X-Container-Read': '.r:*'})144 self.mock_con.delete_container.assert_called_once_with(container_name)145 def test_authenticated_read(self):146 t = template_format.parse(swift_template)147 properties = t['Resources']['S3Bucket']['Properties']148 properties['AccessControl'] = 'AuthenticatedRead'149 stack = utils.parse_stack(t)150 container_name = utils.PhysName(stack.name, 'test_resource')151 self.mock_con.put_container.return_value = None152 self.mock_con.delete_container.return_value = None153 rsrc = self.create_resource(t, stack, 'S3Bucket')154 scheduler.TaskRunner(rsrc.delete)()155 self.mock_con.put_container.assert_called_once_with(156 container_name,157 {'X-Container-Write': 'test_tenant:test_username',158 'X-Container-Read': 'test_tenant'})159 self.mock_con.delete_container.assert_called_once_with(container_name)160 def test_website(self):161 t = template_format.parse(swift_template)162 stack = utils.parse_stack(t)163 container_name = utils.PhysName(stack.name, 'test_resource')164 self.mock_con.put_container.return_value = None165 self.mock_con.delete_container.return_value = None166 rsrc = self.create_resource(t, stack, 'S3BucketWebsite')167 scheduler.TaskRunner(rsrc.delete)()168 self.mock_con.put_container.assert_called_once_with(169 container_name,170 {'X-Container-Meta-Web-Error': 'error.html',171 'X-Container-Meta-Web-Index': 'index.html',172 'X-Container-Write': 'test_tenant:test_username',173 'X-Container-Read': '.r:*'})174 self.mock_con.delete_container.assert_called_once_with(container_name)175 def test_delete_exception(self):176 t = template_format.parse(swift_template)177 stack = utils.parse_stack(t)178 container_name = utils.PhysName(stack.name, 'test_resource')179 self.mock_con.put_container.return_value = None180 self.mock_con.delete_container.side_effect = sc.ClientException(181 'Test Delete Failure')182 rsrc = self.create_resource(t, stack, 'S3Bucket')183 self.assertRaises(exception.ResourceFailure,184 scheduler.TaskRunner(rsrc.delete))185 self.mock_con.put_container.assert_called_once_with(186 container_name,187 {'X-Container-Write': 'test_tenant:test_username',188 'X-Container-Read': 'test_tenant:test_username'})189 self.mock_con.delete_container.assert_called_once_with(container_name)190 def test_delete_not_found(self):191 t = template_format.parse(swift_template)192 stack = utils.parse_stack(t)193 container_name = utils.PhysName(stack.name, 'test_resource')194 self.mock_con.put_container.return_value = None195 self.mock_con.delete_container.side_effect = sc.ClientException(196 'Gone', http_status=404)197 rsrc = self.create_resource(t, stack, 'S3Bucket')198 scheduler.TaskRunner(rsrc.delete)()199 self.mock_con.put_container.assert_called_once_with(200 container_name,201 {'X-Container-Write': 'test_tenant:test_username',202 'X-Container-Read': 'test_tenant:test_username'})203 self.mock_con.delete_container.assert_called_once_with(container_name)204 def test_delete_conflict_not_empty(self):205 t = template_format.parse(swift_template)206 stack = utils.parse_stack(t)207 container_name = utils.PhysName(stack.name, 'test_resource')208 self.mock_con.put_container.return_value = None209 self.mock_con.delete_container.side_effect = sc.ClientException(210 'Not empty', http_status=409)211 self.mock_con.get_container.return_value = (212 {'name': container_name}, [{'name': 'test_object'}])213 rsrc = self.create_resource(t, stack, 'S3Bucket')214 deleter = scheduler.TaskRunner(rsrc.delete)215 ex = self.assertRaises(exception.ResourceFailure, deleter)216 self.assertIn("ResourceActionNotSupported: resources.test_resource: "217 "The bucket you tried to delete is not empty",218 six.text_type(ex))219 self.mock_con.put_container.assert_called_once_with(220 container_name,221 {'X-Container-Write': 'test_tenant:test_username',222 'X-Container-Read': 'test_tenant:test_username'})223 self.mock_con.delete_container.assert_called_once_with(container_name)224 self.mock_con.get_container.assert_called_once_with(container_name)225 def test_delete_conflict_empty(self):226 cfg.CONF.set_override('action_retry_limit', 0)227 t = template_format.parse(swift_template)228 stack = utils.parse_stack(t)229 container_name = utils.PhysName(stack.name, 'test_resource')230 self.mock_con.put_container.return_value = None231 self.mock_con.delete_container.side_effect = sc.ClientException(232 'Conflict', http_status=409)233 self.mock_con.get_container.return_value = (234 {'name': container_name}, [])235 rsrc = self.create_resource(t, stack, 'S3Bucket')236 deleter = scheduler.TaskRunner(rsrc.delete)237 ex = self.assertRaises(exception.ResourceFailure, deleter)238 self.assertIn("Conflict", six.text_type(ex))239 self.mock_con.put_container.assert_called_once_with(240 container_name,241 {'X-Container-Write': 'test_tenant:test_username',242 'X-Container-Read': 'test_tenant:test_username'})243 self.mock_con.delete_container.assert_called_once_with(container_name)244 self.mock_con.get_container.assert_called_once_with(container_name)245 def test_delete_retain(self):246 t = template_format.parse(swift_template)247 bucket = t['Resources']['S3Bucket']248 bucket['DeletionPolicy'] = 'Retain'249 stack = utils.parse_stack(t)250 # first run, with retain policy251 self.mock_con.put_container.return_value = None252 rsrc = self.create_resource(t, stack, 'S3Bucket')253 scheduler.TaskRunner(rsrc.delete)()254 self.assertEqual((rsrc.DELETE, rsrc.COMPLETE), rsrc.state)255 self.mock_con.put_container.assert_called_once_with(256 utils.PhysName(stack.name, 'test_resource'),257 {'X-Container-Write': 'test_tenant:test_username',...
doc_store.py
Source:doc_store.py
1import time2import logging3import subprocess4import requests5from pathlib import Path6logger = logging.getLogger(__name__)7ELASTICSEARCH_CONTAINER_NAME = "elasticsearch"8OPENSEARCH_CONTAINER_NAME = "opensearch"9MILVUS1_CONTAINER_NAME = "milvus1"10WEAVIATE_CONTAINER_NAME = "weaviate"11def launch_es(sleep=15, delete_existing=False):12 # Start an Elasticsearch server via Docker13 logger.debug("Starting Elasticsearch ...")14 if delete_existing:15 _ = subprocess.run(16 [f"docker rm --force {ELASTICSEARCH_CONTAINER_NAME}"],17 shell=True,18 stdout=subprocess.DEVNULL)19 status = subprocess.run(20 [21 f'docker run -d -p 9200:9200 -e "discovery.type=single-node" --name {ELASTICSEARCH_CONTAINER_NAME} elasticsearch:7.9.2'22 ],23 shell=True,24 )25 if status.returncode:26 logger.warning(27 "Tried to start Elasticsearch through Docker but this failed. "28 "It is likely that there is already an existing Elasticsearch instance running. "29 )30 else:31 time.sleep(sleep)32def launch_opensearch(sleep=15, delete_existing=False):33 # Start an OpenSearch server via docker34 logger.debug("Starting OpenSearch...")35 # This line is needed since it is not possible to start a new docker container with the name opensearch if there is a stopped image with the same now36 # docker rm only succeeds if the container is stopped, not if it is running37 if delete_existing:38 _ = subprocess.run([f"docker rm --force {OPENSEARCH_CONTAINER_NAME}"],39 shell=True,40 stdout=subprocess.DEVNULL)41 status = subprocess.run(42 [43 f'docker run -d -p 9201:9200 -p 9600:9600 -e "discovery.type=single-node" --name {OPENSEARCH_CONTAINER_NAME} opensearchproject/opensearch:1.2.4'44 ],45 shell=True,46 )47 if status.returncode:48 logger.warning(49 "Tried to start OpenSearch through Docker but this failed. "50 "It is likely that there is already an existing OpenSearch instance running. "51 )52 else:53 time.sleep(sleep)54def launch_weaviate(sleep=15):55 # Start a Weaviate server via Docker56 logger.debug("Starting Weaviate ...")57 status = subprocess.run(58 [59 "docker run -d -p 8080:8080 --env AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED='true' --env PERSISTENCE_DATA_PATH='/var/lib/weaviate' --name {WEAVIATE_CONTAINER_NAME} semitechnologies/weaviate:1.7.2"60 ],61 shell=True,62 )63 if status.returncode:64 logger.warning(65 "Tried to start Weaviate through Docker but this failed. "66 "It is likely that there is already an existing Weaviate instance running. "67 )68 else:69 time.sleep(sleep)70def stop_container(container_name, delete_container=False):71 logger.debug(f"Stopping {container_name}...")72 status = subprocess.run([f"docker stop {container_name}"], shell=True)73 if status.returncode:74 logger.warning(75 f"Tried to stop {container_name} but this failed. "76 f"It is likely that there was no Docker container with the name {container_name}"77 )78 if delete_container:79 status = subprocess.run([f"docker rm {container_name}"], shell=True)80def stop_opensearch(delete_container=False):81 stop_container(OPENSEARCH_CONTAINER_NAME, delete_container)82def stop_elasticsearch(delete_container=False):83 stop_container(ELASTICSEARCH_CONTAINER_NAME, delete_container)84def stop_milvus(delete_container=False):85 stop_container(MILVUS1_CONTAINER_NAME, delete_container)86def stop_weaviate(delete_container=False):87 stop_container(WEAVIATE_CONTAINER_NAME, delete_container)88def stop_service(document_store, delete_container=False):89 ds_class = str(type(document_store))90 if "OpenSearchDocumentStore" in ds_class:91 stop_opensearch(delete_container)92 elif "ElasticsearchDocumentStore" in ds_class:93 stop_elasticsearch(delete_container)94 elif "MilvusDocumentStore" in ds_class:95 stop_milvus(delete_container)96 elif "WeaviateDocumentStore" in ds_class:97 stop_weaviate(delete_container)98 else:99 logger.warning(100 f"No support yet for auto stopping the service behind a {type(document_store)}"101 )102def launch_milvus(sleep=15, delete_existing=False):103 # Start a Milvus server via docker104 logger.debug("Starting Milvus ...")105 milvus_dir = Path.home() / "milvus"106 milvus_dir.mkdir(exist_ok=True)107 request = requests.get(108 "https://github.com/milvus-io/milvus/releases/download/v2.0.0/milvus-standalone-docker-compose.yml"109 )110 with open(milvus_dir / "docker-compose.yml", "wb") as f:111 f.write(request.content)112 status = subprocess.run(["cd /home/$USER/milvus/ && docker-compose up -d"],113 shell=True)114 if status.returncode:115 logger.warning(116 "Tried to start Milvus through Docker but this failed. "117 "It is likely that there is already an existing Milvus instance running. "118 )119 else:120 time.sleep(sleep)121def launch_milvus1(sleep=15):122 # Start a Milvus (version <2.0.0) server via docker123 logger.debug("Starting Milvus ...")124 logger.warning(125 "Automatic Milvus config creation not yet implemented. "126 "If you are starting Milvus using launch_milvus(), "127 "make sure you have a properly populated milvus/conf folder. "128 "See (https://milvus.io/docs/v1.0.0/milvus_docker-cpu.md) for more details."129 )130 status = subprocess.run(131 [132 f"docker run -d --name {MILVUS1_CONTAINER_NAME} \133 -p 19530:19530 \134 -p 19121:19121 \135 milvusdb/milvus:1.1.0-cpu-d050721-5e559c"136 ],137 shell=True,138 )139 if status.returncode:140 logger.warning(141 "Tried to start Milvus through Docker but this failed. "142 "It is likely that there is already an existing Milvus instance running. "143 )144 else:...
clear_docker.py
Source:clear_docker.py
1import os2def delete_image(image_name):3 return f"gnome-terminal -e 'bash -c \"docker image rm -f {image_name}; bash\" '" 4def delete_container(container_name):5 return f"gnome-terminal -e 'bash -c \"docker container rm -f {container_name}; bash\" '" 6os.system(delete_container('addaghor_react_1'))7os.system(delete_container('addaghor_post_1'))8os.system(delete_container('addaghor_consumer_1'))9os.system(delete_image('addaghor_react:latest'))10os.system(delete_image('addaghor_post:latest'))...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tempest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
