Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.