multiple cov-failed-under flags

src/pytest_cov/engine.py

@@ -1,5 +1,7 @@
 """Coverage controllers for use by pytest-cov and nose-cov."""
 
+import collections
+import copy
 import os
 import random
 import socket
@@ -12,6 +14,19 @@
 from .compat import StringIO, workeroutput, workerinput
 
 
+_CovResult = collections.namedtuple("_CovResult", "result,config")
+
+
+def _work_around_config_mutation(cov, method, *args, **kwargs):
+    fn = getattr(cov, method)
+    old_config = cov.config
+    try:
+        cov.config = config = copy.copy(old_config)
+        return _CovResult(fn(*args, **kwargs), config)
+    finally:
+        cov.config = old_config
+
+
 class CovController(object):
     """Base class for different plugin implementations."""
 
@@ -80,14 +95,19 @@ def sep(stream, s, txt):
             out = '%s %s %s\n' % (s * sep_len, txt, s * (sep_len + sep_extra))
             stream.write(out)
 
-    def summary(self, stream):
+    def summary(self, cov_fail_under, stream):
         """Produce coverage reports."""
-        total = 0
 
-        if not self.cov_report:
+        def _get_total(cfu):
             with open(os.devnull, 'w') as null:
-                total = self.cov.report(show_missing=True, ignore_errors=True, file=null)
-                return total
+                return _work_around_config_mutation(
+                    self.cov, "report", show_missing=True, ignore_errors=True, include=cfu.include, omit=cfu.omit, file=null
+                ).result
+
+        totals = {cfu: _get_total(cfu) for cfu in cov_fail_under}
+
+        if not self.cov_report:
+            return totals
 
         # Output coverage section header.
         if len(self.node_descs) == 1:
@@ -107,29 +127,26 @@ def summary(self, stream):
             skip_covered = isinstance(self.cov_report, dict) and 'skip-covered' in self.cov_report.values()
             if hasattr(coverage, 'version_info') and coverage.version_info[0] >= 4:
                 options.update({'skip_covered': skip_covered or None})
-            total = self.cov.report(**options)
+            _work_around_config_mutation(self.cov, "report", **options)
 
         # Produce annotated source code report if wanted.
         if 'annotate' in self.cov_report:
             annotate_dir = self.cov_report['annotate']
-            self.cov.annotate(ignore_errors=True, directory=annotate_dir)
-            # We need to call Coverage.report here, just to get the total
-            # Coverage.annotate don't return any total and we need it for --cov-fail-under.
-            total = self.cov.report(ignore_errors=True, file=StringIO())
+            _work_around_config_mutation(self.cov, "annotate", ignore_errors=True, directory=annotate_dir)
             if annotate_dir:
                 stream.write('Coverage annotated source written to dir %s\n' % annotate_dir)
             else:
                 stream.write('Coverage annotated source written next to source\n')
 
         # Produce html report if wanted.
         if 'html' in self.cov_report:
-            total = self.cov.html_report(ignore_errors=True, directory=self.cov_report['html'])
-            stream.write('Coverage HTML written to dir %s\n' % self.cov.config.html_dir)
+            v = _work_around_config_mutation(self.cov, "html_report", ignore_errors=True, directory=self.cov_report['html'])
+            stream.write('Coverage HTML written to dir %s\n' % v.config.html_dir)
 
         # Produce xml report if wanted.
         if 'xml' in self.cov_report:
-            total = self.cov.xml_report(ignore_errors=True, outfile=self.cov_report['xml'])
-            stream.write('Coverage XML written to file %s\n' % self.cov.config.xml_output)
+            v = _work_around_config_mutation(self.cov, "xml_report", ignore_errors=True, outfile=self.cov_report['xml'])
+            stream.write('Coverage XML written to file %s\n' % v.config.xml_output)
 
         # Report on any failed workers.
         if self.failed_workers:
@@ -139,7 +156,7 @@ def summary(self, stream):
             for node in self.failed_workers:
                 stream.write('%s\n' % node.gateway.id)
 
-        return total
+        return totals
 
 
 class Central(CovController):
@@ -323,7 +340,7 @@ def finish(self):
                 'cov_worker_data': buff.getvalue(),
             })
 
-    def summary(self, stream):
+    def summary(self, *args, **kwargs):
         """Only the master reports so do nothing."""
 
         pass

src/pytest_cov/plugin.py

@@ -47,15 +47,57 @@ def __call__(self, parser, namespace, values, option_string=None):
         namespace.cov_report[report_type] = file
 
 
-def pytest_addoption(parser):
-    """Add options to control coverage."""
-
-    def fail_under(num_str):
+class FailUnder(object):
+    @staticmethod
+    def __number(num_str):
+        if num_str.endswith('%'):
+            num_str = num_str[:-1]
         try:
             return int(num_str)
         except ValueError:
             return float(num_str)
 
+    @classmethod
+    def parse(cls, unparsed):
+        items = iter(unparsed.split(":"))
+        limit = cls.__number(next(items, None))
+
+        include = []
+        omit = []
+        for item in items:
+            if item.startswith("-"):
+                omit.append(item[1:])
+                continue
+            if item.startswith("+"):
+                include.append(item[1:])
+                continue
+            include.append(item)
+
+        return cls(
+            limit=limit,
+            include=include or None,
+            omit=omit or None,
+        )
+
+    def __str__(self):
+        return ":".join(
+            ['{}%'.format(self.limit)]
+            + ['+{}'.format(v) for v in (self.include or [])]
+            + ['-{}'.format(v) for v in (self.omit or [])]
+        )
+
+    def __repr__(self):
+        return "FailUnder.parse({})".format(self)
+
+    def __init__(self, limit, include=None, omit=None):
+        self.limit = limit
+        self.include = include
+        self.omit = omit
+
+
+def pytest_addoption(parser):
+    """Add options to control coverage."""
+
     group = parser.getgroup(
         'cov', 'coverage reporting with distributed testing support')
     group.addoption('--cov', action='append', default=[], metavar='SOURCE',
@@ -79,7 +121,7 @@ def fail_under(num_str):
     group.addoption('--no-cov', action='store_true', default=False,
                     help='Disable coverage report completely (useful for debuggers). '
                          'Default: False')
-    group.addoption('--cov-fail-under', action='store', metavar='MIN', type=fail_under,
+    group.addoption('--cov-fail-under', nargs="?", action='append', default=[], metavar='MIN', type=FailUnder.parse,
                     help='Fail if the total coverage is less than MIN.')
     group.addoption('--cov-append', action='store_true', default=False,
                     help='Do not delete coverage but append to current. '
@@ -125,7 +167,7 @@ def __init__(self, options, pluginmanager, start=True):
         self.pid = None
         self.cov_controller = None
         self.cov_report = compat.StringIO()
-        self.cov_total = None
+        self.cov_totals = None
         self.failed = False
         self._started = False
         self._disabled = False
@@ -172,8 +214,8 @@ class Config(object):
         self.cov_controller.start()
         self._started = True
         cov_config = self.cov_controller.cov.config
-        if self.options.cov_fail_under is None and hasattr(cov_config, 'fail_under'):
-            self.options.cov_fail_under = cov_config.fail_under
+        if not self.options.cov_fail_under and hasattr(cov_config, 'fail_under'):
+            self.options.cov_fail_under = [FailUnder(limit=cov_config.fail_under)]
 
     def _is_worker(self, session):
         return compat.workerinput(session.config, None) is not None
@@ -218,8 +260,8 @@ def _should_report(self):
         return not (self.failed and self.options.no_cov_on_fail)
 
     def _failed_cov_total(self):
-        cov_fail_under = self.options.cov_fail_under
-        return cov_fail_under is not None and self.cov_total < cov_fail_under
+        totals = self.cov_totals
+        return any(totals.get(cfu, 0) < cfu.limit for cfu in self.options.cov_fail_under)
 
     # we need to wrap pytest_runtestloop. by the time pytest_sessionfinish
     # runs, it's too late to set testsfailed
@@ -238,7 +280,10 @@ def pytest_runtestloop(self, session):
 
         if not self._is_worker(session) and self._should_report():
             try:
-                self.cov_total = self.cov_controller.summary(self.cov_report)
+                self.cov_totals = self.cov_controller.summary(
+                    self.options.cov_fail_under,
+                    self.cov_report,
+                )
             except CoverageException as exc:
                 message = 'Failed to generate report: %s\n' % exc
                 session.config.pluginmanager.getplugin("terminalreporter").write(
@@ -247,8 +292,8 @@ def pytest_runtestloop(self, session):
                     warnings.warn(pytest.PytestWarning(message))
                 else:
                     session.config.warn(code='COV-2', message=message)
-                self.cov_total = 0
-            assert self.cov_total is not None, 'Test coverage should never be `None`'
+                self.cov_totals = {}
+            assert self.cov_totals is not None, 'Test coverage should never be `None`'
             if self._failed_cov_total():
                 # make sure we get the EXIT_TESTSFAILED exit code
                 compat_session.testsfailed += 1
@@ -265,21 +310,24 @@ def pytest_terminal_summary(self, terminalreporter):
         if self.cov_controller is None:
             return
 
-        if self.cov_total is None:
+        if self.cov_totals is None:
             # we shouldn't report, or report generation failed (error raised above)
             return
 
         terminalreporter.write('\n' + self.cov_report.getvalue() + '\n')
+        for cfu in self.options.cov_fail_under:
+            if cfu.limit <= 0:
+                continue
 
-        if self.options.cov_fail_under is not None and self.options.cov_fail_under > 0:
-            failed = self.cov_total < self.options.cov_fail_under
+            total = self.cov_totals.get(cfu, 0)
+            failed = total < cfu.limit
             markup = {'red': True, 'bold': True} if failed else {'green': True}
             message = (
-                '{fail}Required test coverage of {required}% {reached}. '
+                '{fail}Required test coverage of {required} {reached}. '
                 'Total coverage: {actual:.2f}%\n'
                 .format(
-                    required=self.options.cov_fail_under,
-                    actual=self.cov_total,
+                    required=cfu,
+                    actual=total,
                     fail="FAIL " if failed else "",
                     reached="not reached" if failed else "reached"
                 )

tests/test_pytest_cov.py

@@ -1,3 +1,4 @@
+import textwrap
 import glob
 import os
 import platform
@@ -357,6 +358,29 @@ def test_cov_min_float_value(testdir):
     ])
 
 
+def test_cov_min_multi(testdir):
+    script = textwrap.dedent("""\
+    import sut
+
+    def some_legacy_code():
+        strategy = sut.AbstractFactoryBean()
+        assert strategy.business_critical()
+    """)
+    script = testdir.makepyfile(SCRIPT, legacy=script)
+
+    result = testdir.runpytest('-v',
+                               '--cov=%s' % script.dirpath(),
+                               '--cov-report=term-missing',
+                               '--cov-fail-under=61.53',
+                               '--cov-fail-under=88.88:test_*',
+                               script)
+    assert result.ret == 0
+    result.stdout.fnmatch_lines([
+        'Required test coverage of 61.53% reached. Total coverage: 61.54%',
+        'Required test coverage of 88.88%:+test_* reached. Total coverage: 88.89%',
+    ])
+
+
 def test_cov_min_float_value_not_reached(testdir):
     script = testdir.makepyfile(SCRIPT)