Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_merge_split():
path_1, path_2 = '/pex/path/1:/pex/path/2', '/pex/path/3:/pex/path/4'
result = PexInfo._merge_split(path_1, path_2)
assert result == ['/pex/path/1', '/pex/path/2', '/pex/path/3', '/pex/path/4']
path_1, path_2 = '/pex/path/1:', '/pex/path/3:/pex/path/4'
result = PexInfo._merge_split(path_1, path_2)
assert result == ['/pex/path/1', '/pex/path/3', '/pex/path/4']
path_1, path_2 = '/pex/path/1::/pex/path/2', '/pex/path/3:/pex/path/4'
result = PexInfo._merge_split(path_1, path_2)
assert result == ['/pex/path/1', '/pex/path/2', '/pex/path/3', '/pex/path/4']
path_1, path_2 = '/pex/path/1::/pex/path/2', '/pex/path/3:/pex/path/4'
result = PexInfo._merge_split(path_1, None)
assert result == ['/pex/path/1', '/pex/path/2']
result = PexInfo._merge_split(None, path_2)
assert result == ['/pex/path/3', '/pex/path/4']
"""))
colorized_isort_pex = os.path.join(tmpdir, 'colorized_isort.pex')
args = [
'--sources-directory', project_code_dir,
'--entry-point', 'colorized_isort:run',
'--output-file', colorized_isort_pex
]
result = run_pex_command(args + resolver_args + requirements)
result.assert_success()
# 4. Now the tool builds a "dehydrated" PEX using the standard pex + resolve settings as the
# template.
ptex_cache = os.path.join(tmpdir, '.ptex')
colorized_isort_pex_info = PexInfo.from_pex(colorized_isort_pex)
colorized_isort_pex_info.pex_root = ptex_cache
# Force the standard pex to extract its code. An external tool like Pants would already know the
# orignal source code file paths, but we need to discover here.
colorized_isort_pex_code_dir = os.path.join(
colorized_isort_pex_info.zip_unsafe_cache,
colorized_isort_pex_info.code_hash
)
env = os.environ.copy()
env.update(PEX_ROOT=ptex_cache, PEX_INTERPRETER='1', PEX_FORCE_LOCAL='1')
subprocess.check_call([colorized_isort_pex, '-c', ''], env=env)
colorized_isort_ptex_code_dir = os.path.join(tmpdir, 'colorized_isort_ptex_code_dir')
safe_mkdir(colorized_isort_ptex_code_dir)
code = []
def test_force_local():
with nested(yield_pex_builder(), temporary_dir(), temporary_filename()) as (
pb, pex_root, pex_file):
pb.info.pex_root = pex_root
pb.build(pex_file)
code_cache = PEXEnvironment.force_local(pex_file, pb.info)
assert os.path.exists(pb.info.zip_unsafe_cache)
assert len(os.listdir(pb.info.zip_unsafe_cache)) == 1
assert [os.path.basename(code_cache)] == os.listdir(pb.info.zip_unsafe_cache)
assert set(os.listdir(code_cache)) == set([PexInfo.PATH, '__main__.py', '__main__.pyc'])
# idempotence
assert PEXEnvironment.force_local(pex_file, pb.info) == code_cache
def _from_pex(arg):
if arg == f'{home_path}/myapp.pex':
return PexInfo({"code_hash": 1})
else:
return PexInfo({"code_hash": 2})
config=pants_ini_config
)
self.assert_success(pants_run)
self.assertIn('python3', pants_run.stdout_data)
# Binary task.
pants_run = self.run_pants(
command=['binary', '{}:test_bin'.format(os.path.join(self.testproject,
'test_target_with_no_sources'))],
config=pants_ini_config
)
self.assert_success(pants_run)
# Ensure proper interpreter constraints were passed to built pexes.
py2_pex = os.path.join(os.getcwd(), 'dist', 'test_bin.pex')
py2_info = PexInfo.from_pex(py2_pex)
self.assertIn('CPython>3', py2_info.interpreter_constraints)
# Cleanup.
os.remove(py2_pex)
def assert_pex_attribute(self, pex, attr, value):
self.assertTrue(os.path.exists(pex))
pex_info = PexInfo.from_pex(pex)
self.assertEqual(getattr(pex_info, attr), value)
colorized_isort_ptex_code_dir = os.path.join(tmpdir, 'colorized_isort_ptex_code_dir')
safe_mkdir(colorized_isort_ptex_code_dir)
code = []
for root, dirs, files in os.walk(colorized_isort_pex_code_dir):
rel_root = os.path.relpath(root, colorized_isort_pex_code_dir)
for f in files:
# Don't ship compiled python from the code extract above, the target interpreter will not
# match ours in general.
if f.endswith('.pyc'):
continue
rel_path = os.path.normpath(os.path.join(rel_root, f))
# The root __main__.py is special for any zipapp including pex, let it write its own
# __main__.py bootstrap. Similarly. PEX-INFO is special to pex and we want the PEX-INFO for
# The ptex pex, not the pex being ptexed.
if rel_path in ('__main__.py', PexInfo.PATH):
continue
os.symlink(os.path.join(root, f), os.path.join(colorized_isort_ptex_code_dir, rel_path))
code.append(rel_path)
ptex_code_dir = os.path.join(tmpdir, 'ptex_code_dir')
ptex_info = dict(code=code, resolver_settings=resolver_settings)
with safe_open(os.path.join(ptex_code_dir, 'PTEX-INFO'), 'w') as fp:
json.dump(ptex_info, fp)
with safe_open(os.path.join(ptex_code_dir, 'IPEX-INFO'), 'w') as fp:
fp.write(colorized_isort_pex_info.dump())
with safe_open(os.path.join(ptex_code_dir, 'ptex.py'), 'w') as fp:
fp.write(dedent("""\
import json
exec_pex_parent = os.path.join(self.workdir, 'executable_pex')
executable_file_content = self._get_executable_file_content(exec_pex_parent, modules)
hasher = hashlib.sha1()
hasher.update(reqs_pex.path().encode())
hasher.update(srcs_pex.path().encode())
hasher.update(executable_file_content.encode())
exec_file_hash = hasher.hexdigest()
exec_pex_path = os.path.realpath(os.path.join(exec_pex_parent, exec_file_hash))
if not os.path.isdir(exec_pex_path):
with safe_concurrent_creation(exec_pex_path) as safe_path:
# Write the entry point.
safe_mkdir(safe_path)
with open(os.path.join(safe_path, '{}.py'.format(self._EXEC_NAME)), 'w') as outfile:
outfile.write(executable_file_content)
pex_info = (target.pexinfo if isinstance(target, PythonBinary) else None) or PexInfo()
# Override any user-specified entry point, under the assumption that the
# executable_file_content does what the user intends (including, probably, calling that
# underlying entry point).
pex_info.entry_point = self._EXEC_NAME
pex_info.pex_path = ':'.join(pex.path() for pex in (reqs_pex, srcs_pex) if pex)
builder = PEXBuilder(safe_path, interpreter, pex_info=pex_info)
builder.freeze(bytecode_compile=False)
pex = PEX(exec_pex_path, interpreter)
with self.context.new_workunit(name='eval',
labels=[WorkUnitLabel.COMPILER, WorkUnitLabel.RUN,
WorkUnitLabel.TOOL],
cmd=' '.join(pex.cmdline())) as workunit:
returncode = pex.run(stdout=workunit.output('stdout'), stderr=workunit.output('stderr'))
workunit.set_outcome(WorkUnit.SUCCESS if returncode == 0 else WorkUnit.FAILURE)