<commit_before># coding=utf-8
# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import absolute_import, division, print_function, unicode_literals

import functools
import os
from builtins import open
from contextlib import contextmanager
from textwrap import dedent

import coverage
from six.moves import configparser

from pants.backend.python.tasks.gather_sources import GatherSources
from pants.backend.python.tasks.pytest_prep import PytestPrep
from pants.backend.python.tasks.pytest_run import PytestResult, PytestRun
from pants.backend.python.tasks.resolve_requirements import ResolveRequirements
from pants.backend.python.tasks.select_interpreter import SelectInterpreter
from pants.base.exceptions import ErrorWhileTesting, TaskError
from pants.build_graph.target import Target
from pants.source.source_root import SourceRootConfig
from pants.util.contextutil import pushd, temporary_dir, temporary_file
from pants.util.dirutil import safe_mkdtemp, safe_rmtree
from pants_test.backend.python.tasks.python_task_test_base import PythonTaskTestBase
from pants_test.subsystem.subsystem_util import init_subsystem
from pants_test.task_test_base import ensure_cached


class PytestTestBase(PythonTaskTestBase):
  @classmethod
  def task_type(cls):
    return PytestRun

  _CONFTEST_CONTENT = '# I am an existing root-level conftest file.'

  def run_tests(self, targets, *passthru_args, **options):
    """Run the tests in the specified targets, with the specified PytestRun task options."""
    context = self._prepare_test_run(targets, *passthru_args, **options)
    self._do_run_tests(context)
    return context

  def run_failing_tests(self, targets, failed_targets, *passthru_args, **options):
    context = self._prepare_test_run(targets, *passthru_args, **options)
    with self.assertRaises(ErrorWhileTesting) as cm:
      self._do_run_tests(context)
    self.assertEqual(set(failed_targets), set(cm.exception.failed_targets))
    return context

  def try_run_tests(self, targets, *passthru_args, **options):
    try:
      self.run_tests(targets, *passthru_args, **options)
      return []
    except ErrorWhileTesting as e:
      return e.failed_targets

  def _prepare_test_run(self, targets, *passthru_args, **options):
    test_options = {
      'colors': False,
      'level': 'info'  # When debugging a test failure it may be helpful to set this to 'debug'.
    }
    test_options.update(options)
    self.set_options(**test_options)

    # The easiest way to create products required by the PythonTest task is to
    # execute the relevant tasks.
    si_task_type = self.synthesize_task_subtype(SelectInterpreter, 'si_scope')
    rr_task_type = self.synthesize_task_subtype(ResolveRequirements, 'rr_scope')
    gs_task_type = self.synthesize_task_subtype(GatherSources, 'gs_scope')
    pp_task_type = self.synthesize_task_subtype(PytestPrep, 'pp_scope')
    context = self.context(for_task_types=[si_task_type, rr_task_type, gs_task_type, pp_task_type],
                           target_roots=targets,
                           passthru_args=list(passthru_args))
    si_task_type(context, os.path.join(self.pants_workdir, 'si')).execute()
    rr_task_type(context, os.path.join(self.pants_workdir, 'rr')).execute()
    gs_task_type(context, os.path.join(self.pants_workdir, 'gs')).execute()
    pp_task_type(context, os.path.join(self.pants_workdir, 'pp')).execute()
    return context

  def _do_run_tests(self, context):
    pytest_run_task = self.create_task(context)
    with pushd(self.build_root):
      pytest_run_task.execute()


class PytestTestEmpty(PytestTestBase):
  def test_empty(self):
    self.run_tests(targets=[])


class PytestTestFailedPexRun(PytestTestBase):<commit_msg>Work around production `coverage` float. (#6452)<commit_after># coding=utf-8
# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import absolute_import, division, print_function, unicode_literals

import functools
import os
from builtins import open
from contextlib import contextmanager
from textwrap import dedent

import coverage
from six.moves import configparser

from pants.backend.python.tasks.gather_sources import GatherSources
from pants.backend.python.tasks.pytest_prep import PytestPrep
from pants.backend.python.tasks.pytest_run import PytestResult, PytestRun
from pants.backend.python.tasks.resolve_requirements import ResolveRequirements
from pants.backend.python.tasks.select_interpreter import SelectInterpreter
from pants.base.exceptions import ErrorWhileTesting, TaskError
from pants.build_graph.target import Target
from pants.source.source_root import SourceRootConfig
from pants.util.contextutil import pushd, temporary_dir, temporary_file
from pants.util.dirutil import safe_mkdtemp, safe_rmtree
from pants_test.backend.python.tasks.python_task_test_base import PythonTaskTestBase
from pants_test.subsystem.subsystem_util import init_subsystem
from pants_test.task_test_base import ensure_cached


# NB: Our production code depends on `pytest-cov` which indirectly depends on `coverage`, but in
# this test we have a direct dependency on `coverage` in order to load data files and test that
# coverage is collected by `pytest-cov` and has expected values. Unfortunately, `pytest-cov` has a
# floating dependency on `coverage` and `coverage` has changed its data file format in the past (eg:
# https://pypi.org/project/coverage/5.0a2/). If the default data format differs between `coverage`
# float and `coverage` pinned, we'll fail to read coverage data here in this test. We work around
# this by adding a pinned `coverage` requirement that matches our test dependency to the
# `PytestPrep` production requirements here.
class PytestPrepCoverageVersionPinned(PytestPrep):
  def extra_requirements(self):
    extra_reqs = list(super(PytestPrepCoverageVersionPinned, self).extra_requirements())
    extra_reqs.append('coverage=={}'.format(coverage.__version__))
    return extra_reqs


class PytestTestBase(PythonTaskTestBase):
  @classmethod
  def task_type(cls):
    return PytestRun

  _CONFTEST_CONTENT = '# I am an existing root-level conftest file.'

  def run_tests(self, targets, *passthru_args, **options):
    """Run the tests in the specified targets, with the specified PytestRun task options."""
    context = self._prepare_test_run(targets, *passthru_args, **options)
    self._do_run_tests(context)
    return context

  def run_failing_tests(self, targets, failed_targets, *passthru_args, **options):
    context = self._prepare_test_run(targets, *passthru_args, **options)
    with self.assertRaises(ErrorWhileTesting) as cm:
      self._do_run_tests(context)
    self.assertEqual(set(failed_targets), set(cm.exception.failed_targets))
    return context

  def try_run_tests(self, targets, *passthru_args, **options):
    try:
      self.run_tests(targets, *passthru_args, **options)
      return []
    except ErrorWhileTesting as e:
      return e.failed_targets

  def _prepare_test_run(self, targets, *passthru_args, **options):
    test_options = {
      'colors': False,
      'level': 'info'  # When debugging a test failure it may be helpful to set this to 'debug'.
    }
    test_options.update(options)
    self.set_options(**test_options)

    # The easiest way to create products required by the PythonTest task is to
    # execute the relevant tasks.
    si_task_type = self.synthesize_task_subtype(SelectInterpreter, 'si_scope')
    rr_task_type = self.synthesize_task_subtype(ResolveRequirements, 'rr_scope')
    gs_task_type = self.synthesize_task_subtype(GatherSources, 'gs_scope')
    pp_task_type = self.synthesize_task_subtype(PytestPrepCoverageVersionPinned, 'pp_scope')
    context = self.context(for_task_types=[si_task_type, rr_task_type, gs_task_type, pp_task_type],
                           target_roots=targets,
                           passthru_args=list(passthru_args))
    si_task_type(context, os.path.join(self.pants_workdir, 'si')).execute()
    rr_task_type(context, os.path.join(self.pants_workdir, 'rr')).execute()
    gs_task_type(context, os.path.join(self.pants_workdir, 'gs')).execute()
    pp_task_type(context, os.path.join(self.pants_workdir, 'pp')).execute()
    return context

  def _do_run_tests(self, context):
    pytest_run_task = self.create_task(context)
    with pushd(self.build_root):
      pytest_run_task.execute()


class PytestTestEmpty(PytestTestBase):
  def test_empty(self):
    self.run_tests(targets=[])


class PytestTestFailedPexRun(PytestTestBase):
{'fraction_non_alphanumeric': 0.08607361255336334, 'fraction_numerical': 0.0024229837313949464, 'mean_word_length': 5.121468926553672, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5562841', 'n_tokens_mistral': 2730, 'n_tokens_neox': 2673, 'n_words': 659}