This document is for an old release 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.tools.verify.test_data

from __future__ import print_function

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

from galaxy.util import asbool

    "git --work-tree $dir --git-dir $dir/.git fetch && "
    "git --work-tree $dir --git-dir $dir/.git merge origin/master"

    "Warning failed to update test repository $dir - "
    "update stdout was [$stdout] and stderr was [$stderr]."

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

[docs]class TestDataResolver(object):
[docs] def __init__(self, env_var='GALAXY_TEST_FILE_DIR', environ=os.environ): file_dirs = environ.get(env_var, None) 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): if not self.resolvers: filename = None else: resolver = self.resolvers[0] filename = resolver.path(name) if not resolver.exists(filename): for resolver in self.resolvers[1:]: if resolver.exists(name): filename = resolver.path(name) else: # For backward compat. returning first path if none # exist - though I don't know if this function is ever # actually used in a context where one should return # a file even if it doesn't exist (e.g. a prefix or # or something) - I am pretty sure it is not used in # such a fashion in the context of tool tests. filename = resolver.path(name) return os.path.abspath(filename)
[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(object):
[docs] def __init__(self, file_dir): self.file_dir = file_dir
[docs] def exists(self, filename): return os.path.exists(self.path(filename))
[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(repository) repo_path = os.path.join(repo_cache, m.hexdigest()) super(GitDataResolver, self).__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(GitDataResolver, self).exists(filename) if exists_now or not self.fetch_data or self.updated: return exists_now self.update_repository() return super(GitDataResolver, self).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("git clone '%s' '%s'" % (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("Executing %s" % 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))