KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use rollback_stack method in localstack
Best Python code snippet using localstack_python
commands.py
Source:commands.py 
1# vim: tabstop=4 shiftwidth=4 softtabstop=42# Copyright 2012 OpenStack LLC.3# All Rights Reserved.4#5# Licensed under the Apache License, Version 2.0 (the "License"); you may6# not use this file except in compliance with the License. You may obtain7# a copy of the License at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the14# License for the specific language governing permissions and limitations15# under the License.16import pdb17import functools18import logging19import types20import kulcloud.db.api as db_api21from kulcloud import utils22LOG = logging.getLogger(__name__)23class RollbackContext(object):24 def __init__(self):25 self.rollback_stack = []26 def add_rollback(self, rollback):27 self.rollback_stack.append(rollback)28class RollbackContextManager(object):29 def __init__(self, context=None):30 if context is None:31 self.context = RollbackContext()32 else:33 self.context = context34 def __enter__(self):35 return self.context36 def __exit__(self, exc_type, exc_value, exc_tb):37 good = exc_type is None38 if not good:39 LOG.error("Rollback because of: %s", exc_value,40 exc_info=(exc_value, exc_type, exc_tb))41 rollback_stack = self.context.rollback_stack42 while rollback_stack:43 rollback_stack.pop()(good)44 if not good:45 raise exc_type, exc_value, exc_tb46class Rollback(Exception):47 pass48def with_rollback(func):49 @functools.wraps(func)50 def __inner(ctx, *args, **kwargs):51 gen = func(ctx, *args, **kwargs)52 if not isinstance(gen, types.GeneratorType):53 LOG.critical("Expected generator, got %r instead", gen)54 raise RuntimeError(55 "Commands with rollback must be generator functions")56 try:57 res = gen.next()58 except StopIteration:59 LOG.warn("Command %s finished w/o yielding", func.__name__)60 else:61 def fin(good):62 if good:63 gen.close()64 else:65 try:66 gen.throw(Rollback)67 except Rollback:68 pass69 except Exception:70 LOG.exception("Exception during rollback.")71 ctx.add_rollback(fin)72 return res73 return __inner74def ignore_exceptions(func):75 @functools.wraps(func)76 def __inner(*args, **kwargs):77 try:78 return func(*args, **kwargs)79 except Exception:80 LOG.exception("Got exception while executing %s. Ignored.",81 func.__name__)82 return __inner83def rserver_exist(ctx, rs):84 LOG.debug("Checking rserver already exist or not rserver : %s", rs)85 LOG.debug("RServer parent_id: %s", rs['parent_id'])86 return ctx.device.real_server_exist(rs)87@with_rollback88def create_rserver(ctx, rs):89 try:90 # We can't create multiple RS with the same IP. So parent_id points to91 # RS which already deployed and has this IP92 LOG.debug("Creating rserver command execution with rserver: %s", rs)93 LOG.debug("RServer parent_id: %s", rs['parent_id'])94 if not rs['parent_id']:95 ctx.device.create_real_server(rs)96 rs['deployed'] = 'True'97 db_api.server_update(ctx.conf, rs['id'], rs)98 yield99 except Exception:100 with utils.save_and_reraise_exception():101 ctx.device.delete_real_server(rs)102 rs['deployed'] = 'False'103 db_api.server_update(ctx.conf, rs['id'], rs)104@ignore_exceptions105def delete_rserver(ctx, rs):106 rss = []107 LOG.debug("Got delete RS request")108 if rs['parent_id'] == None:109 rss = db_api.server_get_all_by_parent_id(ctx.conf, rs['id'])110 LOG.debug("List of servers: %s", rss)111 ctx.device.delete_real_server(rs)112 if len(rss) > 0:113 for rs_child in rss:114 db_api.server_update(ctx.conf, rs_child['id'],115 {'parent_id': rss[-1]['id']})116 db_api.server_update(ctx.conf, rss[-1]['id'],117 {'parent_id': '', 'deployed': 'True'})118def create_sticky(ctx, sticky):119 ctx.device.create_stickiness(sticky)120 sticky['deployed'] = 'True'121 db_api.sticky_update(ctx.conf, sticky['id'], sticky)122@ignore_exceptions123def delete_sticky(ctx, sticky):124 ctx.device.delete_stickiness(sticky)125 sticky['deployed'] = 'False'126 db_api.sticky_update(ctx.conf, sticky['id'], sticky)127@ignore_exceptions128def delete_server_farm(ctx, sf):129 ctx.device.delete_server_farm(sf)130 sf['deployed'] = 'False'131 db_api.serverfarm_update(ctx.conf, sf['id'], sf)132@with_rollback133def create_server_farm(ctx, sf_ref):134 try:135 predictor_ref = db_api.predictor_get_by_sf_id(ctx.conf, sf_ref['id'])136 ctx.device.create_server_farm(sf_ref, predictor_ref)137 db_api.serverfarm_update(ctx.conf, sf_ref['id'], {'deployed': True})138 yield139 except Exception:140 with utils.save_and_reraise_exception():141 delete_server_farm(ctx, sf_ref)142@with_rollback143def add_rserver_to_server_farm(ctx, server_farm, rserver):144 try:145 if (rserver.get('parent_id') and rserver['parent_id'] != ""):146 #Nasty hack. We need to think how todo this more elegant147 rserver['name'] = rserver['parent_id']148 ctx.device.add_real_server_to_server_farm(server_farm, rserver)149 yield150 except Exception:151 with utils.save_and_reraise_exception():152 ctx.device.delete_real_server_from_server_farm(server_farm,153 rserver)154@ignore_exceptions155def delete_rserver_from_server_farm(ctx, server_farm, rserver):156 ctx.device.delete_real_server_from_server_farm(server_farm, rserver)157@ignore_exceptions158def delete_probe(ctx, probe):159 ctx.device.delete_probe(probe)160 probe['deployed'] = 'False'161 db_api.probe_update(ctx.conf, probe['id'], probe)162@with_rollback163def create_probe(ctx, probe):164 try:165 ctx.device.create_probe(probe)166 db_api.probe_update(ctx.conf, probe['id'], {'deployed': True})167 yield168 except Exception:169 with utils.save_and_reraise_exception():170 delete_probe(ctx, probe)171@with_rollback172def add_probe_to_server_farm(ctx, server_farm, probe):173 try:174 ctx.device.add_probe_to_server_farm(server_farm, probe)175 yield176 except Exception:177 with utils.save_and_reraise_exception():178 ctx.device.delete_probe_from_server_farm(server_farm, probe)179@ignore_exceptions180def remove_probe_from_server_farm(ctx, server_farm, probe):181 ctx.device.delete_probe_from_server_farm(server_farm, probe)182def activate_rserver(ctx, server_farm, rserver):183 ctx.device.activate_real_server(server_farm, rserver)184def suspend_rserver(ctx, server_farm, rserver):185 ctx.device.suspend_real_server(server_farm, rserver)186@ignore_exceptions187def delete_vip(ctx, vip):188 ctx.device.delete_virtual_ip(vip)189 vip['deployed'] = 'False'190 db_api.virtualserver_update(ctx.conf, vip['id'], vip)191@with_rollback192def create_vip(ctx, vip, server_farm):193 try:194 ctx.device.create_virtual_ip(vip, server_farm)195 db_api.virtualserver_update(ctx.conf, vip['id'], {'deployed': True})196 yield197 except Exception:198 with utils.save_and_reraise_exception():199 delete_vip(ctx, vip)200def create_loadbalancer(ctx, sf_ref, vips, servers, probes, stickies):201 # vips are mapping to "virtualIps" list on CreateLB args202 # servers are mapping to "nodes" list on CreateLB args203 # probes are mapping to "healthMoitor" list on CreateLB args204 create_server_farm(ctx, sf_ref)205 for vip_ref in vips:206 create_vip(ctx, vip_ref, sf_ref)207 for probe_ref in probes:208 add_probe_to_loadbalancer(ctx, sf_ref, probe_ref)209 for server_ref in servers:210 add_node_to_loadbalancer(ctx, sf_ref, server_ref)211def delete_loadbalancer(ctx, sf_ref, vips, servers, probes, stickies):212 for vip_ref in vips:213 delete_vip(ctx, vip_ref)214 for server_ref in servers:215 remove_node_from_loadbalancer(ctx, sf_ref, server_ref)216 for probe_ref in probes:217 remove_probe_from_loadbalancer(ctx, sf_ref, probe_ref)218 for sticky_ref in stickies:219 delete_sticky(ctx, sticky_ref)220 delete_server_farm(ctx, sf_ref)221def add_node_to_loadbalancer(ctx, sf, rserver):222 if(rserver_exist(ctx, rserver)):223 return224 else:225 create_rserver(ctx, rserver)226 add_rserver_to_server_farm(ctx, sf, rserver)227def remove_node_from_loadbalancer(ctx, sf, rserver):228 if(rserver_exist(ctx, rserver)):229 delete_rserver_from_server_farm(ctx, sf, rserver)230 delete_rserver(ctx, rserver)231 else:232 return233def add_probe_to_loadbalancer(ctx, sf_ref, probe_ref):234 create_probe(ctx, probe_ref)235 add_probe_to_server_farm(ctx, sf_ref, probe_ref)236def remove_probe_from_loadbalancer(ctx, sf_ref, probe_ref):237 remove_probe_from_server_farm(ctx, sf_ref, probe_ref)238 delete_probe(ctx, probe_ref)239def add_sticky_to_loadbalancer(ctx, balancer, sticky):240 create_sticky(ctx, sticky)241def remove_sticky_from_loadbalancer(ctx, balancer, sticky):...
dispatcher.py
Source:dispatcher.py
1"""2This file contains the OperationDispatcher class and the OperationError class3Creator: Danilo Carvalho4"""5import logging6from lbgenerator.lbrad.operations import *7class OperationDispatcher(object):8 """9 This class is responsable for executing all requested operations in the10 order in which they are requested. It also checks for errors and executes11 rollback of the operations when needed.12 """13 valid_operations={14 "db_base_create": CreateBaseOperation, 15 "db_base_read": ReadBaseOperation, 16 "db_base_delete": DeleteBaseOperation, 17 "db_doc_create": CreateDocumentOperation, 18 "db_doc_read": ReadDocumentOperation, 19 "db_doc_read_full": ReadFullDocumentOperation, 20 "db_doc_read_partial": PartialReadDocumentOperation, 21 "db_doc_list": ListDocumentOperation, 22 "db_doc_update": UpdateDocumentOperation, 23 "db_doc_update_partial": PartialUpdateDocumentOperation, 24 "db_doc_delete": DeleteDocumentOperation, 25 "db_file_create": CreateFileOperation, 26 "db_file_create_multipart": MultipartCreateFileOperation, 27 "db_file_read": ReadFileOperation, 28 "db_file_delete": DeleteFileOperation29 # TODO: Others! By John Doe30 }31 def __init__(self, request_data, request_url=None):32 """33 Params:34 - request_data (dict): must contain:35 - "operations" (list): a list of operations to be executed (see 36 operations.py)37 - "transaction" (boolean) [optional]: defines whether this request 38 is a39 transaction. A transaction will be executed "atomically" so if 40 an error occurs in one of its operations, all of them will be 41 rolled back (undone).42 - request_url (string) [optional]: the url used by the client in the 43 request44 """45 # NOTE: Logger! By Questor46 self.logger=logging.getLogger("DEBUG")47 self.logger.setLevel(logging.WARNING)48 ch=logging.StreamHandler()49 ch.setLevel(logging.WARNING)50 formatter=logging.Formatter('%(levelname)s - %(message)s')51 ch.setFormatter(formatter)52 self.logger.addHandler(ch)53 self.request_data=request_data54 self.request_url=request_url55 # NOTE: Create operation queue! By John Doe56 self.operation_queue=[]57 # NOTE: Create result list! By John Doe58 self.result_list=[]59 # NOTE: ROLLBACK - If the request is a transaction, operations will be60 # rolled back in case of failure! By John Doe61 self.transaction=True62 # NOTE: Create rollback operation "stack=[]"! By John Doe63 self.rollback_stack=[]64 def dispatch(self):65 try:66 self._parse_request()67 self._run_operations()68 except OperationError as e:69 # TODO: return error! By John Doe70 return { "error": str(e.args) }71 return self.result_list72 def _parse_request(self):73 if "transaction" in self.request_data:74 self.transaction=self.request_data["transaction"]75 op_list=self.request_data["operations"]76 for op_params in op_list:77 op_type=op_params["op_type"]78 if op_type not in OperationDispatcher.valid_operations:79 raise OperationError("Invalid operations: " + op_type)80 op_params["request_url"]=self.request_url81 op_class=OperationDispatcher.valid_operations[op_type]82 operation=op_class(op_params, self.transaction)83 self.operation_queue.append(operation)84 def _run_operations(self):85 success=True86 for op in self.operation_queue:87 result=op.run()88 self.result_list.append(result)89 if self.transaction:90 if result["success"]:91 context=op.get_context()92 if context is not None:93 self.rollback_stack.append(context)94 else:95 # TODO: Add "empty" operation to the stack! By John Doe96 pass97 else:98 # NOTE: Prevent other operations from running! By John Doe99 success=False100 break101 # NOTE: If there was an error in one of the operations of the102 # transaction! By John Doe103 if self.transaction:104 while len(self.rollback_stack) > 0:105 context=self.rollback_stack.pop()106 if success:107 # NOTE: Commit transactions! By John Doe108 context.session.commit()109 else:110 # NOTE: Rollback by executing all undo operations in the111 # stack! By John Doe112 context.session.rollback()113 i=len(self.rollback_stack)114 self.result_list[i]["undone"]=True115 context.session.close()116 # TODO: for each operation in the transaction that was not117 # executed add a "not executed" result to "self.result_list"!118 # By John Doe119 # TODO: Test! By John Doe120 def register_operation(self, key, op_class):121 """122 Registers a custom operation.123 Args:124 - key: the operation string to be used in a request. ex: 125 "bd_create_op"126 - op_class: the operation class, it needs to be a subclass of 127 LBOperation128 """129 if key in OperationDispatcher.valid_operations:130 # TODO: Raise error! By John Doe131 return132 if not isinstance(op_class, LBOperation):133 # TODO: Raise error! By John Doe134 return135 OperationDispatcher.valid_operations[key]=op_class136class OperationError(RuntimeError):137 def __init__(self, msg):138 self.msg=msg139 def __str__(self):...
configure-qtcreator.py
Source:configure-qtcreator.py
1#!/usr/bin/env python32#3# Copyright (C) 2017 Pelagicore AB4#5# This Source Code Form is subject to the terms of the Mozilla Public6# License, v. 2.0. If a copy of the MPL was not distributed with this7# file, You can obtain one at http://mozilla.org/MPL/2.0/.8#9# SPDX-License-Identifier: MPL-2.010#11"""12This script is used to configure qt-creator with a PELUX-SDK.13It is assumed that the environment-setup script for the SDK has14been sourced before this script is executed.15"""16import os17import shutil18import time19import sys20class CommandException(Exception):21 pass22class QtCreatorBootstrapper(object):23 def __init__(self, sdktool_path):24 self.rollback_stack = []25 epoch_time = int(time.time())26 self.name = "PELUX-SDK-{}".format(epoch_time)27 self.cxx_toolchain_id = "ProjectExplorer.ToolChain.Gcc:" + self.name + "CXX"28 self.cc_toolchain_id = "ProjectExplorer.ToolChain.Gcc:" + self.name + "CC"29 self.device_id = self.name + "_device"30 self.cmake_id = self.name + "_cmake"31 self.qtversion_id = self.name + "_qt"32 self.kit_id = self.name + "_kit"33 arch_env_var = os.environ["ARCH"]34 if "arm" in arch_env_var:35 self.architecture = "arm"36 self.bits = "64bit" if "64" in arch_env_var else "32bit"37 elif "x86" in arch_env_var:38 self.architecture = "x86"39 self.bits = "64bit" if "64" in arch_env_var else "32bit"40 else:41 self.architecture = "unknown"42 self.bits = "unknown"43 self.sdktool = sdktool_path44 assert shutil.which(self.sdktool) is not None45 def __exec_cmd(self, cmd_to_exec):46 print("Executing command: {}".format(cmd_to_exec))47 complete_command = "{} {}".format(self.sdktool, cmd_to_exec)48 if os.system(complete_command) != 0:49 raise CommandException("Failed to execute {}".format(complete_command))50 def __add_rollback_cmd(self, rollback_cmd):51 self.rollback_stack.append(rollback_cmd)52 def add_cxx_toolchain(self):53 self.__add_toolchain(id=self.cxx_toolchain_id, compiler_env_var="CXX", lang="Cxx")54 def add_cc_toolchain(self):55 self.__add_toolchain(id=self.cc_toolchain_id, compiler_env_var="CC", lang="C")56 def __add_toolchain(self, id, compiler_env_var, lang):57 abi = "{}-linux-generic-elf-{}".format(self.architecture, self.bits)58 compiler_binary_name = os.environ[(str(compiler_env_var))].split()[0]59 compiler_path = shutil.which(compiler_binary_name)60 self.__exec_cmd("addTC"61 + " --name " + self.name + "_" + compiler_env_var62 + " --language " + lang63 + " --id " + id64 + " --abi " + abi65 + " --path " + compiler_path66 )67 self.__add_rollback_cmd("rmTC --id " + id)68 def add_cmake(self):69 cmake_path = shutil.which("cmake")70 self.__exec_cmd("addCMake"71 + " --name " + self.name72 + " --id " + self.cmake_id73 + " --path " + cmake_path74 )75 self.__add_rollback_cmd("rmCMake --id " + self.cmake_id)76 def add_qt_version(self):77 qmake = shutil.which("qmake")78 self.__exec_cmd("addQt"79 + " --name " + self.name80 + " --id " + self.qtversion_id81 + " --type RemoteLinux.EmbeddedLinuxQt"82 + " --qmake " + qmake83 )84 self.__add_rollback_cmd("rmQt --id " + self.qtversion_id)85 def add_kit(self, use_qt=False):86 qt_params = ""87 if use_qt:88 qt_params += " --qt " + self.qtversion_id + " --mkspec \"" + os.environ["QMAKESPEC"] + "\""89 self.__exec_cmd("addKit"90 + " --name " + self.name91 + " --id " + self.kit_id92 + " --debuggerengine 1"93 + " --debugger " + shutil.which(os.environ["GDB"])94 + " --devicetype GenericLinuxOsType"95 + " --sysroot " + os.environ["SDKTARGETSYSROOT"]96 + " --Cxxtoolchain " + self.cxx_toolchain_id97 + " --Ctoolchain " + self.cc_toolchain_id98 + " --cmake " + self.cmake_id99 + qt_params100 + self.generate_env_arguments()101 )102 self.__add_rollback_cmd("rmKit --id " + self.kit_id)103 def generate_env_arguments(self):104 env_args = ""105 for key, value in os.environ.items():106 env_args += " --env \"{}={}\"".format(key, value.strip())107 return env_args108 def rollback(self):109 for _ in self.rollback_stack:110 rollback_cmd = self.rollback_stack.pop()111 self.__exec_cmd(rollback_cmd)112 self.rollback_stack = []113def qt_installed():114 return shutil.which("qmake") is not None115if __name__ == "__main__":116 if len(sys.argv) != 2:117 print("Need to pass path to the sdktool")118 quit()119 bootstrapper = QtCreatorBootstrapper(sys.argv[1])120 try:121 bootstrapper.add_cxx_toolchain()122 bootstrapper.add_cc_toolchain()123 bootstrapper.add_cmake()124 if qt_installed():125 bootstrapper.add_qt_version()126 bootstrapper.add_kit(qt_installed())127 except CommandException as e:128 bootstrapper.rollback()129 raise e...
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 localstack 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?
