KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use result_upload_file method in autotest

Best Python code snippet using autotest_python

bkr_proxy.py

Source:bkr_proxy.py

...234    def task_upload_file(self, task_id, localfile, remotepath=''):235        self.cmd_log.write('task_upload_file: task_id(%s) localfile=%s, remotepath=%s\n' %236                           (task_id, localfile, remotepath))237        self._upload_file(localfile, remotepath, self.recipe_id, task_id)238    def result_upload_file(self, task_id, result_id, localfile, remotepath=''):239        self.cmd_log.write('result_upload_file: task_id(%s), result_id(%s)'240                           ' localfile=%s, remotepath=%s\n' %241                           (task_id, result_id, localfile, remotepath))242        self._upload_file(localfile, remotepath, self.recipe_id, task_id, result_id)243    def get_recipe(self):244        self.cmd_log.write('get_recipe: GET %s\n' % self.recipe_id)245        path = make_path_bkrcache(self.recipe_id)246        try:247            rpath = self.labc_url + make_path_recipe(self.recipe_id)248            utils.get_file(rpath, path)249        except Exception:250            # local will fall through to here251            if not os.path.isfile(path):252                raise BkrProxyException("No remote or cached recipe %s" % self.recipe_id)...

main.py

Source:main.py

1from flask import Flask, jsonify, request, Response2import boilerplate3from hseling_api_nauchpop.process import process_topic,\4    process_rb, process_term, process_ner5ALLOWED_EXTENSIONS = ['txt']6# log = getLogger(__name__)7app = Flask(__name__)8app.config.update(9    CELERY_BROKER_URL=boilerplate.CELERY_BROKER_URL,10    CELERY_RESULT_BACKEND=boilerplate.CELERY_RESULT_BACKEND11)12celery = boilerplate.make_celery(app)13@celery.task14def task_topic(file_ids_list=None):15    data_to_process = boilerplate.get_process_data(file_ids_list)16    processed_file_ids = list()17    for processed_file_id, contents in process_topic(data_to_process):18        processed_file_ids.append(19            boilerplate.add_processed_file(20                boilerplate.TOPIC_PREFIX,21                processed_file_id,22                contents,23                extension='txt'24            ))25    return processed_file_ids26@celery.task27def task_rb(file_ids_list=None):28    data_to_process = boilerplate.get_process_data(file_ids_list)29    processed_file_ids = list()30    for processed_file_id, contents in process_rb(data_to_process):31        processed_file_ids.append(32            boilerplate.add_processed_file(33                boilerplate.RB_PREFIX,34                processed_file_id,35                contents,36                extension='txt'37            ))38    return processed_file_ids39@celery.task40def task_term(file_ids_list=None):41    data_to_process = boilerplate.get_process_data(file_ids_list)42    processed_file_ids = list()43    for processed_file_id, contents in process_term(data_to_process):44        processed_file_ids.append(45            boilerplate.add_processed_file(46                boilerplate.TERM_PREFIX,47                processed_file_id,48                contents,49                extension='txt'50            ))51    return processed_file_ids52@celery.task53def task_ner(file_ids_list=None):54    data_to_process = boilerplate.get_process_data(file_ids_list)55    processed_file_ids = list()56    for processed_file_id, contents in process_ner(data_to_process):57        processed_file_ids.append(58            boilerplate.add_processed_file(59                boilerplate.NER_PREFIX,60                processed_file_id,61                contents,62                extension='txt'63            ))64    return processed_file_ids65@app.route('/upload', methods=['GET', 'POST'])66def upload_endpoint():67    if request.method == 'POST':68        uploaded_files = request.files.getlist("file[]")69        # upload_file = request.files['file']70        # make upload for few files71        result_upload_file = {}72        for num, upload_file in enumerate(uploaded_files):73            if uploaded_files[num].filename == '':74                return jsonify({"error": boilerplate.ERROR_NO_SELECTED_FILE})75            if uploaded_files[num] and boilerplate.allowed_file(76                    uploaded_files[num].filename,77                    allowed_extensions=ALLOWED_EXTENSIONS):78                result_done = boilerplate.save_file(uploaded_files[num])79                result = {str(num): result_done}80                result_upload_file.update(result)81        return jsonify(result_upload_file)82    return boilerplate.get_upload_form()83@app.route('/files/<path:file_id>')84def get_file_endpoint(file_id):85    if file_id in boilerplate.list_files(recursive=True):86        contents = boilerplate.get_file(file_id)87        return Response(contents, mimetype='text/plain')88    return jsonify({'error': boilerplate.ERROR_NO_SUCH_FILE})89@app.route('/files')90def list_files_endpoint():91    return jsonify({'file_ids': boilerplate.list_files(recursive=True)})92@app.route('/process')93@app.route("/process/<file_ids>", methods=['GET', 'POST'])94def process_endpoint(file_ids=None):95    if request.method == 'POST':96        process_types = request.data97        process_types = process_types.decode('utf-8')98        all_process_types = process_types.split(',')99        file_ids_list = file_ids and file_ids.split(",")100        task_list = []101        for process_type in all_process_types:102            if process_type == 'topic':103                task_1 = task_topic.delay(file_ids_list)104                task_dict_1 = {"task_1_id": str(task_1)}105                task_list.append(task_dict_1)106            elif process_type == 'rb':107                task_2 = task_rb.delay(file_ids_list)108                task_dict_2 = {"task_2_id": str(task_2)}109                task_list.append(task_dict_2)110            elif process_type == 'term':111                task_3 = task_term.delay(file_ids_list)112                task_dict_3 = {"task_3_id": str(task_3)}113                task_list.append(task_dict_3)114            elif process_type == 'ner':115                task_4 = task_ner.delay(file_ids_list)116                task_dict_4 = {"task_4_id": str(task_4)}117                task_list.append(task_dict_4)118            else:119                pass120        modules_to_process = {}121        if not task_list:122            return jsonify(123                {'error': boilerplate.ERROR_NO_PROCESS_TYPE_SPECIFIED})124        elif not file_ids_list:125            return jsonify(126                {'error': boilerplate.ERROR_NO_SUCH_FILE})127        else:128            for t in task_list:129                modules_to_process.update(t)130            return jsonify(modules_to_process)131@app.route("/status/<task_id>")132def status_endpoint(task_id):133    return jsonify(boilerplate.get_task_status(task_id))134def get_endpoints(ctx):135    def endpoint(name, description, active=True):136        return {137            "name": name,138            "description": description,139            "active": active140        }141    all_endpoints = [142        endpoint("root", boilerplate.ENDPOINT_ROOT),143        endpoint("scrap", boilerplate.ENDPOINT_SCRAP,144                 not ctx["restricted_mode"]),145        endpoint("upload", boilerplate.ENDPOINT_UPLOAD),146        endpoint("process", boilerplate.ENDPOINT_PROCESS),147        endpoint("status", boilerplate.ENDPOINT_STATUS)148    ]149    return {ep["name"]: ep for ep in all_endpoints if ep}150@app.route("/")151def main_endpoint():152    ctx = {"restricted_mode": boilerplate.RESTRICTED_MODE}153    return jsonify({"endpoints": get_endpoints(ctx)})154if __name__ == "__main__":155    app.run(host='0.0.0.0', debug=True, port=80)...

beah_dummy.py

Source:beah_dummy.py

...53        result_id = self.rpc2.task_result(task_id, 'pass_', task_name, '1079',54            '(Pass)')55        #no chunking56        print '%s T:%s TR:%s result_upload_file' % (self.machine_name, task_id, result_id)57        self.rpc2.result_upload_file(result_id, '/',58            'load_test--result', len(self.recipe_log), self.recipe_log_md5, 0, self.recipe_log_enc)59        print '%s T:%s task_stop' % (self.machine_name, task_id)60        self.rpc2.task_stop(task_id, 'stop', 'OK')61class BeahDummyManager(object):62    """63    It's too expensive to create a separate beah_dummy process for potentially 64    thousands of dummy systems at the same time. So this is a tiny server 65    process which runs multiple beah_dummies in their own greenlet.66    To start a new beah_dummy:67        PUT /beah_dummy/fqdn.example.com68    """69    # TODO we could also implement DELETE to terminate a running beah_dummy70    def wsgi(self, environ, start_response):71        if not environ['PATH_INFO'].startswith('/beah_dummy/'):...

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.