Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _save_stream_to_file method in avocado

Best Python code snippet using avocado_python

jobs.py

Source:jobs.py

...59        for line in astring.iter_tabular_output(test_matrix,60                                                header=header,61                                                strip=True):62            LOG_UI.debug(line)63    def _save_stream_to_file(self, stream, filename):64        """Save stream to a file.65        Directory must exists before calling this function.66        """67        dirname = os.path.dirname(filename)68        os.makedirs(dirname, exist_ok=True)69        with open(filename, 'ab') as output_file:70            output_file.write(stream)71    def configure(self, parser):72        """73        Add the subparser for the assets action.74        :param parser: The Avocado command line application parser75        :type parser: :class:`avocado.core.parser.ArgumentParser`76        """77        parser = super(Jobs, self).configure(parser)78        subcommands = parser.add_subparsers(dest='jobs_subcommand',79                                            metavar='sub-command')80        subcommands.required = True81        help_msg = 'List all known jobs by Avocado'82        subcommands.add_parser('list', help=help_msg)83        help_msg = ('Show details about a specific job. When passing a Job '84                    'ID, you can use any Job Reference (job_id, "latest", '85                    'or job results path).')86        show_parser = subcommands.add_parser('show', help=help_msg)87        settings.register_option(section='jobs.show',88                                 key='job_id',89                                 help_msg='JOB id',90                                 metavar='JOBID',91                                 default='latest',92                                 nargs='?',93                                 positional_arg=True,94                                 parser=show_parser)95        help_msg = ('Download output files generated by tests on '96                    'AVOCADO_TEST_OUTPUT_DIR')97        output_files_parser = subcommands.add_parser('get-output-files',98                                                     help=help_msg)99        settings.register_option(section='jobs.get.output_files',100                                 key='job_id',101                                 help_msg='JOB id',102                                 metavar='JOBID',103                                 default=None,104                                 positional_arg=True,105                                 parser=output_files_parser)106        settings.register_option(section='jobs.get.output_files',107                                 key='destination',108                                 help_msg='Destination path',109                                 metavar='DESTINATION',110                                 default=None,111                                 positional_arg=True,112                                 parser=output_files_parser)113    def handle_list_command(self, jobs_results):114        """Called when 'avocado jobs list' command is executed."""115        for filename in jobs_results.values():116            with open(filename, 'r') as fp:117                job = json.load(fp)118                try:119                    started_ts = job['tests'][0]['start']120                    started = datetime.fromtimestamp(started_ts)121                except IndexError:122                    continue123                LOG_UI.info("%-40s %-26s %3s (%s/%s/%s/%s)",124                            job['job_id'],125                            str(started),126                            job['total'],127                            job['pass'],128                            job['skip'],129                            job['errors'],130                            job['failures'])131        return exit_codes.AVOCADO_ALL_OK132    def _download_tests(self, tests, destination, job_id, spawner):133        for test in tests:134            test_id = test.get('id')135            LOG_UI.info("Downloading files for test %s", test_id)136            try:137                files_buffers = spawner().stream_output(job_id, test_id)138                for filename, stream in files_buffers:139                    dest = os.path.join(destination,140                                        test_id.replace('/', '_'),141                                        filename)142                    self._save_stream_to_file(stream, dest)143            except SpawnerException as ex:144                LOG_UI.error("Error: Failed to download: %s. Exiting...", ex)145                return exit_codes.AVOCADO_GENERIC_CRASH146        return exit_codes.AVOCADO_ALL_OK147    def handle_output_files_command(self, config):148        """Called when 'avocado jobs get-output-files' command is executed."""149        job_id = config.get('jobs.get.output_files.job_id')150        destination = config.get('jobs.get.output_files.destination')151        results_dir = get_job_results_dir(job_id)152        results_file = os.path.join(results_dir, 'results.json')153        config_file = os.path.join(results_dir, 'jobdata/args.json')154        try:155            config_data = self._get_data_from_file(config_file)156            results_data = self._get_data_from_file(results_file)...

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.