How to use stack_id method in localstack

Best Python code snippet using localstack_python

heat.py

Source:heat.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 contextlib13from django.conf import settings14from heatclient import client as heat_client15from heatclient.common import template_format16from heatclient.common import template_utils17from heatclient.common import utils as heat_utils18from oslo_serialization import jsonutils19import six20from six.moves.urllib import request21from horizon import exceptions22from horizon.utils import functions as utils23from horizon.utils.memoized import memoized24from openstack_dashboard.api import base25from openstack_dashboard.contrib.developer.profiler import api as profiler26def format_parameters(params):27    parameters = {}28    for count, p in enumerate(params, 1):29        parameters['Parameters.member.%d.ParameterKey' % count] = p30        parameters['Parameters.member.%d.ParameterValue' % count] = params[p]31    return parameters32@memoized33def heatclient(request, password=None):34    api_version = "1"35    insecure = getattr(settings, 'OPENSTACK_SSL_NO_VERIFY', False)36    cacert = getattr(settings, 'OPENSTACK_SSL_CACERT', None)37    endpoint = base.url_for(request, 'orchestration')38    kwargs = {39        'token': request.user.token.id,40        'insecure': insecure,41        'ca_file': cacert,42        'username': request.user.username,43        'password': password44        # 'timeout': args.timeout,45        # 'ca_file': args.ca_file,46        # 'cert_file': args.cert_file,47        # 'key_file': args.key_file,48    }49    client = heat_client.Client(api_version, endpoint, **kwargs)50    client.format_parameters = format_parameters51    return client52@profiler.trace53def stacks_list(request, marker=None, sort_dir='desc', sort_key='created_at',54                paginate=False, filters=None):55    limit = getattr(settings, 'API_RESULT_LIMIT', 1000)56    page_size = utils.get_page_size(request)57    if paginate:58        request_size = page_size + 159    else:60        request_size = limit61    kwargs = {'sort_dir': sort_dir, 'sort_key': sort_key}62    if marker:63        kwargs['marker'] = marker64    if filters:65        kwargs.update(filters)66        if 'status' in kwargs:67            kwargs['status'] = kwargs['status'].replace(' ', '_').upper()68    stacks_iter = heatclient(request).stacks.list(limit=request_size,69                                                  **kwargs)70    has_prev_data = False71    has_more_data = False72    stacks = list(stacks_iter)73    if paginate:74        if len(stacks) > page_size:75            stacks.pop()76            has_more_data = True77            if marker is not None:78                has_prev_data = True79        elif sort_dir == 'asc' and marker is not None:80            has_more_data = True81        elif marker is not None:82            has_prev_data = True83    return (stacks, has_more_data, has_prev_data)84def _ignore_if(key, value):85    if key != 'get_file' and key != 'type':86        return True87    if not isinstance(value, six.string_types):88        return True89    if (key == 'type' and90            not value.endswith(('.yaml', '.template'))):91        return True92    return False93@profiler.trace94def get_template_files(template_data=None, template_url=None):95    if template_data:96        tpl = template_data97    elif template_url:98        with contextlib.closing(request.urlopen(template_url)) as u:99            tpl = u.read()100    else:101        return {}, None102    if not tpl:103        return {}, None104    if isinstance(tpl, six.binary_type):105        tpl = tpl.decode('utf-8')106    template = template_format.parse(tpl)107    files = {}108    _get_file_contents(template, files)109    return files, template110def _get_file_contents(from_data, files):111    if not isinstance(from_data, (dict, list)):112        return113    if isinstance(from_data, dict):114        recurse_data = from_data.values()115        for key, value in from_data.items():116            if _ignore_if(key, value):117                continue118            if not value.startswith(('http://', 'https://')):119                raise exceptions.GetFileError(value, 'get_file')120            if value not in files:121                file_content = heat_utils.read_url_content(value)122                if template_utils.is_template(file_content):123                    template = get_template_files(template_url=value)[1]124                    file_content = jsonutils.dumps(template)125                files[value] = file_content126    else:127        recurse_data = from_data128    for value in recurse_data:129        _get_file_contents(value, files)130@profiler.trace131def stack_delete(request, stack_id):132    return heatclient(request).stacks.delete(stack_id)133@profiler.trace134def stack_get(request, stack_id):135    return heatclient(request).stacks.get(stack_id)136@profiler.trace137def template_get(request, stack_id):138    return heatclient(request).stacks.template(stack_id)139@profiler.trace140def stack_create(request, password=None, **kwargs):141    return heatclient(request, password).stacks.create(**kwargs)142@profiler.trace143def stack_preview(request, password=None, **kwargs):144    return heatclient(request, password).stacks.preview(**kwargs)145@profiler.trace146def stack_update(request, stack_id, password=None, **kwargs):147    return heatclient(request, password).stacks.update(stack_id, **kwargs)148@profiler.trace149def snapshot_create(request, stack_id):150    return heatclient(request).stacks.snapshot(stack_id)151@profiler.trace152def snapshot_list(request, stack_id):153    return heatclient(request).stacks.snapshot_list(stack_id)154@profiler.trace155def snapshot_show(request, stack_id, snapshot_id):156    return heatclient(request).stacks.snapshot_show(stack_id, snapshot_id)157@profiler.trace158def snapshot_delete(request, stack_id, snapshot_id):159    return heatclient(request).stacks.snapshot_delete(stack_id, snapshot_id)160@profiler.trace161def events_list(request, stack_name):162    return heatclient(request).events.list(stack_name)163@profiler.trace164def resources_list(request, stack_name):165    return heatclient(request).resources.list(stack_name)166@profiler.trace167def resource_get(request, stack_id, resource_name):168    return heatclient(request).resources.get(stack_id, resource_name)169@profiler.trace170def resource_metadata_get(request, stack_id, resource_name):171    return heatclient(request).resources.metadata(stack_id, resource_name)172@profiler.trace173def template_validate(request, **kwargs):174    return heatclient(request).stacks.validate(**kwargs)175@profiler.trace176def action_check(request, stack_id):177    return heatclient(request).actions.check(stack_id)178@profiler.trace179def action_suspend(request, stack_id):180    return heatclient(request).actions.suspend(stack_id)181@profiler.trace182def action_resume(request, stack_id):183    return heatclient(request).actions.resume(stack_id)184@profiler.trace185def resource_types_list(request, filters=None):186    return heatclient(request).resource_types.list(filters=filters)187@profiler.trace188def resource_type_get(request, resource_type):189    return heatclient(request).resource_types.get(resource_type)190@profiler.trace191def service_list(request):192    return heatclient(request).services.list()193@profiler.trace194def template_version_list(request):195    return heatclient(request).template_versions.list()196@profiler.trace197def template_function_list(request, template_version):...

test_stack_operations.py

Source:test_stack_operations.py

1"""2Copyright 2016 Rackspace3Licensed under the Apache License, Version 2.0 (the "License");4you may not use this file except in compliance with the License.5You may obtain a copy of the License at6    http://www.apache.org/licenses/LICENSE-2.07Unless required by applicable law or agreed to in writing, software8distributed under the License is distributed on an "AS IS" BASIS,9WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.10See the License for the specific language governing permissions and11limitations under the License.12"""13from heattests import base14class TestStackOperations(base.TestBase):15    def setUp(self):16        super(TestStackOperations, self).setUp()17    def test_list_stack(self):18        resp = self.heat_client.list_stacks()19        self.assertEqual(resp.status_code, 200)20    def test_create_stack(self):21        stack_name = self.generate_random_string(prefix='Heat_QE')22        temp_url = self.config.template_url23        resp = self.heat_client.create_stack(stack_name=stack_name,24                                             template_url=temp_url)25        self.assertEqual(resp.status_code, 201)26        self.stack_list[resp.json()['stack']['id']] = stack_name27    def test_find_stack(self):28        stack_name = self.generate_random_string(prefix='Heat_QE')29        temp_url = self.config.template_url30        resp = self.heat_client.create_stack(stack_name=stack_name,31                                             template_url=temp_url)32        self.assertEqual(resp.status_code, 201)33        body = resp.json()34        stack_id = body['stack']['id']35        self.stack_list[stack_id] = stack_name36        resp = self.heat_client.find_stack(stack_id=stack_id)37        body = resp.json()38        self.assertEqual(resp.status_code, 200)39        self.assertEqual(body['stack']['stack_name'], stack_name)40        self.assertEqual(body['stack']['id'], stack_id)41        self.assertEqual(body['stack']['stack_user_project_id'],42                         self.user_project_id)43    def test_show_stack(self):44        stack_name = self.generate_random_string(prefix='Heat_QE')45        temp_url = self.config.template_url46        resp = self.heat_client.create_stack(stack_name=stack_name,47                                             template_url=temp_url)48        self.assertEqual(resp.status_code, 201)49        body = resp.json()50        stack_id = body['stack']['id']51        self.stack_list[stack_id] = stack_name52        resp = self.heat_client.show_stack(stack_name=stack_name,53                                           stack_id=stack_id)54        body = resp.json()55        self.assertEqual(resp.status_code, 200)56        self.assertEqual(body['stack']['stack_name'], stack_name)57        self.assertEqual(body['stack']['id'], stack_id)58        self.assertEqual(body['stack']['stack_user_project_id'],59                         self.user_project_id)60    def test_preview_stack(self):61        stack_name = self.generate_random_string(prefix='Heat_QE')62        temp_url = self.config.template_url63        resp = self.heat_client.preview_stack(stack_name=stack_name,64                                              template_url=temp_url)65        self.assertEqual(resp.status_code, 200)66    def test_delete_stack(self):67        stack_name = self.generate_random_string(prefix='Heat_QE')68        temp_url = self.config.template_url69        resp = self.heat_client.create_stack(stack_name=stack_name,70                                             template_url=temp_url)71        self.assertEqual(resp.status_code, 201)72        body = resp.json()73        stack_id = body['stack']['id']74        resp = self.heat_client.delete_stack(stack_name=stack_name,75                                             stack_id=stack_id)76        self.assertEqual(resp.status_code, 204)77    def test_update_stack(self):78        stack_name = self.generate_random_string(prefix='Heat_QE')79        temp_url = self.config.template_url80        resp = self.heat_client.create_stack(stack_name=stack_name,81                                             template_url=temp_url)82        self.assertEqual(resp.status_code, 201)83        body = resp.json()84        stack_id = body['stack']['id']85        self.stack_list[stack_id] = stack_name86        resp = self.heat_client.update_stack(stack_name=stack_name,87                                             stack_id=stack_id,88                                             template_url=temp_url)89        self.assertEqual(resp.status_code, 202)90    def test_abandon_stack(self):91        self.skipTest('Not working - Test needs to be updated')92        # Add the use of the test.yaml and check nova to ensure server exists93        stack_name = self.generate_random_string(prefix='Heat_QE')94        temp_url = 'https://raw.githubusercontent.com/BeenzSyed/heat-ci/' \95                   'add_test_template/dev/test.yaml'96        params = {'flavor': '1 GB Performance',97                  'image': '9d29f10e-4fc2-4556-8d25-532d1784329a'}98        resp = self.heat_client.create_stack(stack_name=stack_name,99                                             template_url=temp_url,100                                             parameters=params)101        self.assertEqual(resp.status_code, 201)102        body = resp.json()103        stack_id = body['stack']['id']104        resp = self.heat_client.abandon_stack(stack_name=stack_name,105                                              stack_id=stack_id)106        self.assertEqual(resp.status_code, 200)107    def test_adopt_stack(self):108        self.skipTest('Not working - Adopt stack might be taken out of Heat')109        stack_name = self.generate_random_string(prefix='Heat_QE')110        temp_url = 'https://raw.githubusercontent.com/BeenzSyed/heat-ci/' \111                   'add_test_template/dev/test.yaml'112        params = {'flavor': '1 GB Performance',113                  'image': '9d29f10e-4fc2-4556-8d25-532d1784329a'}114        resp = self.heat_client.create_stack(stack_name=stack_name,115                                             template_url=temp_url,116                                             parameters=params)117        self.assertEqual(resp.status_code, 201)118        body = resp.json()119        stack_id = body['stack']['id']120        self.stack_list.append(stack_id)121        self.stack_url = resp.headers['location']122        # Wait to ensure that stack has completed123        self.heat_client.wait_for_stack_status(124            location=self.stack_url,125            status='CREATE_COMPLETE',126            abort_on_status='FAILED',127            retry_interval=5,128            retry_timeout=300)129        resp = self.heat_client.abandon_stack(stack_name=stack_name,130                                              stack_id=stack_id)131        self.assertEqual(resp.status_code, 200)132        body = resp.json()133        resp = self.heat_client.adopt_stack(stack_name, adopt_stack_data=body,134                                            template_url=temp_url)135        self.assertEqual(resp.status_code, 201)136    def tearDown(self):...

app.py

Source:app.py

1"""The Endpoints to operate on the STACKS in the context of an RPN calculator"""2from flask import Flask, jsonify, abort3import operator4from flask_swagger_ui import get_swaggerui_blueprint5OPERATORS = {'+': operator.add, 6             '-': operator.sub, 7             '*': operator.mul, 8             'd': operator.truediv} # using the string literal "d" and not "/" because "/" is a special character in the context of urls 9                                    # and can not be escaped with Swagger10                                    # https://github.com/swagger-api/swagger-ui/issues/459111                                    # https://stackoverflow.com/questions/42335178/swagger-wildcard-path-parameters12                                    # --> "This [use of slash in url query params] is not supported as of OpenAPI 3.1"13                                    # It could be nonetheless encoded differently and passed through a more traditional HTML, JS frontend14STACKS                  = {}15class StackCount:16    def __init__(self, stack_count):17        self.stack_count = stack_count18    19    def add_stack(self):20        self.stack_count +=121app                     = Flask(__name__)22SWAGGER_URL             = '/swagger'23API_URL                 = '/static/swagger.json'24app.secret_key          = "super secret key"25swaggerui_blueprint = get_swaggerui_blueprint(26    SWAGGER_URL,27    API_URL,28    config={29        'app_name': 'rpn_flask'30    }31)32app.register_blueprint(swaggerui_blueprint, url_prefix = SWAGGER_URL)33@app.route("/", methods=['GET'])34def Home():35    """36    """37    return "<p>Hello! <br> \38            <b><a href='http://127.0.0.1:5000/swagger'>Click here</a></b> <br> \39            to go to Swagger UI </p>", 20140@app.route("/rpn/op", methods=['GET'])41def get_list_operands():42    """43    Lists all the available operands44    @return: 201: an object with the literal string "list_operands" as key and an array of available operators as a value45    with application/json mimetype.46    """47    list_op = list(OPERATORS.keys())48    49    return jsonify({'list_operators':list_op, 50                    'information':'the letter "d" to be used as a division operator'}), 20151@app.route("/rpn/stack/<int:stack_id>", methods=['GET'])52def get_stack(stack_id):53    """54    Lists existing stack(s) 55    56    @param stack_id: get : ID of the stack on which we add an operand or apply an operator57    @return: 200: an array representing the content of the choosen stack58    with application/json mimetype.59    @raise 404: if stack_id not found60    """61    if stack_id in STACKS.keys():62        return jsonify(STACKS[stack_id])63    else:64        abort(404)65    66@app.route("/rpn/stack", methods=['GET'])67def get_stack_list():68    """69    Lists existing stack(s) 70    @return: 200: an object representing the existing stacks and their content71    with application/json mimetype.72    """73    print('/rpn/stack get')74    return jsonify(STACKS)75@app.route("/rpn/stack", methods=['POST'])76def create_stack():77    """78    Create a brand new stack, a stack identifier will be automtically generated79    @return: 201: an empty payload.80    """81    obj_stack_count.stack_count         += 182    STACKS[obj_stack_count.stack_count] = []83    return '', 20184@app.route("/rpn/stack/<int:stack_id>/val/<int:val>", methods=['POST'])85def push_val_to_stack(stack_id, val):86    """87    Adds a value to a stack, given a stack id88    @param val: post : Operand - or "value" - to be added to the stack 89    @param stack_id: post : ID of the stack on which we add an operand or apply an operator90    @return: 201: an empty payload.91    @raise 404: if stack_id not found92    """93    if stack_id in STACKS.keys():94        STACKS[stack_id].append(val)95    else:96        abort(404)97    return '', 20198@app.route("/rpn/stack/<int:stack_id>", methods=['DELETE'])99def delete_stack(stack_id):100    """101    Deletes a stack, given a stack id102    @param stack_id: delete : ID of the stack that we want to delete103    @return: 204: an empty payload.104    @raise 404: if stack_id not found105    """106    if stack_id not in STACKS.keys():107        abort(404)108    del STACKS[stack_id]109    return '', 204110@app.route("/rpn/op/<string:operator>/stack/<int:stack_id>", methods=['POST'])111def apply_operator(operator, stack_id):112    """113    Applies operator entered by user to two last values of a given stack114    @param operator: post : Operator to be applied on two last operands of the stack115    @param stack_id: post : ID of the stack on which we add an operand or apply an operator116    @return: 201: an empty payload.117    """118    first_pop       = STACKS[stack_id].pop() 119    second_pop      = STACKS[stack_id].pop()120    result          = OPERATORS[operator](second_pop, first_pop) 121    STACKS[stack_id].append(result)122    return '', 201123@app.route("/rpn/clear_stack/<int:stack_id>", methods=['DELETE'])124def clear_stack(stack_id):125    """126    Clears a stack, given a certain stack ID127    @param stack_id: delete : ID of the stack that we want to delete128    @return: 204: an empty payload.129    """130    STACKS[stack_id].clear()131    return '', 204132if __name__ == '__main__':133    obj_stack_count         = StackCount(0)...

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.