Show code snippets on demand

mypy/build.py

@@ -36,7 +36,8 @@
 from mypy.indirection import TypeIndirectionVisitor
 from mypy.errors import Errors, CompileError, ErrorInfo, report_internal_error
 from mypy.util import (
-    DecodeError, decode_python_encoding, is_sub_path, get_mypy_comments, module_prefix
+    DecodeError, decode_python_encoding, is_sub_path, get_mypy_comments, module_prefix,
+    read_py_file
 )
 if TYPE_CHECKING:
     from mypy.report import Reports  # Avoid unconditional slow import
@@ -197,9 +198,12 @@ def _build(sources: List[BuildSource],
         reports = Reports(data_dir, options.report_dirs)
 
     source_set = BuildSourceSet(sources)
+    cached_read = fscache.read
     errors = Errors(options.show_error_context,
                     options.show_column_numbers,
-                    options.show_error_codes)
+                    options.show_error_codes,
+                    options.pretty,
+                    lambda path: read_py_file(path, cached_read, options.python_version))
     plugin, snapshot = load_plugins(options, errors, stdout)
 
     # Construct a build manager object to hold state during the build.

mypy/dmypy/client.py

@@ -18,7 +18,7 @@
 from mypy.dmypy_util import DEFAULT_STATUS_FILE, receive
 from mypy.ipc import IPCClient, IPCException
 from mypy.dmypy_os import alive, kill
-from mypy.util import check_python_version
+from mypy.util import check_python_version, get_terminal_width
 
 from mypy.version import __version__
 
@@ -469,6 +469,7 @@ def request(status_file: str, command: str, *, timeout: Optional[int] = None,
     # Tell the server whether this request was initiated from a human-facing terminal,
     # so that it can format the type checking output accordingly.
     args['is_tty'] = sys.stdout.isatty()
+    args['terminal_width'] = get_terminal_width()
     bdata = json.dumps(args).encode('utf8')
     _, name = get_status(status_file)
     try:

mypy/dmypy_server.py

@@ -263,6 +263,7 @@ def run_command(self, command: str, data: Dict[str, object]) -> Dict[str, object
             if command not in {'check', 'recheck', 'run'}:
                 # Only the above commands use some error formatting.
                 del data['is_tty']
+                del data['terminal_width']
             elif int(os.getenv('MYPY_FORCE_COLOR', '0')):
                 data['is_tty'] = True
             return method(self, **data)
@@ -290,7 +291,8 @@ def cmd_stop(self) -> Dict[str, object]:
         os.unlink(self.status_file)
         return {}
 
-    def cmd_run(self, version: str, args: Sequence[str], is_tty: bool) -> Dict[str, object]:
+    def cmd_run(self, version: str, args: Sequence[str],
+                is_tty: bool, terminal_width: int) -> Dict[str, object]:
         """Check a list of files, triggering a restart if needed."""
         try:
             # Process options can exit on improper arguments, so we need to catch that and
@@ -323,18 +325,20 @@ def cmd_run(self, version: str, args: Sequence[str], is_tty: bool) -> Dict[str,
             return {'out': '', 'err': str(err), 'status': 2}
         except SystemExit as e:
             return {'out': stdout.getvalue(), 'err': stderr.getvalue(), 'status': e.code}
-        return self.check(sources, is_tty)
+        return self.check(sources, is_tty, terminal_width)
 
-    def cmd_check(self, files: Sequence[str], is_tty: bool) -> Dict[str, object]:
+    def cmd_check(self, files: Sequence[str],
+                  is_tty: bool, terminal_width: int) -> Dict[str, object]:
         """Check a list of files."""
         try:
             sources = create_source_list(files, self.options, self.fscache)
         except InvalidSourceList as err:
             return {'out': '', 'err': str(err), 'status': 2}
-        return self.check(sources, is_tty)
+        return self.check(sources, is_tty, terminal_width)
 
     def cmd_recheck(self,
                     is_tty: bool,
+                    terminal_width: int,
                     remove: Optional[List[str]] = None,
                     update: Optional[List[str]] = None) -> Dict[str, object]:
         """Check the same list of files we checked most recently.
@@ -360,21 +364,23 @@ def cmd_recheck(self,
         t1 = time.time()
         manager = self.fine_grained_manager.manager
         manager.log("fine-grained increment: cmd_recheck: {:.3f}s".format(t1 - t0))
-        res = self.fine_grained_increment(sources, is_tty, remove, update)
+        res = self.fine_grained_increment(sources, is_tty, terminal_width,
+                                          remove, update)
         self.fscache.flush()
         self.update_stats(res)
         return res
 
-    def check(self, sources: List[BuildSource], is_tty: bool) -> Dict[str, Any]:
+    def check(self, sources: List[BuildSource],
+              is_tty: bool, terminal_width: int) -> Dict[str, Any]:
         """Check using fine-grained incremental mode.
 
         If is_tty is True format the output nicely with colors and summary line
-        (unless disabled in self.options).
+        (unless disabled in self.options). Also pass the terminal_width to formatter.
         """
         if not self.fine_grained_manager:
-            res = self.initialize_fine_grained(sources, is_tty)
+            res = self.initialize_fine_grained(sources, is_tty, terminal_width)
         else:
-            res = self.fine_grained_increment(sources, is_tty)
+            res = self.fine_grained_increment(sources, is_tty, terminal_width)
         self.fscache.flush()
         self.update_stats(res)
         return res
@@ -387,7 +393,7 @@ def update_stats(self, res: Dict[str, Any]) -> None:
             manager.stats = {}
 
     def initialize_fine_grained(self, sources: List[BuildSource],
-                                is_tty: bool) -> Dict[str, Any]:
+                                is_tty: bool, terminal_width: int) -> Dict[str, Any]:
         self.fswatcher = FileSystemWatcher(self.fscache)
         t0 = time.time()
         self.update_sources(sources)
@@ -449,12 +455,13 @@ def initialize_fine_grained(self, sources: List[BuildSource],
             print_memory_profile(run_gc=False)
 
         status = 1 if messages else 0
-        messages = self.pretty_messages(messages, len(sources), is_tty)
+        messages = self.pretty_messages(messages, len(sources), is_tty, terminal_width)
         return {'out': ''.join(s + '\n' for s in messages), 'err': '', 'status': status}
 
     def fine_grained_increment(self,
                                sources: List[BuildSource],
                                is_tty: bool,
+                               terminal_width: int,
                                remove: Optional[List[str]] = None,
                                update: Optional[List[str]] = None,
                                ) -> Dict[str, Any]:
@@ -484,12 +491,16 @@ def fine_grained_increment(self,
 
         status = 1 if messages else 0
         self.previous_sources = sources
-        messages = self.pretty_messages(messages, len(sources), is_tty)
+        messages = self.pretty_messages(messages, len(sources), is_tty, terminal_width)
         return {'out': ''.join(s + '\n' for s in messages), 'err': '', 'status': status}
 
     def pretty_messages(self, messages: List[str], n_sources: int,
-                        is_tty: bool = False) -> List[str]:
+                        is_tty: bool = False, terminal_width: Optional[int] = None) -> List[str]:
         use_color = self.options.color_output and is_tty
+        fit_width = self.options.pretty and is_tty
+        if fit_width:
+            messages = self.formatter.fit_in_terminal(messages,
+                                                      fixed_terminal_width=terminal_width)
         if self.options.error_summary:
             summary = None  # type: Optional[str]
             if messages:

mypy/errors.py

@@ -3,14 +3,15 @@
 import traceback
 from collections import OrderedDict, defaultdict
 
-from typing import Tuple, List, TypeVar, Set, Dict, Optional, TextIO
+from typing import Tuple, List, TypeVar, Set, Dict, Optional, TextIO, Callable
 from typing_extensions import Final
 
 from mypy.scope import Scope
 from mypy.options import Options
 from mypy.version import __version__ as mypy_version
 from mypy.errorcodes import ErrorCode
 from mypy import errorcodes as codes
+from mypy.util import DEFAULT_SOURCE_OFFSET
 
 T = TypeVar('T')
 allowed_duplicates = ['@overload', 'Got:', 'Expected:']  # type: Final
@@ -156,10 +157,15 @@ class Errors:
     def __init__(self,
                  show_error_context: bool = False,
                  show_column_numbers: bool = False,
-                 show_error_codes: bool = False) -> None:
+                 show_error_codes: bool = False,
+                 pretty: bool = False,
+                 read_source: Optional[Callable[[str], Optional[List[str]]]] = None) -> None:
         self.show_error_context = show_error_context
         self.show_column_numbers = show_column_numbers
         self.show_error_codes = show_error_codes
+        self.pretty = pretty
+        # We use fscache to read source code when showing snippets.
+        self.read_source = read_source
         self.initialize()
 
     def initialize(self) -> None:
@@ -179,7 +185,11 @@ def reset(self) -> None:
         self.initialize()
 
     def copy(self) -> 'Errors':
-        new = Errors(self.show_error_context, self.show_column_numbers)
+        new = Errors(self.show_error_context,
+                     self.show_column_numbers,
+                     self.show_error_codes,
+                     self.pretty,
+                     self.read_source)
         new.file = self.file
         new.import_ctx = self.import_ctx[:]
         new.type_name = self.type_name[:]
@@ -402,10 +412,13 @@ def raise_error(self) -> None:
                            use_stdout=True,
                            module_with_blocker=self.blocker_module())
 
-    def format_messages(self, error_info: List[ErrorInfo]) -> List[str]:
+    def format_messages(self, error_info: List[ErrorInfo],
+                        source_lines: Optional[List[str]]) -> List[str]:
         """Return a string list that represents the error messages.
 
-        Use a form suitable for displaying to the user.
+        Use a form suitable for displaying to the user. If self.pretty
+        is True also append a relevant trimmed source code line (only for
+        severity 'error').
         """
         a = []  # type: List[str]
         errors = self.render_messages(self.sort_messages(error_info))
@@ -427,6 +440,17 @@ def format_messages(self, error_info: List[ErrorInfo]) -> List[str]:
                 # displaying duplicate error codes.
                 s = '{}  [{}]'.format(s, code.code)
             a.append(s)
+            if self.pretty:
+                # Add source code fragment and a location marker.
+                if severity == 'error' and source_lines and line > 0:
+                    source_line = source_lines[line - 1]
+                    if column < 0:
+                        # Something went wrong, take first non-empty column.
+                        column = len(source_line) - len(source_line.lstrip())
+                    # Note, currently coloring uses the offset to detect source snippets,
+                    # so these offsets should not be arbitrary.
+                    a.append(' ' * DEFAULT_SOURCE_OFFSET + source_line)
+                    a.append(' ' * (DEFAULT_SOURCE_OFFSET + column) + '^')
         return a
 
     def file_messages(self, path: str) -> List[str]:
@@ -437,7 +461,11 @@ def file_messages(self, path: str) -> List[str]:
         if path not in self.error_info_map:
             return []
         self.flushed_files.add(path)
-        return self.format_messages(self.error_info_map[path])
+        source_lines = None
+        if self.pretty:
+            assert self.read_source
+            source_lines = self.read_source(path)
+        return self.format_messages(self.error_info_map[path], source_lines)
 
     def new_messages(self) -> List[str]:
         """Return a string list of new error messages.

mypy/main.py

@@ -66,6 +66,8 @@ def main(script_path: Optional[str],
     formatter = util.FancyFormatter(stdout, stderr, options.show_error_codes)
 
     def flush_errors(new_messages: List[str], serious: bool) -> None:
+        if options.pretty:
+            new_messages = formatter.fit_in_terminal(new_messages)
         messages.extend(new_messages)
         f = stderr if serious else stdout
         try:
@@ -582,6 +584,11 @@ def add_invertible_flag(flag: str,
     add_invertible_flag('--show-error-codes', default=False,
                         help="Show error codes in error messages",
                         group=error_group)
+    add_invertible_flag('--pretty', default=False,
+                        help="Use visually nicer output in error messages:"
+                             " Use soft word wrap, show source code snippets,"
+                             " and error location markers",
+                        group=error_group)
     add_invertible_flag('--no-color-output', dest='color_output', default=True,
                         help="Do not colorize error messages",
                         group=error_group)

mypy/options.py

@@ -244,6 +244,8 @@ def __init__(self) -> None:
         self.shadow_file = None  # type: Optional[List[List[str]]]
         self.show_column_numbers = False  # type: bool
         self.show_error_codes = False
+        # Use soft word wrap and show trimmed source snippets with error location markers.
+        self.pretty = False
         self.dump_graph = False
         self.dump_deps = False
         self.logical_deps = False

mypy/test/testfinegrained.py

@@ -197,7 +197,7 @@ def get_options(self,
         return options
 
     def run_check(self, server: Server, sources: List[BuildSource]) -> List[str]:
-        response = server.check(sources, is_tty=False)
+        response = server.check(sources, is_tty=False, terminal_width=-1)
         out = cast(str, response['out'] or response['err'])
         return out.splitlines()
 

mypy/test/testformatter.py

@@ -0,0 +1,51 @@
+from unittest import TestCase, main
+
+from mypy.util import trim_source_line, split_words
+
+
+class FancyErrorFormattingTestCases(TestCase):
+    def test_trim_source(self) -> None:
+        assert trim_source_line('0123456789abcdef',
+                                max_len=16, col=5, min_width=2) == ('0123456789abcdef', 0)
+
+        # Locations near start.
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=0, min_width=2) == ('0123456...', 0)
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=4, min_width=2) == ('0123456...', 0)
+
+        # Middle locations.
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=5, min_width=2) == ('...1234567...', -2)
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=6, min_width=2) == ('...2345678...', -1)
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=8, min_width=2) == ('...456789a...', 1)
+
+        # Locations near the end.
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=11, min_width=2) == ('...789abcd...', 4)
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=13, min_width=2) == ('...9abcdef', 6)
+        assert trim_source_line('0123456789abcdef',
+                                max_len=7, col=15, min_width=2) == ('...9abcdef', 6)
+
+    def test_split_words(self) -> None:
+        assert split_words('Simple message') == ['Simple', 'message']
+        assert split_words('Message with "Some[Long, Types]"'
+                           ' in it') == ['Message', 'with',
+                                         '"Some[Long, Types]"', 'in', 'it']
+        assert split_words('Message with "Some[Long, Types]"'
+                           ' and [error-code]') == ['Message', 'with', '"Some[Long, Types]"',
+                                                    'and', '[error-code]']
+        assert split_words('"Type[Stands, First]" then words') == ['"Type[Stands, First]"',
+                                                                   'then', 'words']
+        assert split_words('First words "Then[Stands, Type]"') == ['First', 'words',
+                                                                   '"Then[Stands, Type]"']
+        assert split_words('"Type[Only, Here]"') == ['"Type[Only, Here]"']
+        assert split_words('OneWord') == ['OneWord']
+        assert split_words(' ') == ['', '']
+
+
+if __name__ == '__main__':
+    main()