Improve the output of the inmanta compile and inmanta export comm…

…ands to make it more clear to the end-user when the command failed. (Issue #5258, PR #6532)

# Description

Improve visual appearance of the compiler on failure

closes #5258

# Self Check:

- [ ] Attached issue to pull request
- [ ] Changelog entry
- [ ] Type annotations are present
- [ ] Code is clear and sufficiently documented
- [ ] No (preventable) type errors (check using make mypy or make mypy-diff)
- [ ] Sufficient test cases (reproduces the bug/tests the requested feature)
- [ ] Correct, in line with design
- [ ] End user documentation is included or an issue is created for end-user documentation (add ref to issue here: )
- [ ] If this PR fixes a race condition in the test suite, also push the fix to the relevant stable branche(s) (see [test-fixes](https://internal.inmanta.com/development/core/tasks/build-master.html#test-fixes) for more info)

changelogs/unreleased/improve-visual-appearence-compiler.yml

@@ -0,0 +1,8 @@
+---
+description: Improve the output of the `inmanta compile` and `inmanta export` commands to make it more clear to the end-user when the command failed.
+issue-nr: 5258
+issue-repo: inmanta-core
+change-type: minor
+destination-branches: [master, iso6]
+sections:
+  minor-improvement: "{{description}}"

src/inmanta/app.py

@@ -31,9 +31,13 @@
 """
 import argparse
 import asyncio
+import contextlib
+import dataclasses
+import enum
 import json
 import logging
 import os
+import shutil
 import signal
 import socket
 import sys
@@ -48,6 +52,7 @@
 from types import FrameType
 from typing import Any, Callable, Coroutine, Dict, Optional
 
+import click
 from tornado import gen
 from tornado.httpclient import AsyncHTTPClient
 from tornado.ioloop import IOLoop
@@ -364,18 +369,23 @@ def compile_project(options: argparse.Namespace) -> None:
     )
     module.Project.get(options.main_file, strict_deps_check=strict_deps_check)
 
+    summary_reporter = CompileSummaryReporter()
     if options.profile:
         import cProfile
         import pstats
 
-        cProfile.runctx("do_compile()", globals(), {}, "run.profile")
+        with summary_reporter.compiler_exception.capture():
+            cProfile.runctx("do_compile()", globals(), {}, "run.profile")
         p = pstats.Stats("run.profile")
         p.strip_dirs().sort_stats("time").print_stats(20)
     else:
         t1 = time.time()
-        do_compile()
+        with summary_reporter.compiler_exception.capture():
+            do_compile()
         LOGGER.debug("Compile time: %0.03f seconds", time.time() - t1)
 
+    summary_reporter.print_summary_and_exit(show_stack_traces=options.errors)
+
 
 @command("list-commands", help_msg="Print out an overview of all commands", add_verbose_flag=False)
 def list_commands(options: argparse.Namespace) -> None:
@@ -591,40 +601,180 @@ def export(options: argparse.Namespace) -> None:
 
     from inmanta.export import Exporter  # noqa: H307
 
-    exp = None
+    summary_reporter = CompileSummaryReporter()
+
     types: Optional[Dict[str, inmanta_type.Type]]
     scopes: Optional[Namespace]
-    try:
-        (types, scopes) = do_compile()
-    except Exception as e:
-        exp = e
-        types, scopes = (None, None)
+
+    with summary_reporter.compiler_exception.capture():
+        try:
+            (types, scopes) = do_compile()
+        except Exception:
+            types, scopes = (None, None)
+            raise
 
     # Even if the compile failed we might have collected additional data such as unknowns. So
     # continue the export
 
     export = Exporter(options)
-    results = export.run(
-        types,
-        scopes,
-        metadata=metadata,
-        model_export=options.model_export,
-        export_plugin=options.export_plugin,
-        partial_compile=options.partial_compile,
-        resource_sets_to_remove=options.delete_resource_set,
-    )
-    version = results[0]
-
-    if exp is not None:
-        raise exp
+    with summary_reporter.exporter_exception.capture():
+        results = export.run(
+            types,
+            scopes,
+            metadata=metadata,
+            model_export=options.model_export,
+            export_plugin=options.export_plugin,
+            partial_compile=options.partial_compile,
+            resource_sets_to_remove=options.delete_resource_set,
+        )
 
-    if options.deploy:
+    if not summary_reporter.is_failure() and options.deploy:
+        version = results[0]
         conn = protocol.SyncClient("compiler")
         LOGGER.info("Triggering deploy for version %d" % version)
         tid = cfg_env.get()
         agent_trigger_method = const.AgentTriggerMethod.get_agent_trigger_method(options.full_deploy)
         conn.release_version(tid, version, True, agent_trigger_method)
 
+    summary_reporter.print_summary_and_exit(show_stack_traces=options.errors)
+
+
+class Color(enum.Enum):
+    RED = "red"
+    GREEN = "green"
+
+
+@dataclasses.dataclass
+class ExceptionCollector:
+    """
+    This class defines a context manager that captures any unhandled exception raised within the context.
+    """
+
+    exception: Optional[Exception] = None
+
+    def has_exception(self) -> bool:
+        return self.exception is not None
+
+    @contextlib.contextmanager
+    def capture(self) -> abc.Iterator["ExceptionCollector"]:
+        """
+        Record any exceptions raised within the context manager. The exception will not be re-raised.
+        """
+        try:
+            yield self
+        except Exception as e:
+            self.exception = e
+
+
+class CompileSummaryReporter:
+    """
+    Contains the logic to print a summary at the end of the `inmanta compile` or `inmanta export`
+    command that provides an overview on whether the command was successful or not.
+    """
+
+    def __init__(self) -> None:
+        self.compiler_exception: ExceptionCollector = ExceptionCollector()
+        self.exporter_exception: ExceptionCollector = ExceptionCollector()
+
+    def is_failure(self) -> bool:
+        """
+        Return true iff an exception has occurred during the compile or export stage.
+        """
+        return self.compiler_exception.has_exception() or self.exporter_exception.has_exception()
+
+    def _get_global_status(self) -> str:
+        """
+        Return the global status of the run.
+        """
+        if self.compiler_exception.has_exception():
+            return "COMPILATION FAILURE"
+        elif self.exporter_exception.has_exception():
+            return "EXPORT FAILURE"
+        else:
+            return "SUCCESS"
+
+    def _get_header(self, header_text: str) -> str:
+        """
+        Return a header for the summary with the given header_text.
+        """
+        terminal_width = shutil.get_terminal_size()[0]
+        minimal_header = f"= {header_text.upper()} ="
+        length_minimal_header = len(minimal_header)
+        if terminal_width <= length_minimal_header:
+            return minimal_header
+        else:
+            nr_equals_signs_to_add_each_side = int((terminal_width - length_minimal_header) / 2)
+            extra_equals_signs_each_side = "=" * nr_equals_signs_to_add_each_side
+            return f"{extra_equals_signs_each_side}{minimal_header}{extra_equals_signs_each_side}"
+
+    def _get_exception_to_report(self) -> Exception:
+        """
+        Return the exception that should be reported in the summary. Compiler exceptions take precedence
+        over exporter exceptions because they happen first.
+        """
+        assert self.is_failure()
+        if self.compiler_exception.has_exception():
+            assert self.compiler_exception.exception is not None
+            return self.compiler_exception.exception
+        else:
+            assert self.exporter_exception.exception is not None
+            return self.exporter_exception.exception
+
+    def _get_error_message(self) -> str:
+        """
+        Return the error message associated with `self._get_exception_to_report()`.
+        """
+        exc = self._get_exception_to_report()
+        if isinstance(exc, CompilerException):
+            error_message = exc.format_trace(indent="  ").strip("\n")
+            # Add explainer text if any
+            from inmanta.compiler.help.explainer import ExplainerFactory
+
+            helpmsg = ExplainerFactory().explain_and_format(exc, plain=not _is_on_tty())
+            if helpmsg is not None:
+                helpmsg = helpmsg.strip("\n")
+                return f"{error_message}\n\n{helpmsg}"
+        else:
+            error_message = str(exc).strip("\n")
+        return f"Error: {error_message}"
+
+    def _get_stack_trace(self) -> str:
+        """
+        Return the stack trace associated with `self._get_exception_to_report()`.
+        """
+        exc = self._get_exception_to_report()
+        return "".join(traceback.format_exception(exc)).strip("\n")
+
+    def _print_to_stderr(self, text: str = "", bold: bool = False, **kwargs: object) -> None:
+        """
+        Prints the given text to stderr with the given styling requirements. On a tty the text
+        is printed in green in case of success and in red in case of a failure.
+        """
+        if _is_on_tty():
+            color = Color.RED if self.is_failure() else Color.GREEN
+            text = click.style(text, fg=color.value, bold=bold)
+        print(text, file=sys.stderr, **kwargs)
+
+    def print_summary(self, show_stack_traces: bool) -> None:
+        """
+        Print the summary of the compile run.
+        """
+        self._print_to_stderr()
+        if show_stack_traces and self.is_failure():
+            self._print_to_stderr(text=self._get_header(header_text="EXCEPTION TRACE"))
+            self._print_to_stderr(text=self._get_stack_trace(), end="\n\n")
+
+        self._print_to_stderr(text=self._get_header(header_text=self._get_global_status()))
+        if self.is_failure():
+            self._print_to_stderr(text=self._get_error_message())
+
+    def print_summary_and_exit(self, show_stack_traces: bool) -> None:
+        """
+        Print the compile summary and exit with a 0 status code in case of success or 1 in case of failure.
+        """
+        self.print_summary(show_stack_traces)
+        exit(1 if self.is_failure() else 0)
+
 
 def cmd_parser() -> argparse.ArgumentParser:
     # create the argument compiler

tests/test_app.py

@@ -28,6 +28,7 @@
 
 import inmanta.util
 from inmanta import const
+from inmanta.app import CompileSummaryReporter
 
 
 def get_command(
@@ -558,3 +559,58 @@ def test_init_project(tmpdir):
     (stdout, stderr, return_code) = run_without_tty(args, killtime=15, termtime=10)
     assert return_code != 0
     assert any("already exists" in error for error in stderr)
+
+
+def test_compiler_summary_reporter(monkeypatch, capsys) -> None:
+    """
+    Test whether the CompileSummaryReporter class produces correct output.
+    """
+    # Success
+    summary_reporter = CompileSummaryReporter()
+    with summary_reporter.compiler_exception.capture():
+        pass
+    assert not summary_reporter.is_failure()
+    summary_reporter.print_summary(show_stack_traces=True)
+    assert re.match(r"\n=+ SUCCESS =+\n", capsys.readouterr().err)
+
+    # Compile failure
+    summary_reporter = CompileSummaryReporter()
+    with summary_reporter.compiler_exception.capture():
+        raise Exception("This is a compilation failure")
+    assert summary_reporter.is_failure()
+    summary_reporter.print_summary(show_stack_traces=False)
+    output = capsys.readouterr().err
+    assert re.match(r"\n=+ COMPILATION FAILURE =+\nError: This is a compilation failure\n", output)
+    assert "= EXCEPTION TRACE =" not in output
+    summary_reporter.print_summary(show_stack_traces=True)
+    output = capsys.readouterr().err
+    assert re.match(
+        r"\n=+ EXCEPTION TRACE =+\n(.|\n)*\n=+ COMPILATION FAILURE =+\nError: This is a compilation failure\n", output
+    )
+
+    # Compile failure and export failure
+    summary_reporter = CompileSummaryReporter()
+    with summary_reporter.compiler_exception.capture():
+        raise Exception("This is a compilation failure")
+    with summary_reporter.exporter_exception.capture():
+        raise Exception("This is an export failure")
+    assert summary_reporter.is_failure()
+    summary_reporter.print_summary(show_stack_traces=False)
+    output = capsys.readouterr().err
+    assert re.match(r"\n=+ COMPILATION FAILURE =+\nError: This is a compilation failure\n", output)
+    assert "= EXCEPTION TRACE =" not in output
+
+    # Export failure
+    summary_reporter = CompileSummaryReporter()
+    with summary_reporter.compiler_exception.capture():
+        pass
+    with summary_reporter.exporter_exception.capture():
+        raise Exception("This is an export failure")
+    assert summary_reporter.is_failure()
+    summary_reporter.print_summary(show_stack_traces=False)
+    output = capsys.readouterr().err
+    assert re.match(r"\n=+ EXPORT FAILURE =+\nError: This is an export failure\n", output)
+    assert "= EXCEPTION TRACE =" not in output
+    summary_reporter.print_summary(show_stack_traces=True)
+    output = capsys.readouterr().err
+    assert re.match(r"\n=+ EXCEPTION TRACE =+\n(.|\n)*\n=+ EXPORT FAILURE =+\nError: This is an export failure\n", output)