def run_python_test()

in src/python/pants/backend/python/rules/python_test_runner.py [0:0]

• 88 lines of code
• 17 McCabe index (conditional complexity)

def run_python_test(test_target, pytest, python_setup, source_root_config, pex_build_environment, subprocess_encoding_environment):
  """Runs pytest for one target."""

  # TODO: Inject versions and digests here through some option, rather than hard-coding it.
  url = 'https://github.com/pantsbuild/pex/releases/download/v1.6.6/pex'
  digest = Digest('61bb79384db0da8c844678440bd368bcbfac17bbdb865721ad3f9cb0ab29b629', 1826945)
  pex_snapshot = yield Get(Snapshot, UrlToFetch(url, digest))

  # TODO(7726): replace this with a proper API to get the `closure` for a
  # TransitiveHydratedTarget.
  transitive_hydrated_targets = yield Get(
    TransitiveHydratedTargets, BuildFileAddresses((test_target.address,))
  )
  all_targets = [t.adaptor for t in transitive_hydrated_targets.closure]

  # Produce a pex containing pytest and all transitive 3rdparty requirements.
  all_target_requirements = []
  for maybe_python_req_lib in all_targets:
    # This is a python_requirement()-like target.
    if hasattr(maybe_python_req_lib, 'requirement'):
      all_target_requirements.append(str(maybe_python_req_lib.requirement))
    # This is a python_requirement_library()-like target.
    if hasattr(maybe_python_req_lib, 'requirements'):
      for py_req in maybe_python_req_lib.requirements:
        all_target_requirements.append(str(py_req.requirement))

  # Sort all user requirement strings to increase the chance of cache hits across invocations.
  all_requirements = sorted(all_target_requirements + list(pytest.get_requirement_strings()))

  # TODO(#7061): This str() can be removed after we drop py2!
  python_binary = text_type(sys.executable)
  interpreter_constraint_args = parse_interpreter_constraints(
    python_setup, python_target_adaptors=all_targets
  )
  interpreter_search_paths = text_type(create_path_env_var(python_setup.interpreter_search_paths))

  # TODO: This is non-hermetic because the requirements will be resolved on the fly by
  # pex27, where it should be hermetically provided in some way.
  output_pytest_requirements_pex_filename = 'pytest-with-requirements.pex'
  requirements_pex_argv = [
    python_binary,
    './{}'.format(pex_snapshot.files[0]),
    '-e', 'pytest:main',
    '-o', output_pytest_requirements_pex_filename,
  ] + interpreter_constraint_args + [
    # TODO(#7061): This text_type() wrapping can be removed after we drop py2!
    text_type(req) for req in all_requirements
  ]
  pex_resolve_env = {'PATH': interpreter_search_paths}
  # TODO(#6071): merge the two dicts via ** unpacking once we drop Py2.
  pex_resolve_env.update(pex_build_environment.invocation_environment_dict)
  requirements_pex_request = ExecuteProcessRequest(
    argv=tuple(requirements_pex_argv),
    env=pex_resolve_env,
    input_files=pex_snapshot.directory_digest,
    description='Resolve requirements: {}'.format(", ".join(all_requirements)),
    output_files=(output_pytest_requirements_pex_filename,),
  )
  requirements_pex_response = yield Get(
    ExecuteProcessResult, ExecuteProcessRequest, requirements_pex_request)

  source_roots = source_root_config.get_source_roots()

  # Gather sources and adjust for the source root.
  # TODO: make TargetAdaptor return a 'sources' field with an empty snapshot instead of raising to
  # simplify the hasattr() checks here!
  # TODO(7714): restore the full source name for the stdout of the Pytest run.
  sources_snapshots_and_source_roots = []
  for maybe_source_target in all_targets:
    if hasattr(maybe_source_target, 'sources'):
      tgt_snapshot = maybe_source_target.sources.snapshot
      tgt_source_root = source_roots.find_by_path(maybe_source_target.address.spec_path)
      sources_snapshots_and_source_roots.append((tgt_snapshot, tgt_source_root))
  all_sources_digests = yield [
    Get(
      Digest,
      DirectoryWithPrefixToStrip(
        directory_digest=snapshot.directory_digest,
        prefix=source_root.path
      )
    )
    for snapshot, source_root
    in sources_snapshots_and_source_roots
  ]

  sources_digest = yield Get(
    Digest, DirectoriesToMerge(directories=tuple(all_sources_digests)),
  )

  inits_digest = yield Get(InjectedInitDigest, Digest, sources_digest)

  all_input_digests = [
    sources_digest,
    inits_digest.directory_digest,
    requirements_pex_response.output_directory_digest,
  ]
  merged_input_files = yield Get(
    Digest,
    DirectoriesToMerge,
    DirectoriesToMerge(directories=tuple(all_input_digests)),
  )

  pex_exe_env = {'PATH': interpreter_search_paths}
  # TODO(#6071): merge the two dicts via ** unpacking once we drop Py2.
  pex_exe_env.update(subprocess_encoding_environment.invocation_environment_dict)

  request = ExecuteProcessRequest(
    argv=(python_binary, './{}'.format(output_pytest_requirements_pex_filename)),
    env=pex_exe_env,
    input_files=merged_input_files,
    description='Run pytest for {}'.format(test_target.address.reference()),
  )

  result = yield Get(FallibleExecuteProcessResult, ExecuteProcessRequest, request)
  status = Status.SUCCESS if result.exit_code == 0 else Status.FAILURE

  yield TestResult(
    status=status,
    stdout=result.stdout.decode('utf-8'),
    stderr=result.stderr.decode('utf-8'),
  )