Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/galaxy/tool_util/verify/test_data.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import hashlib | |
2 import os | |
3 import re | |
4 import subprocess | |
5 from string import Template | |
6 | |
7 from galaxy.util import ( | |
8 asbool, | |
9 in_directory, | |
10 smart_str | |
11 ) | |
12 | |
13 UPDATE_TEMPLATE = Template( | |
14 "git --work-tree $dir --git-dir $dir/.git fetch && " | |
15 "git --work-tree $dir --git-dir $dir/.git merge origin/master" | |
16 ) | |
17 | |
18 UPDATE_FAILED_TEMPLATE = Template( | |
19 "Warning failed to update test repository $dir - " | |
20 "update stdout was [$stdout] and stderr was [$stderr]." | |
21 ) | |
22 | |
23 | |
24 LIST_SEP = re.compile(r"\s*,\s*") | |
25 | |
26 | |
27 class TestDataResolver: | |
28 | |
29 def __init__(self, file_dirs=None, env_var='GALAXY_TEST_FILE_DIR', environ=os.environ): | |
30 if file_dirs is None: | |
31 file_dirs = environ.get(env_var, None) | |
32 if file_dirs is None: | |
33 file_dirs = "test-data,https://github.com/galaxyproject/galaxy-test-data.git" | |
34 if file_dirs: | |
35 self.resolvers = [build_resolver(u, environ) for u in LIST_SEP.split(file_dirs)] | |
36 else: | |
37 self.resolvers = [] | |
38 | |
39 def get_filename(self, name): | |
40 for resolver in self.resolvers or []: | |
41 if not resolver.exists(name): | |
42 continue | |
43 filename = resolver.path(name) | |
44 if filename: | |
45 return os.path.abspath(filename) | |
46 | |
47 def get_filecontent(self, name): | |
48 filename = self.get_filename(name=name) | |
49 with open(filename, mode='rb') as f: | |
50 return f.read() | |
51 | |
52 def get_directory(self, name): | |
53 return self.get_filename(name=name) | |
54 | |
55 | |
56 def build_resolver(uri, environ): | |
57 if uri.startswith("http") and uri.endswith(".git"): | |
58 return GitDataResolver(uri, environ) | |
59 else: | |
60 return FileDataResolver(uri) | |
61 | |
62 | |
63 class FileDataResolver: | |
64 | |
65 def __init__(self, file_dir): | |
66 self.file_dir = file_dir | |
67 | |
68 def exists(self, filename): | |
69 path = os.path.abspath(self.path(filename)) | |
70 return os.path.exists(path) and in_directory(path, self.file_dir) | |
71 | |
72 def path(self, filename): | |
73 return os.path.join(self.file_dir, filename) | |
74 | |
75 | |
76 class GitDataResolver(FileDataResolver): | |
77 | |
78 def __init__(self, repository, environ): | |
79 self.repository = repository | |
80 self.updated = False | |
81 repo_cache = environ.get("GALAXY_TEST_DATA_REPO_CACHE", "test-data-cache") | |
82 m = hashlib.md5() | |
83 m.update(smart_str(repository)) | |
84 repo_path = os.path.join(repo_cache, m.hexdigest()) | |
85 super().__init__(repo_path) | |
86 # My preference would be for this to be false, but for backward compat | |
87 # will leave it as true for now. | |
88 self.fetch_data = asbool(environ.get("GALAXY_TEST_FETCH_DATA", "true")) | |
89 | |
90 def exists(self, filename): | |
91 exists_now = super().exists(filename) | |
92 if exists_now or not self.fetch_data or self.updated: | |
93 return exists_now | |
94 self.update_repository() | |
95 return super().exists(filename) | |
96 | |
97 def update_repository(self): | |
98 self.updated = True | |
99 if not os.path.exists(self.file_dir): | |
100 parent_dir = os.path.dirname(self.file_dir) | |
101 if not os.path.exists(parent_dir): | |
102 os.makedirs(parent_dir) | |
103 self.execute(f"git clone '{self.repository}' '{self.file_dir}'") | |
104 update_command = UPDATE_TEMPLATE.safe_substitute(dir=self.file_dir) | |
105 self.execute(update_command) | |
106 | |
107 def execute(self, cmd): | |
108 subprocess_kwds = dict( | |
109 shell=True, | |
110 stdout=subprocess.PIPE, | |
111 stderr=subprocess.PIPE, | |
112 ) | |
113 print("Executing %s" % cmd) | |
114 p = subprocess.Popen(cmd, **subprocess_kwds) | |
115 stdout, stderr = p.communicate() | |
116 if p.returncode != 0: | |
117 kwds = { | |
118 'dir': self.file_dir, | |
119 'stdout': stdout, | |
120 'stderr': stderr, | |
121 } | |
122 print(UPDATE_FAILED_TEMPLATE.substitute(**kwds)) |