Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use describe_activity method in localstack
Best Python code snippet using localstack_python
step_functions_resource.py
Source:step_functions_resource.py 
...111 if definition_meta.get('Activity'):112 activity_name = definition_meta['Activity']113 activity_arn = 'arn:aws:states:{0}:{1}:activity:{2}'.format(114 self.region, self.account_id, activity_name)115 activity_info = self.sf_conn.describe_activity(116 arn=activity_arn)117 if not activity_info:118 raise AssertionError('Activity does not exists: %s',119 activity_name)120 activity_arn = activity_info['activityArn']121 del definition_copy['States'][key]['Activity']122 definition_copy['States'][key]['Resource'] = activity_arn123 machine_info = self.sf_conn.create_state_machine(machine_name=name,124 role_arn=role_arn,125 definition=definition_copy)126 event_sources = meta.get('event_sources')127 if event_sources:128 for trigger_meta in event_sources:129 trigger_type = trigger_meta['resource_type']130 func = self.CREATE_TRIGGER[trigger_type]131 func(name, trigger_meta)132 _LOG.info('Created state machine %s.', machine_info['stateMachineArn'])133 return self.describe_step_function(name=name, meta=meta, arn=arn)134 def describe_step_function(self, name, meta, arn=None):135 if not arn:136 arn = self._build_sm_arn(name, self.region)137 response = self.sf_conn.describe_state_machine(arn)138 return {139 arn: build_description_obj(response, name, meta)140 }141 @staticmethod142 def _build_sm_arn(self, name, region):143 return f'arn:aws:states:{region}:{self.account_id}:stateMachine:{name}'144 def _create_cloud_watch_trigger_from_meta(self, name, trigger_meta):145 required_parameters = ['target_rule', 'input', 'iam_role']146 validate_params(name, trigger_meta, required_parameters)147 rule_name = trigger_meta['target_rule']148 input = trigger_meta['input']149 sf_role = trigger_meta['iam_role']150 sf_arn = self._build_sm_arn(name, self.region)151 sf_description = self.sf_conn.describe_state_machine(arn=sf_arn)152 if sf_description.get('status') == 'ACTIVE':153 sf_role_arn = self.iam_conn.check_if_role_exists(sf_role)154 if sf_role_arn:155 self.cw_events_conn.add_rule_sf_target(rule_name, sf_arn,156 input,157 sf_role_arn)158 _LOG.info('State machine %s subscribed to cloudwatch rule %s',159 name, rule_name)160 CREATE_TRIGGER = {161 'cloudwatch_rule_trigger': _create_cloud_watch_trigger_from_meta162 }163 @unpack_kwargs164 def _create_activity_from_meta(self, name, meta):165 arn = self.build_activity_arn(name=name)166 response = self.sf_conn.describe_activity(arn)167 if response:168 _LOG.warn('Activity %s exists.', name)169 return {170 arn: build_description_obj(response, name, meta)171 }172 response = self.sf_conn.create_activity(name=name)173 _LOG.info('Activity %s is created.', name)174 return {175 arn: build_description_obj(response, name, meta)176 }177 def describe_activity(self, name, meta):178 arn = self.build_activity_arn(name=name)179 response = self.sf_conn.describe_activity(arn=arn)180 return {181 arn: build_description_obj(response, name, meta)182 }183 def build_activity_arn(self, name):184 arn = 'arn:aws:states:{0}:{1}:activity:{2}'.format(self.region,185 self.account_id,186 name)...
app.py
Source:app.py
1from flask import Flask,render_template,request,session,logging,url_for,redirect,flash2from flask_login import LoginManager, login_user, logout_user, login_required3from flask_wtf import CSRFProtect4from functools import wraps5from flask_mysqldb import MySQL6import MySQLdb.cursors7import re8#comentarios9from config import config10# Models:11from models.ModelUser import ModelUser12# Entities:13from models.entities.User import User14app = Flask(__name__)15#Mysql Connection16db = MySQL(app)17#Variable to login manager18login_manager_app = LoginManager(app)19#Variable for CSRF token20csrf = CSRFProtect()21@login_manager_app.user_loader22def load_user(id):23 return ModelUser.get_by_id(db, id) 24#Register api25@app.route("/register", methods=["GET","POST"])26def register():27 # Check if POST requests exist (user submitted form)28 if request.method == "POST":29 # Create variables for easy access30 user = User(0,request.form["username"],request.form["password"],fullname = request.form["fullname"],email = request.form["email"],usertype = request.form["usertype"])31 username = request.form["username"]32 fullname = request.form["fullname"]33 password = request.form["password"]34 confirm_password = request.form["confirm_password"]35 ##secure_password = sha256_crypt.encrypt(str(password))36 email = request.form["email"]37 usertype = request.form["usertype"]38 # Check if account exists using MySQL39 cursor = db.connection.cursor(MySQLdb.cursors.DictCursor)40 cursor.execute('SELECT * FROM users WHERE username = %s', (username,))41 account = cursor.fetchone()42 # If account exists show error and validation checks43 if account:44 flash("Account already exists!", "bg-red-100 text-center")45 # If is invalid email address46 elif not re.match(r'[^@]+@[^@]+\.[^@]+', email):47 flash("Invalid email address!" , "bg-red-100 text-center")48 # If username contain especial caracters49 elif not re.match(r'[A-Za-z0-9]+', username):50 flash("Username must contain only characters and numbers!", "bg-red-100")51 # If password is not the same52 elif password != confirm_password:53 flash("Password does not match", "bg-red-100")54 else:55 # Account doesnt exists and the form data is valid, now insert new account into accounts table56 cur = db.connection.cursor()57 cur.execute('INSERT INTO users(username,fullname,password,email,usertype) VALUES (%s,%s,%s,%s,%s)',(username, fullname, password, email, usertype))58 db.connection.commit()59 flash("You have successfully registered!", "bg-green-100")60 return render_template("login.html")61 return render_template("register.html")62#Login api63@app.route("/")64@app.route("/login", methods=["GET", "POST"])65def login():66 # Check if "username" and "password" POST requests exist (user submitted form)67 if request.method == "POST":68 user = User(0,request.form['username'], request.form['password'])69 logged_user = ModelUser.login(db, user)70 # Account doesnt exist or username/password incorrect71 if logged_user != None:72 if logged_user.password:73 login_user(logged_user)74 return redirect(url_for("home"))75 else:76 flash("Invalid password!", "bgr-red-100")77 else:78 flash("User not found!", "bgr-red-100")79 # Show the login form with message (if any)80 return render_template("login.html")81#logout api82@app.route("/logout")83def logout():84 # Remove session data, this will log the user out85 logout_user()86 # Redirect to login page87 return redirect(url_for("login"))88#home api89@app.route("/login/home")90@login_required91def home():92 return render_template("home.html", user=User)93 94#new activity api95@app.route("/login/newactivity", methods=["GET", "POST"])96@login_required97def newactivity():98 if request.method == "POST":99 # Create variables for easy access100 useract=session["username"]101 start_date = request.form["start_date"]102 finish_date = request.form["finish_date"]103 activity = request.form["activity"]104 describe_activity = request.form["describe_activity"]105 #Insert new activity into activites table106 cur = db.connection.cursor()107 cur.execute('INSERT INTO activities(useract,start_date,finish_date,activity,describe_activity) VALUES (%s,%s,%s,%s,%s)',(useract,start_date,finish_date,activity,describe_activity))108 db.connection.commit()109 return flash("You have successfully save the activity!", "bg-green-100") 110 return render_template("newactivity.html") 111 112#consult activity apli113@app.route("/login/consultactivity", methods=["GET", "POST"])114@login_required115def consultact():116 if request.method == "POST":117 cur= db.connection.cursor()118 cur.execute('SELECT * FROM activities WHERE username = %s, status = NULL ')119 data = cur.fetch()120 activitys = data121 return activitys122 return render_template("consultactivity.html",)123#update activity api124@app.route('/edit/<id>', methods=['POST'])125def update_act(id):126 if request.method == 'POST':127 status = request.form['status']128 cur = db.connection.cursor()129 cur.execute(""" UPDATE activities SET status = %s """, (status))130 flash('Activity Updated Successfully')131 db.connection.commit()132 return redirect(url_for('consultactivity.html'))133#delete activity api134@app.route('/delete/<string:id>', methods = ['POST','GET'])135def delete_act(id):136 cur = db.connection.cursor()137 cur.execute('DELETE FROM activities WHERE id = {0}'.format(id))138 db.connection.commit()139 flash('Activity Removed Successfully')140 return redirect(url_for('consultactivity.html')) 141@app.route("/login/capturefeedback", methods=["GET", "POST"])142@login_required143def consultfeed():144 if request.method == "POST":145 cur= db.connection.cursor()146 cur.execute('SELECT * FROM activities WHERE username = %s, status = OK LIMIT 1 ')147 data = cur.fetch()148 activitys = data149 return activitys150 return render_template("capturefeedback.html",) 151def status_401(error):152 return redirect(url_for('login'))153def status_404(error):154 return "<h1> Pagina no encontrada</h1>", 404155if __name__=='__main__':156 app.config.from_object(config['develoment'])157 csrf.init_app(app)158 app.register_error_handler(401, status_401)159 app.register_error_handler(404, status_404)...
report.py
Source:report.py
...57 activities.sort(key=lambda row: row[1], reverse=False)58 elif sorting_method == "ld": # time descending59 activities.sort(key=lambda row: row[1], reverse=True)60 # Prepare functions to describe activity61 def describe_activity(activity):62 return activity.rjust(activity_max_len, " ")63 # Prepare functions to describe length64 if timedelta_format == "str":65 describe_length = str66 elif timedelta_format == "hm":67 def describe_length(td):68 hours, minutes = extract_hours_minutes(td)69 return "{:6.0f} hours {:2.0f} minutes".format(hours, minutes)70 else:71 timedelta_value_extractors = {72 "d": extract_total_days,73 "h": extract_total_hours,74 "m": extract_total_minutes75 }76 def describe_length(td):77 return "{:6.2f}".format(78 timedelta_value_extractors[timedelta_format](td)79 )80 # Print activities and lengths81 for activity, length in activities:...
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?
