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

How to use upload_result_files method in autotest

Best Python code snippet using autotest_python

k8s.py

Source:k8s.py

...88                )89        except (AttributeError, StopIteration):90            pass91        return termination_state, log92    def upload_result_files(self):93        """Upload any new files to the output dataset."""94        new_checksum_files: List[ChecksumFile] = []95        existing_filenames: Set[str] = {96            file.name for file in self.algorithm_task.input_dataset.files.all()97        }98        for path, _, files in os.walk(self.output_dir):99            fixed_filenames = {os.path.join(path, file) for file in files}100            new_filenames = fixed_filenames - existing_filenames101            for f in new_filenames:102                relative_filename = Path(f).relative_to(self.output_dir)103                checksum_file = ChecksumFile(name=relative_filename)104                with open(f, 'rb') as file_contents:105                    checksum_file.file = SimpleUploadedFile(106                        name=str(relative_filename), content=file_contents.read()107                    )108                checksum_file.save()109                new_checksum_files.append(checksum_file)110                existing_filenames.add(f)111        self.algorithm_task.output_dataset.files.add(*new_checksum_files)112    def container_result(self) -> FinalStateAndLog:113        log: str = ''114        termination_state = None115        while termination_state is None:116            termination_state, log = self.poll_container()117            if log:118                self.algorithm_task.output_log = log119                self.algorithm_task.save()120            # Wait at least 1 second121            time.sleep(1)122        return termination_state, log123    def monitor_container(self):124        termination_state, log = self.container_result()125        # Update task126        self.algorithm_task.output_log = log127        self.algorithm_task.status = (128            AlgorithmTask.Status.SUCCEEDED129            if termination_state.exit_code == 0130            else AlgorithmTask.Status.FAILED131        )132        self.algorithm_task.output_dataset = Dataset.objects.create(133            name=f'Algorithm {self.algorithm.pk}, Task {self.algorithm_task.pk} (Output)'134        )135        self.algorithm_task.save()136        # Upload results...

task_result.py

Source:task_result.py

...58        print('upload_log_files failed: %s.' % task_file_attribute_name)59        return False60    # success61    return True62def upload_result_files(task):63    if not upload_log_files(task):64        return False65    task_file_attribute_name = 'task_scan_result_tar'66    task_file_name = taskfile.upload_task_file(result_bucket_name, task, task_file_attribute_name)67    if task_file_name == '':68        print('upload_result_files failed: %s.' % task_file_attribute_name)69        return False70    # success71    return True72def main():73    print('\nStarting task_result.py ...')74    success = get_env_vars()75    if not success:76        print('get_env_vars failed.  Exit.')77        return78    print('Env vars:')79    print('cache_bucket_name: %s' % cache_bucket_name)80    print('log_bucket_name: %s' % log_bucket_name)81    print('result_bucket_name: %s' % result_bucket_name)82    print('update_task_queue_name: %s' % update_task_queue_name)83    task = taskjson.read_task_json()84    if task is None:85        print("read_task_json failed.  Exit.")86        return87    print("read_task_json completed.")88    print("task:")89    print(task)90    if task["task_status"] != "scan-failed":91        success = upload_cache_files(task)92        if not success:93            print('upload_cache_files failed: task=%s.' % task)94        # upload_result_files method will also upload log files95        success = upload_result_files(task)96        if not success:97            print('upload_result_files failed: task=%s.' % task)98    else:99        success = upload_log_files(task)100        if not success:101            print('upload_log_files failed: task=%s.' % task)102    action = 'update'103    success = taskmessage.send_task_message(update_task_queue_name, action, task)104    if not success:105        print('send_task_message failed.  Exit.')106        return107if __name__ == '__main__':...

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.