Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tool_util.verify.test_data

import hashlib
import os
import re
import subprocess
from string import Template

from galaxy.util import (
    asbool,
    in_directory,
    smart_str
)

UPDATE_TEMPLATE = Template(
    "git --work-tree $dir --git-dir $dir/.git fetch && "
    "git --work-tree $dir --git-dir $dir/.git merge origin/master"
)

UPDATE_FAILED_TEMPLATE = Template(
    "Warning failed to update test repository $dir - "
    "update stdout was [$stdout] and stderr was [$stderr]."
)


LIST_SEP = re.compile(r"\s*,\s*")


[docs]class TestDataResolver: __test__ = False # Prevent pytest from discovering this class (issue #12071)
[docs] def __init__(self, file_dirs=None, env_var='GALAXY_TEST_FILE_DIR', environ=os.environ): if file_dirs is None: file_dirs = environ.get(env_var, None) if file_dirs is None: file_dirs = "test-data,https://github.com/galaxyproject/galaxy-test-data.git" if file_dirs: self.resolvers = [build_resolver(u, environ) for u in LIST_SEP.split(file_dirs)] else: self.resolvers = []
[docs] def get_filename(self, name): for resolver in self.resolvers or []: if not resolver.exists(name): continue filename = resolver.path(name) if filename: return os.path.abspath(filename)
[docs] def get_filecontent(self, name): filename = self.get_filename(name=name) with open(filename, mode='rb') as f: return f.read()
[docs] def get_directory(self, name): return self.get_filename(name=name)
[docs]def build_resolver(uri, environ): if uri.startswith("http") and uri.endswith(".git"): return GitDataResolver(uri, environ) else: return FileDataResolver(uri)
[docs]class FileDataResolver:
[docs] def __init__(self, file_dir): self.file_dir = file_dir
[docs] def exists(self, filename): path = os.path.abspath(self.path(filename)) return os.path.exists(path) and in_directory(path, self.file_dir)
[docs] def path(self, filename): return os.path.join(self.file_dir, filename)
[docs]class GitDataResolver(FileDataResolver):
[docs] def __init__(self, repository, environ): self.repository = repository self.updated = False repo_cache = environ.get("GALAXY_TEST_DATA_REPO_CACHE", "test-data-cache") m = hashlib.md5() m.update(smart_str(repository)) repo_path = os.path.join(repo_cache, m.hexdigest()) super().__init__(repo_path) # My preference would be for this to be false, but for backward compat # will leave it as true for now. self.fetch_data = asbool(environ.get("GALAXY_TEST_FETCH_DATA", "true"))
[docs] def exists(self, filename): exists_now = super().exists(filename) if exists_now or not self.fetch_data or self.updated: return exists_now self.update_repository() return super().exists(filename)
[docs] def update_repository(self): self.updated = True if not os.path.exists(self.file_dir): parent_dir = os.path.dirname(self.file_dir) if not os.path.exists(parent_dir): os.makedirs(parent_dir) self.execute(f"git clone '{self.repository}' '{self.file_dir}'") update_command = UPDATE_TEMPLATE.safe_substitute(dir=self.file_dir) self.execute(update_command)
[docs] def execute(self, cmd): subprocess_kwds = dict( shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) print(f"Executing {cmd}") p = subprocess.Popen(cmd, **subprocess_kwds) stdout, stderr = p.communicate() if p.returncode != 0: kwds = { 'dir': self.file_dir, 'stdout': stdout, 'stderr': stderr, } print(UPDATE_FAILED_TEMPLATE.substitute(**kwds))