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_get_current_platform():
expected_platforms = [Platform.current(), 'linux-x86_64']
assert set(expected_platforms) == set(PythonChroot.get_platforms(['current', 'linux-x86_64']))
def test_run_pex():
incompatible_platforms_warning_msg = 'WARNING: attempting to run PEX with incompatible platforms!'
assert incompatible_platforms_warning_msg not in assert_run_pex()
assert incompatible_platforms_warning_msg not in assert_run_pex(pex_args=['--platform=current'])
assert incompatible_platforms_warning_msg not in assert_run_pex(
pex_args=['--platform={}'.format(Platform.current())]
)
py27 = ensure_python_interpreter(PY27)
stderr_lines = assert_run_pex(python=py27, pex_args=['--platform=macosx-10.13-x86_64-cp-37-m'])
assert incompatible_platforms_warning_msg in stderr_lines
options.cache_dir = make_relative_to_root(options.cache_dir)
options.interpreter_cache_dir = make_relative_to_root(options.interpreter_cache_dir)
with ENV.patch(PEX_VERBOSE=str(options.verbosity)):
with TRACER.timed('Building pex'):
pex_builder = build_pex(reqs, options, resolver_options_builder)
if options.pex_name is not None:
log('Saving PEX file to %s' % options.pex_name, v=options.verbosity)
tmp_name = options.pex_name + '~'
safe_delete(tmp_name)
pex_builder.build(tmp_name)
os.rename(tmp_name, options.pex_name)
return 0
if options.platform != Platform.current():
log('WARNING: attempting to run PEX with differing platform!')
pex_builder.freeze()
log('Running PEX file at %s with args %s' % (pex_builder.path(), cmdline), v=options.verbosity)
pex = PEX(pex_builder.path(), interpreter=pex_builder.interpreter)
sys.exit(pex.run(args=list(cmdline)))
def expand_and_maybe_adjust_platform(interpreter, platform):
"""Adjusts `platform` if it is 'current' and does not match the given `interpreter` platform.
:param interpreter: The target interpreter for the given `platform`.
:type interpreter: :class:`pex.interpreter.PythonInterpreter`
:param platform: The platform name to expand and maybe adjust.
:type platform: text
:returns: The `platform`, potentially adjusted.
:rtype: :class:`pex.platforms.Platform`
"""
# TODO(John Sirois): Kill all usages when https://github.com/pantsbuild/pex/issues/511 is fixed.
cur_plat = Platform.current()
if cur_plat.platform != Platform.create(platform).platform:
# IE: Say we're on OSX and platform was 'linux-x86_64' or 'linux_x86_64-cp-27-cp27mu'.
return Platform.create(platform)
ii = interpreter.identity
if (ii.abbr_impl, ii.impl_ver, ii.abi_tag) == (cur_plat.impl, cur_plat.version, cur_plat.abi):
# IE: Say we're on Linux and platform was 'current' or 'linux-x86_64' or
# 'linux_x86_64-cp-27-cp27mu'and the current extended platform info matches the given
# interpreter exactly.
return cur_plat
# Otherwise we need to adjust the platform to match a local interpreter different from the
# currently executing interpreter.
interpreter_platform = Platform(platform=cur_plat.platform,
impl=ii.abbr_impl,
:param platforms: A list of :class:`Platform`s to resolve requirements for.
Defaults to the platforms specified by PythonSetup.
"""
# Gather and de-dup all requirements.
reqs = OrderedSet()
for req_lib in req_libs:
for req in req_lib.requirements:
reqs.add(req)
# See which ones we need to build.
reqs_to_build = OrderedSet()
find_links = OrderedSet()
for req in reqs:
# TODO: should_build appears to be hardwired to always be True. Get rid of it?
if req.should_build(interpreter.python, Platform.current()):
reqs_to_build.add(req)
log.debug(' Dumping requirement: {}'.format(req))
builder.add_requirement(req.requirement)
if req.repository:
find_links.add(req.repository)
else:
log.debug(' Skipping {} based on version filter'.format(req))
# Resolve the requirements into distributions.
distributions = _resolve_multi(interpreter, reqs_to_build, platforms, find_links)
locations = set()
for platform, dists in distributions.items():
for dist in dists:
if dist.location not in locations:
log.debug(' Dumping distribution: .../{}'.format(os.path.basename(dist.location)))
def make_build_properties(cls, interpreter=None):
from .interpreter import PythonInterpreter
from .platforms import Platform
pi = interpreter or PythonInterpreter.get()
plat = Platform.current()
platform_name = plat.platform
return {
'pex_version': pex_version,
'class': pi.identity.interpreter,
'version': pi.identity.version,
'platform': platform_name,
}
def __new__(cls, platform, impl, version, abi):
if not all((platform, impl, version, abi)):
raise cls.InvalidPlatformError(
'Platform specifiers cannot have blank fields. Given platform={platform!r}, impl={impl!r}, '
'version={version!r}, abi={abi!r}'.format(
platform=platform, impl=impl, version=version, abi=abi
)
)
platform = platform.replace('-', '_').replace('.', '_')
abi = cls._maybe_prefix_abi(impl, version, abi)
return super(Platform, cls).__new__(cls, platform, impl, version, abi)
generated_reqs.add(self._generate_thrift_requirement(thr))
generated_reqs.add(PythonRequirement('thrift', use_2to3=True))
for antlr in targets['antlrs']:
generated_reqs.add(self._generate_antlr_requirement(antlr))
reqs_from_libraries = OrderedSet()
for req_lib in targets['reqs']:
for req in req_lib.payload.requirements:
reqs_from_libraries.add(req)
reqs_to_build = OrderedSet()
find_links = OrderedSet()
for req in reqs_from_libraries | generated_reqs | self._extra_requirements:
if not req.should_build(self._interpreter.python, Platform.current()):
self.debug('Skipping {} based upon version filter'.format(req))
continue
reqs_to_build.add(req)
self._dump_requirement(req.requirement)
if req.repository:
find_links.add(req.repository)
distributions = self._resolve_multi(reqs_to_build, find_links)
locations = set()
for platform, dist_set in distributions.items():
for dist in dist_set:
if dist.location not in locations:
self._dump_distribution(dist)
locations.add(dist.location)
'{index: 2d}: {requirement}'
'{rendered_requirers}'
'\n But this pex only contains:'
'\n {distributions}'.format(
index=index + 1,
requirement=requirement,
rendered_requirers=rendered_requirers,
distributions='\n '.join(os.path.basename(d.location)
for d in distributions_by_key[requirement.key])
)
)
die(
'Failed to execute PEX file. Needed {platform} compatible dependencies for:\n{items}'
.format(
platform=Platform.of_interpreter(self._interpreter),
items='\n'.join(items)
)
)
return resolveds
def process_platform(option, option_str, option_value, parser):
platforms = getattr(parser.values, option.dest, [])
try:
platforms.append(parsed_platform(option_value))
except Platform.InvalidPlatformError as e:
raise OptionValueError("The {} option is invalid:\n{}"
.format(option_str, e))