Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
expected = [
os.path.join(self.test_prefix, 'file1.txt'),
os.path.join(self.test_prefix, 'test_dir', 'file2.txt'),
os.path.join(self.test_prefix, 'test_dir', 'file3.txt'),
os.path.join(self.test_prefix, 'test_dir2', 'file4.dat'),
]
self.assertEqual(sorted(ft.expand_glob_paths(globs)), sorted(expected))
# passing non-glob patterns is fine too
file2 = os.path.join(self.test_prefix, 'test_dir', 'file2.txt')
self.assertEqual(ft.expand_glob_paths([file2]), [file2])
# test expanding of '~' into $HOME value
# hard overwrite $HOME in environment (used by os.path.expanduser) so we can reliably test this
new_home = os.path.join(self.test_prefix, 'home')
ft.mkdir(new_home, parents=True)
ft.write_file(os.path.join(new_home, 'test.txt'), 'test')
os.environ['HOME'] = new_home
self.assertEqual(ft.expand_glob_paths(['~/*.txt']), [os.path.join(new_home, 'test.txt')])
# check behaviour if glob that has no (file) matches is passed
glob_pat = os.path.join(self.test_prefix, 'test_*')
self.assertErrorRegex(EasyBuildError, "No files found using glob pattern", ft.expand_glob_paths, [glob_pat])
def test_expand_glob_paths(self):
"""Test expand_glob_paths function."""
for dirname in ['empty_dir', 'test_dir']:
ft.mkdir(os.path.join(self.test_prefix, dirname), parents=True)
for filename in ['file1.txt', 'test_dir/file2.txt', 'test_dir/file3.txt', 'test_dir2/file4.dat']:
ft.write_file(os.path.join(self.test_prefix, filename), 'gibberish')
globs = [os.path.join(self.test_prefix, '*.txt'), os.path.join(self.test_prefix, '*', '*')]
expected = [
os.path.join(self.test_prefix, 'file1.txt'),
os.path.join(self.test_prefix, 'test_dir', 'file2.txt'),
os.path.join(self.test_prefix, 'test_dir', 'file3.txt'),
os.path.join(self.test_prefix, 'test_dir2', 'file4.dat'),
]
self.assertEqual(sorted(ft.expand_glob_paths(globs)), sorted(expected))
# passing non-glob patterns is fine too
file2 = os.path.join(self.test_prefix, 'test_dir', 'file2.txt')
self.assertEqual(ft.expand_glob_paths([file2]), [file2])
except ImportError:
print "GC3Pie not available, skipping test"
return
# put GC3Pie config in place to use local host and fork/exec
resourcedir = os.path.join(self.test_prefix, 'gc3pie')
gc3pie_cfgfile = os.path.join(self.test_prefix, 'gc3pie_local.ini')
gc3pie_cfgtxt = GC3PIE_LOCAL_CONFIGURATION % {
'resourcedir': resourcedir,
'time': which('time'),
}
write_file(gc3pie_cfgfile, gc3pie_cfgtxt)
output_dir = os.path.join(self.test_prefix, 'subdir', 'gc3pie_output_dir')
# purposely pre-create output dir, and put a file in it (to check whether GC3Pie tries to rename the output dir)
mkdir(output_dir, parents=True)
write_file(os.path.join(output_dir, 'foo'), 'bar')
# remove write permissions on parent dir of specified output dir,
# to check that GC3Pie does not try to rename the (already existing) output directory...
adjust_permissions(os.path.dirname(output_dir), stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH,
add=False, recursive=False)
topdir = os.path.dirname(os.path.abspath(__file__))
build_options = {
'job_backend_config': gc3pie_cfgfile,
'job_max_walltime': 24,
'job_output_dir': output_dir,
'job_polling_interval': 0.2, # quick polling
'job_target_resource': 'ebtestlocalhost',
'robot_path': os.path.join(topdir, 'easyconfigs', 'test_ecs'),
'silent': True,
def setup_categorized_hmns_modules(self):
"""Setup categorized hierarchical modules to run tests on."""
mod_prefix = os.path.join(self.test_installpath, 'modules', 'all')
# simply copy module files under 'CategorizedHMNS/{Core,Compiler,MPI}' to test install path
# EasyBuild is responsible for making sure that the toolchain can be loaded using the short module name
mkdir(mod_prefix, parents=True)
for mod_subdir in ['Core', 'Compiler', 'MPI']:
src_mod_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
'modules', 'CategorizedHMNS', mod_subdir)
copy_dir(src_mod_path, os.path.join(mod_prefix, mod_subdir))
# create empty module file directory to make C/Tcl modules happy
mpi_pref = os.path.join(mod_prefix, 'MPI', 'GCC', '6.4.0-2.28', 'OpenMPI', '2.1.2')
mkdir(os.path.join(mpi_pref, 'base'))
# make sure only modules in the CategorizedHMNS are available
self.reset_modulepath([os.path.join(mod_prefix, 'Core', 'compiler'),
os.path.join(mod_prefix, 'Core', 'toolchain')])
# tweak use statements in modules to ensure correct paths
for modfile in [
os.path.join(mod_prefix, 'Core', 'compiler', 'GCC', '6.4.0-2.28'),
os.path.join(mod_prefix, 'Compiler', 'GCC', '6.4.0-2.28', 'mpi', 'OpenMPI', '2.1.2'),
]:
for line in fileinput.input(modfile, inplace=1):
line = re.sub(r"(module\s*use\s*)/tmp/modules/all",
r"\1%s/modules/all" % self.test_installpath,
line)
sys.stdout.write(line)
:param path: location where Git repository should be initialized
:param repo_name: name of Git repository
:param silent: keep quiet (don't print any messages)
"""
repo_path = os.path.join(path, repo_name)
# copy or init git working directory
git_working_dirs_path = build_option('git_working_dirs_path')
if git_working_dirs_path:
workdir = os.path.join(git_working_dirs_path, repo_name)
if os.path.exists(workdir):
print_msg("copying %s..." % workdir, silent=silent)
copy_dir(workdir, repo_path)
if not os.path.exists(repo_path):
mkdir(repo_path, parents=True)
try:
repo = git.Repo.init(repo_path)
except GitCommandError as err:
raise EasyBuildError("Failed to init git repo at %s: %s", repo_path, err)
_log.debug("temporary git working directory ready at %s", repo_path)
return repo
raise EasyBuildError("Failed to determine path to '%s'", tool)
# -fPIE/-pie and -fPIC are not compatible, so patch out hardcoded occurences of -fPIE/-pie if -fPIC is used
if self.toolchain.options.get('pic', None):
regex_subs.extend([('-fPIE', '-fPIC'), ('"-pie"', '"-fPIC"')])
# patch all CROSSTOOL* scripts to fix hardcoding of locations of binutils/GCC binaries
for path, dirnames, filenames in os.walk(os.getcwd()):
for filename in filenames:
if filename.startswith('CROSSTOOL'):
full_path = os.path.join(path, filename)
self.log.info("Patching %s", full_path)
apply_regex_substitutions(full_path, regex_subs)
tmpdir = '/tmp/easybuild/artificially/extended/path/to/workaround/certain/bazel/bug/'
mkdir(tmpdir, parents=True)
tmpdir = tempfile.mkdtemp(dir=tmpdir)
user_root_tmpdir = tempfile.mkdtemp(suffix='-user_root')
# compose "bazel build" command with all its options...
cmd = [self.cfg['prebuildopts'], 'bazel', '--output_base=%s' % tmpdir,
'--install_base=%s' % os.path.join(tmpdir, 'inst_base'),
'--output_user_root=%s' % user_root_tmpdir, 'build']
# build with optimization enabled
# cfr. https://docs.bazel.build/versions/master/user-manual.html#flag--compilation_mode
cmd.append('--compilation_mode=opt')
# select 'opt' config section (this is *not* the same as --compilation_mode=opt!)
# https://docs.bazel.build/versions/master/user-manual.html#flag--config
cmd.append('--config=opt')
def fetch_easyconfigs_from_pr(pr, path=None, github_user=None):
"""Fetch patched easyconfig files for a particular PR."""
if github_user is None:
github_user = build_option('github_user')
if path is None:
path = build_option('pr_path')
if path is None:
path = tempfile.mkdtemp()
else:
# make sure path exists, create it if necessary
mkdir(path, parents=True)
github_account = build_option('pr_target_account')
github_repo = GITHUB_EASYCONFIGS_REPO
_log.debug("Fetching easyconfigs from %s/%s PR #%s into %s", github_account, github_repo, pr, path)
pr_data, _ = fetch_pr_data(pr, github_account, github_repo, github_user)
pr_merged = pr_data['merged']
pr_closed = pr_data['state'] == GITHUB_STATE_CLOSED and not pr_merged
pr_target_branch = pr_data['base']['ref']
_log.info("Target branch for PR #%s: %s", pr, pr_target_branch)
# download target branch of PR so we can try and apply the PR patch on top of it
repo_target_branch = download_repo(repo=github_repo, account=github_account, branch=pr_target_branch,
github_user=github_user)
timestamp = time.strftime("%Y%m%d-%H%M%S")
backupdir = "%s.%s" % (dir_name, timestamp)
shutil.move(dir_name, backupdir)
self.log.info("Moved old directory %s to %s" % (dir_name, backupdir))
except OSError, err:
raise EasyBuildError("Moving old directory to backup %s %s failed: %s", dir_name, backupdir, err)
if dontcreateinstalldir:
olddir = dir_name
dir_name = os.path.dirname(dir_name)
self.log.info("Cleaning only, no actual creation of %s, only verification/defining of dirname %s" % (olddir, dir_name))
if os.path.exists(dir_name):
return
# if not, create dir as usual
mkdir(dir_name, parents=True)
def install_step(self):
"""Install Python package to a custom path using setup.py"""
abs_pylibdir = os.path.join(self.installdir, self.pylibdir)
mkdir(abs_pylibdir, parents=True)
pythonpath = os.getenv('PYTHONPATH')
env.setvar('PYTHONPATH', "%s:%s" % (abs_pylibdir, pythonpath))
cmd = "python setup.py install --prefix=%s %s" % (self.installdir, self.cfg['installopts'])
run_cmd(cmd, log_all=True, simple=True)
env.setvar('PYTHONPATH', pythonpath)
proc = subprocess_popen_text(cmd, env=os.environ)
(stdout, stderr) = proc.communicate()
if stderr:
raise EasyBuildError("An error occurred when running '%s': %s", ' '.join(cmd), stderr)
if self.testing:
# don't actually update local cache when testing, just return the cache contents
return stdout
else:
try:
cache_fp = os.path.join(self.USER_CACHE_DIR, 'moduleT.lua')
self.log.debug("Updating Lmod spider cache %s with output from '%s'" % (cache_fp, ' '.join(cmd)))
cache_dir = os.path.dirname(cache_fp)
if not os.path.exists(cache_dir):
mkdir(cache_dir, parents=True)
cache_file = open(cache_fp, 'w')
cache_file.write(stdout)
cache_file.close()
except (IOError, OSError) as err:
raise EasyBuildError("Failed to update Lmod spider cache %s: %s", cache_fp, err)