Add fix for sloppy tests that mess with CWD and break coverage's internals. Fixes nedbat/coveragepy#881 and #306 others.

Author: ionelmc

src/pytest_cov/engine.py

@@ -1,6 +1,7 @@
 """Coverage controllers for use by pytest-cov and nose-cov."""
 import contextlib
 import copy
+import functools
 import os
 import random
 import socket
@@ -31,6 +32,25 @@ def _backup(obj, attr):
         setattr(obj, attr, backup)
 
 
+def _ensure_topdir(meth):
+    @functools.wraps(meth)
+    def ensure_topdir_wrapper(self, *args, **kwargs):
+        try:
+            original_cwd = os.getcwd()
+        except OSError:
+            # Looks like it's gone, this is non-ideal because a side-effect will
+            # be introduced in the tests here but we can't do anything about it.
+            original_cwd = None
+        os.chdir(self.topdir)
+        try:
+            return meth(self, *args, **kwargs)
+        finally:
+            if original_cwd is not None:
+                os.chdir(original_cwd)
+
+    return ensure_topdir_wrapper
+
+
 class CovController(object):
     """Base class for different plugin implementations."""
 
@@ -50,15 +70,26 @@ def __init__(self, cov_source, cov_report, cov_config, cov_append, cov_branch, c
         self.node_descs = set()
         self.failed_workers = []
         self.topdir = os.getcwd()
+        self.is_collocated = None
 
+    @contextlib.contextmanager
+    def ensure_topdir(self):
+        original_cwd = os.getcwd()
+        os.chdir(self.topdir)
+        yield
+        os.chdir(original_cwd)
+
+    @_ensure_topdir
     def pause(self):
         self.cov.stop()
         self.unset_env()
 
+    @_ensure_topdir
     def resume(self):
         self.cov.start()
         self.set_env()
 
+    @_ensure_topdir
     def set_env(self):
         """Put info about coverage into the env so that subprocesses can activate coverage."""
         if self.cov_source is None:
@@ -99,6 +130,7 @@ def sep(stream, s, txt):
             out = '%s %s %s\n' % (s * sep_len, txt, s * (sep_len + sep_extra))
             stream.write(out)
 
+    @_ensure_topdir
     def summary(self, stream):
         """Produce coverage reports."""
         total = None
@@ -171,6 +203,7 @@ def summary(self, stream):
 class Central(CovController):
     """Implementation for centralised operation."""
 
+    @_ensure_topdir
     def start(self):
         cleanup()
 
@@ -190,6 +223,7 @@ def start(self):
         self.cov.start()
         self.set_env()
 
+    @_ensure_topdir
     def finish(self):
         """Stop coverage, save data to file and set the list of coverage objects to report on."""
 
@@ -209,6 +243,7 @@ def finish(self):
 class DistMaster(CovController):
     """Implementation for distributed master."""
 
+    @_ensure_topdir
     def start(self):
         cleanup()
 
@@ -259,7 +294,7 @@ def testnodedown(self, node, error):
                 socket.gethostname(), os.getpid(),
                 random.randint(0, 999999),
                 output['cov_worker_node_id']
-                )
+            )
 
             cov = coverage.Coverage(source=self.cov_source,
                                     branch=self.cov_branch,
@@ -284,6 +319,7 @@ def testnodedown(self, node, error):
         node_desc = self.get_node_desc(rinfo.platform, rinfo.version_info)
         self.node_descs.add(node_desc)
 
+    @_ensure_topdir
     def finish(self):
         """Combines coverage data and sets the list of coverage objects to report on."""
 
@@ -299,7 +335,9 @@ def finish(self):
 class DistWorker(CovController):
     """Implementation for distributed workers."""
 
+    @_ensure_topdir
     def start(self):
+
         cleanup()
 
         # Determine whether we are collocated with master.
@@ -323,6 +361,7 @@ def start(self):
         self.cov.start()
         self.set_env()
 
+    @_ensure_topdir
     def finish(self):
         """Stop coverage and send relevant info back to the master."""
         self.unset_env()

src/pytest_cov/plugin.py

@@ -142,6 +142,7 @@ def __init__(self, options, pluginmanager, start=True):
         self.cov_total = None
         self.failed = False
         self._started = False
+        self._start_path = None
         self._disabled = False
         self.options = options
 
@@ -189,6 +190,7 @@ class Config(object):
         )
         self.cov_controller.start()
         self._started = True
+        self._start_path = os.getcwd()
         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

tests/test_pytest_cov.py

@@ -544,6 +544,43 @@ def test_central_with_path_aliasing(testdir, monkeypatch, opts, prop):
     assert result.ret == 0
 
 
+@xdist_params
+def test_borken_cwd(testdir, monkeypatch, opts):
+    mod = testdir.makepyfile(mod='''
+def foobar(a, b):
+    return a + b
+''')
+
+    script = testdir.makepyfile('''
+import os
+import pytest
+import mod
+
+@pytest.fixture
+def bad():
+    if not os.path.exists('/tmp/crappo'):
+        os.mkdir('/tmp/crappo')
+    os.chdir('/tmp/crappo')
+    yield
+    os.rmdir('/tmp/crappo')
+
+def test_foobar(bad):
+    assert mod.foobar(1, 2) == 3
+''')
+    result = testdir.runpytest('-v', '-s',
+                               '--cov=mod',
+                               '--cov-branch',
+                               script, *opts.split())
+
+    result.stdout.fnmatch_lines([
+        '*- coverage: platform *, python * -*',
+        '*mod* 100%',
+        '*1 passed*',
+    ])
+
+    assert result.ret == 0
+
+
 def test_subprocess_with_path_aliasing(testdir, monkeypatch):
     src = testdir.mkdir('src')
     src.join('parent_script.py').write(SCRIPT_PARENT)