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

How to use _download_tests method in avocado

Best Python code snippet using avocado_python

jobs.py

Source:jobs.py

...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)157        except FileNotFoundError as ex:158            LOG_UI.error("Could not get job information: %s", ex)159            return exit_codes.AVOCADO_GENERIC_CRASH160        spawners = {'process': ProcessSpawner,161                    'podman': PodmanSpawner}162        spawner_name = config_data.get('nrunner.spawner')163        spawner = spawners.get(spawner_name)164        if spawner is None:165            msg = ("Could not find the spawner for job %s. This command is "166                   "experimental and only supported when job executed with "167                   "the Spawner architecture.")168            LOG_UI.error(msg, job_id)169            return exit_codes.AVOCADO_GENERIC_CRASH170        return self._download_tests(results_data.get('tests'),171                                    destination,172                                    job_id,173                                    spawner)174    def handle_show_command(self, config):175        """Called when 'avocado jobs show' command is executed."""176        job_id = config.get('jobs.show.job_id')177        results_dir = get_job_results_dir(job_id)178        if results_dir is None:179            LOG_UI.error("Error: Job %s not found", job_id)180            return exit_codes.AVOCADO_GENERIC_CRASH181        results_file = os.path.join(results_dir, 'results.json')182        config_file = os.path.join(results_dir, 'jobdata/args.json')183        try:184            results_data = self._get_data_from_file(results_file)...

_poolq_downloader.py

Source:_poolq_downloader.py

...48    os.system("rm -r poolq-{}/*html".format(latest))49    os.system("mv poolq-{} {}".format(latest, out_path))50    51    52def _download_tests(out_path, poolq_http, latest, tmp_out_path="poolq.test_data.tmp.zip"):53    """"""54    _download_file(url=poolq_http, filename=tmp_out_path)55    os.system("unzip -q {}".format(tmp_out_path))  # unzip56    os.system("rm -r {}".format(tmp_out_path))  # remove temp zip57    os.system("rm -r poolq-{}/*.bat".format(latest))58    os.system("rm -r poolq-{}/*.jar".format(latest))59    os.system("rm -r poolq-{}/*.sh".format(latest))60    os.system("rm -r poolq-{}/*pdf".format(latest))61    os.system("rm -r poolq-{}/*html".format(latest))62    os.system("mv poolq-{}/test-data/* {}".format(latest, out_path))63    os.system("rm -r poolq-{}".format(latest))64def _find_available_poolq_versions(poolq_path, download_path):65    """"""66    soup = _return_soup(poolq_path)67    poolq_distributions = {}68    poolq_distributions["name"] = []69    poolq_distributions["version"] = []70    poolq_distributions["path"] = []71    for line in soup.findAll("a"):72        if line.text.startswith("poolq-") and "zip" in line.text:73            poolq_distributions["name"].append(line.text)74            version = line.text.replace(".zip", "").split("-")[-1]75            poolq_distributions["version"].append(version)76            poolq_distributions["path"].append(77                download_path + "&filename=poolq-{}.zip".format(version)78            )79    return pd.DataFrame.from_dict(poolq_distributions)80class _poolq_downloader:81    """"""82    def __init__(self):83        """"""84        self.poolq_path = (85            "https://portals.broadinstitute.org/gpp/public/dir?dirpath=poolq-downloads"86        )87        self.download_path = "https://portals.broadinstitute.org/gpp/public/dir/download?dirpath=poolq-downloads&filename=poolq-{}.zip"88    def fetch_available(self, verbose=True):89        """"""90        self.version_df = _find_available_poolq_versions(91            self.poolq_path, self.download_path92        )93        self.latest = _find_latest_version(self.version_df, verbose)94    def download(self, out_path="./"):95        """"""96        poolq_http = self.download_path.format(self.latest)97        _download_poolq(out_path, poolq_http, self.latest)98        99    def download_tests(self, out_path):100        101        """ """102        103        poolq_http = self.download_path.format(self.latest)104        _download_tests(out_path, poolq_http, self.latest)105        106def _look_for_poolq(path):107    """Check if poolq exists. If not, download it."""108    109    _poolq_download_path = os.path.join(os.path.dirname(path), "_distribution/")110    get_poolq = _poolq_downloader()111    get_poolq.fetch_available()112    if os.path.exists(_poolq_download_path + "poolq-{}".format(get_poolq.latest)):113        print("\nLatest installation of poolq: poolq-{} previously installed.".format(get_poolq.latest))114    else:115        print("\nDownloading...")...

download.py

Source:download.py

...82    """83    download(DATASET_REMOTE, os.path.join(DATA_PATH, "qmugs.tar.gz"))84    with tarfile.open(os.path.join(DATA_PATH, "qmugs.tar.gz")) as handle:85        handle.extractall(DATA_PATH)86def _download_tests():87    """88    Helper function to download the tests used upon package build.89    """90    tests_tar = os.path.join(DATA_PATH, "test_data.tar.gz")91    download(TESTS_REMOTE, tests_tar)92    with tarfile.open(tests_tar) as handle:93        handle.extractall(DATA_PATH)94if __name__ == "__main__":95    parser = argparse.ArgumentParser(description="General download script")96    parser.add_argument("--training", dest="training", action="store_true", default=False)97    parser.add_argument("--tests", dest="tests", action="store_true", default=False)98    args = parser.parse_args()99    os.makedirs(DATA_PATH, exist_ok=True)100    # Trained models and utils101    LOGGER.info("Now downloading trained models and utils...")102    _download_required()103    # Training data104    if args.training:105        LOGGER.info("Now downloading training data...")106        _download_training()107    # Test files108    if args.tests:109        LOGGER.info("Downloading tests...")...

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.