Merge pull request #7242 from nicoddemus/fix-flaky-test

Introduce _pytest.timing as a way to control timing during internal tests

Conflicts:
  	src/_pytest/pytester.py
	  src/_pytest/runner.py
	  src/_pytest/terminal.py
  	testing/acceptance_test.py

testing/acceptance_test.py

@@ -895,21 +895,20 @@ class TestInvocationVariants:
 
 class TestDurations:
     source = """
-        import time
-        frag = 0.002
+        from _pytest import timing
         def test_something():
             pass
         def test_2():
-            time.sleep(frag*5)
+            timing.sleep(0.010)
         def test_1():
-            time.sleep(frag)
+            timing.sleep(0.002)
         def test_3():
-            time.sleep(frag*10)
+            timing.sleep(0.020)
     """
 
-    def test_calls(self, testdir):
+    def test_calls(self, testdir, mock_timing):
         testdir.makepyfile(self.source)
-        result = testdir.runpytest("--durations=10")
+        result = testdir.runpytest_inprocess("--durations=10")
         assert result.ret == 0
         result.stdout.fnmatch_lines_random(
             ["*durations*", "*call*test_3*", "*call*test_2*"]
@@ -918,16 +917,17 @@ class TestDurations:
             ["(0.00 durations hidden.  Use -vv to show these durations.)"]
         )
 
-    def test_calls_show_2(self, testdir):
+    def test_calls_show_2(self, testdir, mock_timing):
+
         testdir.makepyfile(self.source)
-        result = testdir.runpytest("--durations=2")
+        result = testdir.runpytest_inprocess("--durations=2")
         assert result.ret == 0
         lines = result.stdout.get_lines_after("*slowest*durations*")
         assert "4 passed" in lines[2]
 
-    def test_calls_showall(self, testdir):
+    def test_calls_showall(self, testdir, mock_timing):
         testdir.makepyfile(self.source)
-        result = testdir.runpytest("--durations=0")
+        result = testdir.runpytest_inprocess("--durations=0")
         assert result.ret == 0
         for x in "23":
             for y in ("call",):  # 'setup', 'call', 'teardown':
@@ -937,9 +937,9 @@ class TestDurations:
                 else:
                     raise AssertionError("not found {} {}".format(x, y))
 
-    def test_calls_showall_verbose(self, testdir):
+    def test_calls_showall_verbose(self, testdir, mock_timing):
         testdir.makepyfile(self.source)
-        result = testdir.runpytest("--durations=0", "-vv")
+        result = testdir.runpytest_inprocess("--durations=0", "-vv")
         assert result.ret == 0
         for x in "123":
             for y in ("call",):  # 'setup', 'call', 'teardown':
@@ -949,13 +949,13 @@ class TestDurations:
                 else:
                     raise AssertionError("not found {} {}".format(x, y))
 
-    def test_with_deselected(self, testdir):
+    def test_with_deselected(self, testdir, mock_timing):
         testdir.makepyfile(self.source)
-        result = testdir.runpytest("--durations=2", "-k test_2")
+        result = testdir.runpytest_inprocess("--durations=2", "-k test_2")
         assert result.ret == 0
         result.stdout.fnmatch_lines(["*durations*", "*call*test_2*"])
 
-    def test_with_failing_collection(self, testdir):
+    def test_with_failing_collection(self, testdir, mock_timing):
         testdir.makepyfile(self.source)
         testdir.makepyfile(test_collecterror="""xyz""")
         result = testdir.runpytest("--durations=2", "-k test_1")
@@ -965,36 +965,35 @@ class TestDurations:
         # output
         result.stdout.no_fnmatch_line("*duration*")
 
-    def test_with_not(self, testdir):
+    def test_with_not(self, testdir, mock_timing):
         testdir.makepyfile(self.source)
-        result = testdir.runpytest("-k not 1")
+        result = testdir.runpytest_inprocess("-k not 1")
         assert result.ret == 0
 
 
 class TestDurationWithFixture:
     source = """
         import pytest
-        import time
-        frag = 0.01
+        from _pytest import timing
 
         @pytest.fixture
         def setup_fixt():
-            time.sleep(frag)
+            timing.sleep(2)
 
         def test_1(setup_fixt):
-            time.sleep(frag)
+            timing.sleep(5)
     """
 
-    def test_setup_function(self, testdir):
+    def test_setup_function(self, testdir, mock_timing):
         testdir.makepyfile(self.source)
-        result = testdir.runpytest("--durations=10")
+        result = testdir.runpytest_inprocess("--durations=10")
         assert result.ret == 0
 
         result.stdout.fnmatch_lines_random(
             """
             *durations*
-            * setup *test_1*
-            * call *test_1*
+            5.00s call *test_1*
+            2.00s setup *test_1*
         """
         )
 

testing/conftest.py

@@ -195,3 +195,41 @@ def color_mapping():
                 pytest.skip("doing limited testing because lacking ordered markup")
 
     return ColorMapping
+
+
+@pytest.fixture
+def mock_timing(monkeypatch):
+    """Mocks _pytest.timing with a known object that can be used to control timing in tests
+    deterministically.
+
+    pytest itself should always use functions from `_pytest.timing` instead of `time` directly.
+
+    This then allows us more control over time during testing, if testing code also
+    uses `_pytest.timing` functions.
+
+    Time is static, and only advances through `sleep` calls, thus tests might sleep over large
+    numbers and obtain accurate time() calls at the end, making tests reliable and instant.
+    """
+    import attr
+
+    @attr.s
+    class MockTiming:
+
+        _current_time = attr.ib(default=1590150050.0)
+
+        def sleep(self, seconds):
+            self._current_time += seconds
+
+        def time(self):
+            return self._current_time
+
+        def patch(self):
+            from _pytest import timing
+
+            monkeypatch.setattr(timing, "sleep", self.sleep)
+            monkeypatch.setattr(timing, "time", self.time)
+            monkeypatch.setattr(timing, "perf_counter", self.time)
+
+    result = MockTiming()
+    result.patch()
+    return result

testing/test_junitxml.py

@@ -200,23 +200,23 @@ class TestPython:
         timestamp = datetime.strptime(node["timestamp"], "%Y-%m-%dT%H:%M:%S.%f")
         assert start_time <= timestamp < datetime.now()
 
-    def test_timing_function(self, testdir, run_and_parse):
+    def test_timing_function(self, testdir, run_and_parse, mock_timing):
         testdir.makepyfile(
             """
-            import time, pytest
+            from _pytest import timing
             def setup_module():
-                time.sleep(0.01)
+                timing.sleep(1)
             def teardown_module():
-                time.sleep(0.01)
+                timing.sleep(2)
             def test_sleep():
-                time.sleep(0.01)
+                timing.sleep(4)
         """
         )
         result, dom = run_and_parse()
         node = dom.find_first_by_tag("testsuite")
         tnode = node.find_first_by_tag("testcase")
         val = tnode["time"]
-        assert round(float(val), 2) >= 0.03
+        assert float(val) == 7.0
 
     @pytest.mark.parametrize("duration_report", ["call", "total"])
     def test_junit_duration_report(