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.