Merge pull request #120 from onlyanegg/issue-87

Adding test timing functionality

tests/test_collect.py

@@ -3,18 +3,14 @@
 from pathlib import Path
 from pkgutil import ModuleInfo
 
-from tests.test_util import make_project
-from ward import test, fixture, raises
-from ward.collect import (
-    search_generally,
-    is_test_module,
-    get_module_path,
-    is_excluded_module,
-    remove_excluded_paths,
-    handled_within,
-)
+from ward import fixture, raises, test
+from ward.collect import (get_module_path, handled_within, is_excluded_module,
+                          is_test_module, remove_excluded_paths,
+                          search_generally)
 from ward.testing import Test, each
 
+from tests.test_util import make_project
+
 
 def named():
     assert "fox" == "fox"

tests/test_testing.py

@@ -1,16 +1,16 @@
+import sys
 from collections import defaultdict
 from pathlib import Path
 from unittest import mock
 from unittest.mock import Mock
 
-import sys
-
-from tests.utilities import testable_test, FORCE_TEST_PATH
-from ward import raises, Scope
+from ward import Scope, raises
 from ward.errors import ParameterisationError
-from ward.fixtures import fixture
+from ward.fixtures import FixtureCache, fixture
 from ward.models import WardMeta
-from ward.testing import Test, test, each, ParamMeta
+from ward.testing import ParamMeta, Test, each, test
+
+from tests.utilities import FORCE_TEST_PATH, testable_test
 
 
 def f():
@@ -41,6 +41,11 @@ def _(a=x):
     return Test(fn=_, module_name=mod)
 
 
+@fixture()
+def cache():
+    return FixtureCache()
+
+
 @test("Test.name should return the name of the function it wraps")
 def _(anonymous_test=anonymous_test):
     assert anonymous_test.name == "_"
@@ -77,12 +82,20 @@ def _(anonymous_test=anonymous_test):
     assert not anonymous_test.has_deps
 
 
-@test("Test.__call__ should delegate to the function it wraps")
-def _():
-    mock = Mock()
-    t = Test(fn=mock, module_name=mod)
-    t(1, 2, key="val")
-    mock.assert_called_once_with(1, 2, key="val")
+@test("Test.run should delegate to the function it wraps")
+def _(cache: FixtureCache = cache):
+    called_with = None
+    call_kwargs = (), {}
+
+    def func(key="val", **kwargs):
+        nonlocal called_with, call_kwargs
+        called_with = key
+        call_kwargs = kwargs
+
+    t = Test(fn=func, module_name=mod)
+    t.run(cache)
+    assert called_with == "val"
+    assert call_kwargs == {'kwargs': {}}
 
 
 @test("Test.is_parameterised should return True for parameterised test")
@@ -179,22 +192,23 @@ def invalid_test(a=each(1, 2), b=each(3, 4, 5)):
 def i_print_something():
     print("out")
     sys.stderr.write("err")
+    raise Exception
 
 
 @test("stdout/stderr are captured by default when a test is called")
-def _():
+def _(cache: FixtureCache = cache):
     t = Test(fn=i_print_something, module_name="")
-    t()
-    assert t.sout.getvalue() == "out\n"
-    assert t.serr.getvalue() == "err"
+    result = t.run(cache)
+    assert result.captured_stdout == "out\n"
+    assert result.captured_stderr == "err"
 
 
 @test("stdout/stderr are not captured when Test.capture_output = False")
-def _():
+def _(cache: FixtureCache = cache):
     t = Test(fn=i_print_something, module_name="", capture_output=False)
-    t()
-    assert t.sout.getvalue() == ""
-    assert t.serr.getvalue() == ""
+    result = t.run(cache)
+    assert result.captured_stdout == ""
+    assert result.captured_stderr == ""
 
 
 @fixture

ward/run.py

@@ -1,11 +1,10 @@
+import sys
 from pathlib import Path
 from timeit import default_timer
 from typing import Optional, Tuple
 
 import click
-import sys
 from colorama import init
-
 from ward._ward_version import __version__
 from ward.collect import (
     get_info_for_modules,
@@ -75,6 +74,12 @@
     is_eager=True,
     help="Look for tests in PATH.",
 )
+@click.option(
+    "--show-slowest",
+    type=int,
+    help="Record and display duration of n longest running tests",
+    default=0,
+)
 @click.pass_context
 def run(
     ctx: click.Context,
@@ -86,6 +91,7 @@ def run(
     order: str,
     capture_output: bool,
     config: str,
+    show_slowest: int,
 ):
     start_run = default_timer()
     paths = [Path(p) for p in path]
@@ -107,7 +113,7 @@ def run(
         test_results, time_to_collect=time_to_collect, fail_limit=fail_limit
     )
     time_taken = default_timer() - start_run
-    writer.output_test_result_summary(results, time_taken)
+    writer.output_test_result_summary(results, time_taken, show_slowest)
 
     exit_code = get_exit_code(results)
 

ward/suite.py

@@ -4,9 +4,8 @@
 from typing import Generator, List
 
 from ward import Scope
-from ward.errors import FixtureError
 from ward.fixtures import FixtureCache
-from ward.testing import Test, TestOutcome, TestResult
+from ward.testing import Test, TestResult
 
 
 @dataclass
@@ -26,39 +25,24 @@ def _test_counts_per_module(self):
         return counts
 
     def generate_test_runs(self, order="standard") -> Generator[TestResult, None, None]:
+        """
+        Run tests
+
+        Returns a generator which yields test results
+        """
+
         if order == "random":
             shuffle(self.tests)
 
         num_tests_per_module = self._test_counts_per_module()
         for test in self.tests:
             generated_tests = test.get_parameterised_instances()
             num_tests_per_module[test.path] -= 1
-            for i, generated_test in enumerate(generated_tests):
-                marker = generated_test.marker.name if generated_test.marker else None
-                if marker == "SKIP":
-                    yield generated_test.get_result(TestOutcome.SKIP)
-                    continue
-
-                try:
-                    resolved_vals = generated_test.resolve_args(self.cache, iteration=i)
-                    generated_test.format_description(resolved_vals)
-                    generated_test(**resolved_vals)
-                    outcome = (
-                        TestOutcome.XPASS if marker == "XFAIL" else TestOutcome.PASS
-                    )
-                    yield generated_test.get_result(outcome)
-                except FixtureError as e:
-                    yield generated_test.get_result(TestOutcome.FAIL, e)
-                    continue
-                except Exception as e:
-                    outcome = (
-                        TestOutcome.XFAIL if marker == "XFAIL" else TestOutcome.FAIL
-                    )
-                    yield generated_test.get_result(outcome, e)
-                finally:
-                    self.cache.teardown_fixtures_for_scope(
-                        Scope.Test, scope_key=generated_test.id
-                    )
+            for generated_test in generated_tests:
+                yield generated_test.run(self.cache)
+                self.cache.teardown_fixtures_for_scope(
+                    Scope.Test, scope_key=generated_test.id
+                )
 
             if num_tests_per_module[test.path] == 0:
                 self.cache.teardown_fixtures_for_scope(

ward/terminal.py

@@ -1,23 +1,22 @@
 import inspect
 import os
 import platform
+import sys
 import traceback
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
-from textwrap import wrap, indent
-from typing import Any, Dict, Generator, List, Optional, Iterable
+from textwrap import indent, wrap
+from typing import Any, Dict, Generator, Iterable, List, Optional
 
-import sys
 from colorama import Fore, Style
 from pygments import highlight
 from pygments.formatters.terminal import TerminalFormatter
 from pygments.lexers.python import PythonLexer
 from termcolor import colored, cprint
-
 from ward._ward_version import __version__
 from ward.diff import make_diff
-from ward.expect import TestFailure, Comparison
+from ward.expect import Comparison, TestFailure
 from ward.suite import Suite
 from ward.testing import TestOutcome, TestResult
 
@@ -40,11 +39,23 @@ def multiline_description(s: str, indent: int, width: int) -> str:
     return rv
 
 
-def output_test_result_line(test_result: TestResult):
-    colour = outcome_to_colour(test_result.outcome)
-    bg = f"on_{colour}"
-    padded_outcome = f" {test_result.outcome.name[:4]} "
+def format_test_id(test_result: TestResult) -> (str, str):
+    """
+    Format module name, line number, and test case number
+    """
+
+    test_id = lightblack(
+        f"{format_test_location(test_result)}{format_test_case_number(test_result)}:"
+    )
+
+    return test_id
+
+
+def format_test_location(test_result: TestResult) -> str:
+    return f"{test_result.test.module_name}:{test_result.test.line_number}"
 
+
+def format_test_case_number(test_result: TestResult) -> str:
     # If we're executing a parameterised test
     param_meta = test_result.test.param_meta
     if param_meta.group_size > 1:
@@ -55,11 +66,16 @@ def output_test_result_line(test_result: TestResult):
     else:
         iter_indicator = ""
 
-    mod_name = lightblack(
-        f"{test_result.test.module_name}:"
-        f"{test_result.test.line_number}"
-        f"{iter_indicator}:"
-    )
+    return iter_indicator
+
+
+def output_test_result_line(test_result: TestResult):
+    colour = outcome_to_colour(test_result.outcome)
+    bg = f"on_{colour}"
+    padded_outcome = f" {test_result.outcome.name[:4]} "
+
+    iter_indicator = format_test_case_number(test_result)
+    mod_name = format_test_id(test_result)
     if (
         test_result.outcome == TestOutcome.SKIP
         or test_result.outcome == TestOutcome.XFAIL
@@ -260,7 +276,7 @@ def output_why_test_failed_header(self, test_result: TestResult):
         raise NotImplementedError()
 
     def output_test_result_summary(
-        self, test_results: List[TestResult], time_taken: float
+        self, test_results: List[TestResult], time_taken: float, duration: int
     ):
         raise NotImplementedError()
 
@@ -385,8 +401,10 @@ def result_checkbox(self, expect):
         return result_marker
 
     def output_test_result_summary(
-        self, test_results: List[TestResult], time_taken: float
+        self, test_results: List[TestResult], time_taken: float, show_slowest: int
     ):
+        if show_slowest:
+            self._output_slowest_tests(test_results, show_slowest)
         outcome_counts = self._get_outcome_counts(test_results)
         if test_results:
             chart = self.generate_chart(
@@ -424,6 +442,19 @@ def output_test_result_summary(
 
         print(output)
 
+    def _output_slowest_tests(self, test_results: List[TestResult], num_tests: int):
+        test_results = sorted(
+            test_results, key=lambda r: r.test.timer.duration, reverse=True
+        )
+        self.print_divider()
+        heading = f"{colored('Longest Running Tests:', color='cyan', attrs=['bold'])}\n"
+        print(indent(heading, INDENT))
+        for result in test_results[:num_tests]:
+            test_id = format_test_id(result)
+            message = f"{result.test.timer.duration:.2f} sec {test_id} {result.test.description} "
+            print(indent(message, DOUBLE_INDENT))
+        print()
+
     def output_captured_stderr(self, test_result: TestResult):
         if test_result.captured_stderr:
             captured_stderr_lines = test_result.captured_stderr.split("\n")