1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
import os
import re
import yaml
from .conda import conda, conda_env_load
from .git import git
from .utils import pushd
from glob import glob
from subprocess import run
DMFILE_RE = re.compile(r'^(?P<name>.*)[=<>~\!](?P<version>.*).*$')
def comment_find(s, delims=[';', '#']):
""" Return index of first match
"""
for delim in delims:
pos = s.find(delim)
if pos != -1:
break
return pos
def dmfile(filename):
result = []
with open(filename, 'r') as fp:
for line in fp:
line = line.strip()
comment_pos = comment_find(line)
if comment_pos >= 0:
line = line[:comment_pos]
if not line:
continue
result.append(line)
return result
def env_combine(filename, conda_env, conda_channels=[]):
packages = []
channels_result = '--override-channels '
for line in dmfile(filename):
packages.append(f"'{line}'")
for channel in conda_channels:
channels_result += f'-c {channel} '
packages_result = ' '.join([x for x in packages])
proc = conda('install', '-q', '-y', '-n',
conda_env, channels_result, packages_result)
if proc.stderr:
print(proc.stderr.decode())
def testable_packages(filename, prefix):
""" Scan a mini/anaconda prefix for unpacked packages matching versions
requested by dmfile.
"""
pkgdir = os.path.join(prefix, 'pkgs')
paths = []
for line in dmfile(filename):
match = DMFILE_RE.match(line)
if match:
pkg = match.groupdict()
# Reconstruct ${package}-${version} format from
# ${package}${specifier}${version}
pattern = f"{pkg['name']}-{pkg['version']}*"
# Record path to extracted package
path = ''.join([x for x in glob(os.path.join(pkgdir, pattern))
if os.path.isdir(x)])
paths.append(path)
for root in paths:
info_d = os.path.join(root, 'info')
recipe_d = os.path.join(info_d, 'recipe')
git_log = os.path.join(info_d, 'git')
if not os.path.exists(git_log):
continue
with open(os.path.join(recipe_d, 'meta.yaml')) as yaml_data:
source = yaml.load(yaml_data.read())['source']
if not isinstance(source, dict):
continue
repository = source['git_url']
head = open(git_log).readlines()[1].split()[1]
yield dict(repo=repository, commit=head)
def integration_test(pkg_data, conda_env, results_root='.'):
results_root = os.path.abspath(os.path.join(results_root, 'results'))
src_root = os.path.abspath('src')
if not os.path.exists(src_root):
os.mkdir(src_root, 0o755)
with pushd(src_root) as _:
repo_root = os.path.basename(pkg_data['repo']).replace('.git', '')
if not os.path.exists(repo_root):
git(f"clone --recursive {pkg_data['repo']} {repo_root}")
with pushd(repo_root) as _:
git(f"checkout {pkg_data['commit']}")
with conda_env_load(conda_env):
results = os.path.abspath(os.path.join(results_root,
repo_root,
'result.xml'))
run("pip install -e .[test]".split()).check_returncode()
run(f"pytest -v --junitxml={results}".split(), check=True)
|
