Fix(cicd_bot): Show warnings in the check output

Author: erindru

sqlmesh/core/console.py

@@ -3046,34 +3046,47 @@ class CaptureTerminalConsole(TerminalConsole):
     def __init__(self, console: t.Optional[RichConsole] = None, **kwargs: t.Any) -> None:
         super().__init__(console=console, **kwargs)
         self._captured_outputs: t.List[str] = []
+        self._warnings: t.List[str] = []
         self._errors: t.List[str] = []
 
     @property
     def captured_output(self) -> str:
         return "".join(self._captured_outputs)
 
+    @property
+    def captured_warnings(self) -> str:
+        return "".join(self._warnings)
+
     @property
     def captured_errors(self) -> str:
         return "".join(self._errors)
 
     def consume_captured_output(self) -> str:
-        output = self.captured_output
-        self.clear_captured_outputs()
-        return output
+        try:
+            return self.captured_output
+        finally:
+            self._captured_outputs = []
 
-    def consume_captured_errors(self) -> str:
-        errors = self.captured_errors
-        self.clear_captured_errors()
-        return errors
+    def consume_captured_warnings(self) -> str:
+        try:
+            return self.captured_warnings
+        finally:
+            self._warnings = []
 
-    def clear_captured_outputs(self) -> None:
-        self._captured_outputs = []
+    def consume_captured_errors(self) -> str:
+        try:
+            return self.captured_errors
+        finally:
+            self._errors = []
 
-    def clear_captured_errors(self) -> None:
-        self._errors = []
+    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
+        if short_message not in self._warnings:
+            self._warnings.append(short_message)
+        super().log_warning(short_message, long_message)
 
     def log_error(self, message: str) -> None:
-        self._errors.append(message)
+        if message not in self._errors:
+            self._errors.append(message)
         super().log_error(message)
 
     def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
@@ -3084,9 +3097,8 @@ def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
             super().log_skipped_models(snapshot_names)
 
     def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
-        if errors:
-            self._errors.append("\n".join(str(ex) for ex in errors))
-            super().log_failed_models(errors)
+        self._errors.extend([str(ex) for ex in errors])
+        super().log_failed_models(errors)
 
     def _print(self, value: t.Any, **kwargs: t.Any) -> None:
         with self.console.capture() as capture:
@@ -3431,18 +3443,25 @@ def show_linter_violations(
         self._print(msg)
         self._errors.append(msg)
 
-    def log_error(self, message: str) -> None:
-        super().log_error(f"```\n\\[ERROR] {message}```\n\n")
-
-    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
-        logger.warning(long_message or short_message)
-
-        if not short_message.endswith("\n"):
-            short_message += (
-                "\n"  # so that the closing ``` ends up on a newline which is important for GitHub
-            )
+    @property
+    def captured_warnings(self) -> str:
+        return self._render_alert_block("WARNING", self._warnings)
 
-        self._print(f"```\n\\[WARNING] {short_message}```\n\n")
+    @property
+    def captured_errors(self) -> str:
+        return self._render_alert_block("ERROR", self._errors)
+
+    def _render_alert_block(self, block_type: str, items: t.List[str]) -> str:
+        # GitHub Markdown alert syntax, https://docs.github.com/en/get-started/writing-on-github/getting-started-with-writing-and-formatting-on-github/basic-writing-and-formatting-syntax#alerts
+        if items:
+            block = f"> [!{block_type}]\n"
+            list_indicator = "- " if len(items) > 1 else ""
+            for item in items:
+                item = item.replace("\n", "\n> ")
+                block += f"> {list_indicator}{item}\n"
+            block += "\n"
+            return block
+        return ""
 
     def _print(self, value: t.Any, **kwargs: t.Any) -> None:
         self.console.print(value, **kwargs)
@@ -3469,7 +3488,7 @@ def _print(self, value: t.Any, **kwargs: t.Any) -> None:
         super()._print(value, **kwargs)
         for captured_output in self._captured_outputs:
             print(captured_output)
-        self.clear_captured_outputs()
+        self.consume_captured_output()
 
     def _prompt(self, message: str, **kwargs: t.Any) -> t.Any:
         self._print(message)

sqlmesh/integrations/github/cicd/command.py

@@ -119,10 +119,7 @@ def _update_pr_environment(controller: GithubController) -> bool:
     except Exception as e:
         logger.exception("Error occurred when updating PR environment")
         conclusion = controller.update_pr_environment_check(
-            status=GithubCheckStatus.COMPLETED,
-            exception=e,
-            plan=controller.pr_plan_or_none,
-            plan_flags=controller.pr_plan_flags,
+            status=GithubCheckStatus.COMPLETED, exception=e
         )
         return (
             conclusion is not None

sqlmesh/integrations/github/cicd/controller.py

@@ -494,7 +494,7 @@ def get_plan_summary(self, plan: Plan) -> str:
 
         try:
             # Clear out any output that might exist from prior steps
-            self._console.clear_captured_outputs()
+            self._console.consume_captured_output()
             if plan.restatements:
                 self._console._print("\n**Restating models**\n")
             else:
@@ -522,13 +522,118 @@ def get_plan_summary(self, plan: Plan) -> str:
             if not difference_summary and not missing_dates:
                 return f"No changes to apply.{plan_flags_section}"
 
-            return f"{difference_summary}\n{missing_dates}{plan_flags_section}"
+            warnings_block = self._console.captured_warnings
+            errors_block = self._console.captured_errors
+
+            return f"{warnings_block}{errors_block}{difference_summary}\n{missing_dates}{plan_flags_section}"
         except PlanError as e:
             logger.exception("Plan failed to generate")
             return f"Plan failed to generate. Check for pending or unresolved changes. Error: {e}"
         finally:
             self._console.verbosity = orig_verbosity
 
+    def get_pr_environment_summary(
+        self, conclusion: GithubCheckConclusion, exception: t.Optional[Exception] = None
+    ) -> str:
+        heading = ""
+        summary = ""
+
+        if conclusion.is_success:
+            summary = self._get_pr_environment_summary_success()
+        elif conclusion.is_action_required:
+            heading = f":warning: Action Required to create or update PR Environment `{self.pr_environment_name}` :warning:"
+            summary = self._get_pr_environment_summary_action_required(exception)
+        elif conclusion.is_failure:
+            heading = (
+                f":x: Failed to create or update PR Environment `{self.pr_environment_name}` :x:"
+            )
+            summary = self._get_pr_environment_summary_failure(exception)
+        elif conclusion.is_skipped:
+            heading = f":next_track_button: Skipped creating or updating PR Environment `{self.pr_environment_name}` :next_track_button:"
+            summary = self._get_pr_environment_summary_skipped(exception)
+        else:
+            heading = f":interrobang: Got an unexpected conclusion: {conclusion.value}"
+
+        # note: we just add warnings here, errors will be covered by the "failure" conclusion
+        if warnings := self._console.consume_captured_warnings():
+            summary = f"{warnings}\n{summary}"
+
+        return f"{heading}\n\n{summary}".strip()
+
+    def _get_pr_environment_summary_success(self) -> str:
+        prod_plan = self.prod_plan_with_gaps
+
+        if not prod_plan.has_changes:
+            summary = "No models were modified in this PR.\n"
+        else:
+            intro = self._generate_pr_environment_summary_intro()
+            summary = intro + self._generate_pr_environment_summary_list(prod_plan)
+
+        if prod_plan.user_provided_flags:
+            summary += self._generate_plan_flags_section(prod_plan.user_provided_flags)
+
+        return summary
+
+    def _get_pr_environment_summary_skipped(self, exception: t.Optional[Exception] = None) -> str:
+        if isinstance(exception, NoChangesPlanError):
+            skip_reason = "No changes were detected compared to the prod environment."
+        elif isinstance(exception, TestFailure):
+            skip_reason = "Unit Test(s) Failed so skipping PR creation"
+        else:
+            skip_reason = "A prior stage failed resulting in skipping PR creation."
+
+        return skip_reason
+
+    def _get_pr_environment_summary_action_required(
+        self, exception: t.Optional[Exception] = None
+    ) -> str:
+        plan = self.pr_plan_or_none
+        if isinstance(exception, UncategorizedPlanError) and plan:
+            failure_msg = f"The following models could not be categorized automatically:\n"
+            for snapshot in plan.uncategorized:
+                failure_msg += f"- {snapshot.name}\n"
+            failure_msg += (
+                f"\nRun `sqlmesh plan {self.pr_environment_name}` locally to apply these changes.\n\n"
+                "If you would like the bot to automatically categorize changes, check the [documentation](https://sqlmesh.readthedocs.io/en/stable/integrations/github/) for more information."
+            )
+        else:
+            failure_msg = "Please check the Actions Workflow logs for more information."
+
+        return failure_msg
+
+    def _get_pr_environment_summary_failure(self, exception: t.Optional[Exception] = None) -> str:
+        if exception:
+            logger.debug(f"Got {type(exception).__name__}. Stack trace: " + traceback.format_exc())
+
+        console_output = self._console.consume_captured_output()
+
+        if isinstance(exception, PlanError):
+            failure_msg = f"Plan application failed.\n"
+            if exception.args and (msg := exception.args[0]) and isinstance(msg, str):
+                failure_msg += f"\n{msg}\n"
+            if console_output:
+                failure_msg += f"\n{console_output}"
+        elif isinstance(exception, (SQLMeshError, SqlglotError, ValueError)):
+            # this logic is taken from the global error handler attached to the CLI, which uses `click.echo()` to output the message
+            # so cant be re-used here because it bypasses the Console
+            failure_msg = f"**Error:** {str(exception)}"
+        else:
+            logger.debug(
+                "Got unexpected error. Error Type: "
+                + str(type(exception))
+                + " Stack trace: "
+                + traceback.format_exc()
+            )
+            failure_msg = f"This is an unexpected error.\n\n**Exception:**\n```\n{traceback.format_exc()}\n```"
+
+        if captured_errors := self._console.consume_captured_errors():
+            failure_msg = f"{captured_errors}\n{failure_msg}"
+
+        if plan_flags := self.pr_plan_flags:
+            failure_msg += f"\n\n{self._generate_plan_flags_section(plan_flags)}"
+
+        return failure_msg
+
     def run_tests(self) -> t.Tuple[ModelTextTestResult, str]:
         """
         Run tests for the PR
@@ -539,7 +644,7 @@ def run_linter(self) -> None:
         """
         Run linter for the PR
         """
-        self._console.clear_captured_outputs()
+        self._console.consume_captured_output()
         self._context.lint_models()
 
     def _get_or_create_comment(self, header: str = BOT_HEADER_MSG) -> IssueComment:
@@ -611,7 +716,22 @@ def update_pr_environment(self) -> None:
         Creates a PR environment from the logic present in the PR. If the PR contains changes that are
         uncategorized, then an error will be raised.
         """
-        self._context.apply(self.pr_plan)
+        self._context.apply(self.pr_plan)  # will raise if PR environment creation fails
+
+        # update PR info comment
+        vde_title = "- :eyes: To **review** this PR's changes, use virtual data environment:"
+        comment_value = f"{vde_title}\n  - `{self.pr_environment_name}`"
+        if self.bot_config.enable_deploy_command:
+            comment_value += (
+                "\n- :arrow_forward: To **apply** this PR's plan to prod, comment:\n  - `/deploy`"
+            )
+        dedup_regex = vde_title.replace("*", r"\*") + r".*"
+        updated_comment, _ = self.update_sqlmesh_comment_info(
+            value=comment_value,
+            dedup_regex=dedup_regex,
+        )
+        if updated_comment:
+            self._append_output("created_pr_environment", "true")
 
     def deploy_to_prod(self) -> None:
         """
@@ -855,13 +975,7 @@ def conclusion_handler(
         )
 
     def update_pr_environment_check(
-        self,
-        status: GithubCheckStatus,
-        exception: t.Optional[Exception] = None,
-        plan: t.Optional[Plan] = None,
-        plan_flags: t.Optional[
-            t.Dict[str, UserProvidedFlags]
-        ] = None,  # note: the plan flags are passed separately in case the plan fails to build
+        self, status: GithubCheckStatus, exception: t.Optional[Exception] = None
     ) -> t.Optional[GithubCheckConclusion]:
         """
         Updates the status of the merge commit for the PR environment.
@@ -882,87 +996,7 @@ def update_pr_environment_check(
         def conclusion_handler(
             conclusion: GithubCheckConclusion, exception: t.Optional[Exception]
         ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
-            if conclusion.is_success:
-                prod_plan = self.prod_plan_with_gaps
-
-                if not prod_plan.has_changes:
-                    summary = "No models were modified in this PR.\n"
-                else:
-                    intro = self._generate_pr_environment_summary_intro()
-                    summary = intro + self._generate_pr_environment_summary_list(prod_plan)
-                    if prod_plan.user_provided_flags:
-                        summary += self._generate_plan_flags_section(prod_plan.user_provided_flags)
-
-                vde_title = (
-                    "- :eyes: To **review** this PR's changes, use virtual data environment:"
-                )
-                comment_value = f"{vde_title}\n  - `{self.pr_environment_name}`"
-                if self.bot_config.enable_deploy_command:
-                    comment_value += "\n- :arrow_forward: To **apply** this PR's plan to prod, comment:\n  - `/deploy`"
-                dedup_regex = vde_title.replace("*", r"\*") + r".*"
-                updated_comment, _ = self.update_sqlmesh_comment_info(
-                    value=comment_value,
-                    dedup_regex=dedup_regex,
-                )
-                if updated_comment:
-                    self._append_output("created_pr_environment", "true")
-            else:
-                if isinstance(exception, NoChangesPlanError):
-                    skip_reason = "No changes were detected compared to the prod environment."
-                elif isinstance(exception, TestFailure):
-                    skip_reason = "Unit Test(s) Failed so skipping PR creation"
-                else:
-                    skip_reason = "A prior stage failed resulting in skipping PR creation."
-
-                if not skip_reason and exception:
-                    logger.debug(
-                        f"Got {type(exception).__name__}. Stack trace: " + traceback.format_exc()
-                    )
-
-                captured_errors = self._console.consume_captured_errors()
-                if captured_errors:
-                    logger.debug(f"Captured errors: {captured_errors}")
-                    failure_msg = f"**Errors:**\n{captured_errors}\n"
-                elif isinstance(exception, UncategorizedPlanError) and plan:
-                    failure_msg = f"The following models could not be categorized automatically:\n"
-                    for snapshot in plan.uncategorized:
-                        failure_msg += f"- {snapshot.name}\n"
-                    failure_msg += (
-                        f"\nRun `sqlmesh plan {self.pr_environment_name}` locally to apply these changes.\n\n"
-                        "If you would like the bot to automatically categorize changes, check the [documentation](https://sqlmesh.readthedocs.io/en/stable/integrations/github/) for more information."
-                    )
-                elif isinstance(exception, PlanError):
-                    failure_msg = f"Plan application failed.\n"
-                    if exception.args and (msg := exception.args[0]) and isinstance(msg, str):
-                        failure_msg += f"\n{msg}\n"
-                    if self._console.captured_output:
-                        failure_msg += f"\n{self._console.captured_output}"
-                elif isinstance(exception, (SQLMeshError, SqlglotError, ValueError)):
-                    # this logic is taken from the global error handler attached to the CLI, which uses `click.echo()` to output the message
-                    # so cant be re-used here because it bypasses the Console
-                    failure_msg = f"**Error:** {str(exception)}"
-                else:
-                    logger.debug(
-                        "Got unexpected error. Error Type: "
-                        + str(type(exception))
-                        + " Stack trace: "
-                        + traceback.format_exc()
-                    )
-                    failure_msg = f"This is an unexpected error.\n\n**Exception:**\n```\n{traceback.format_exc()}\n```"
-
-                conclusion_to_summary = {
-                    GithubCheckConclusion.SKIPPED: f":next_track_button: Skipped creating or updating PR Environment `{self.pr_environment_name}`. {skip_reason}",
-                    GithubCheckConclusion.FAILURE: f":x: Failed to create or update PR Environment `{self.pr_environment_name}` :x:\n\n{failure_msg}",
-                    GithubCheckConclusion.CANCELLED: f":stop_sign: Cancelled creating or updating PR Environment `{self.pr_environment_name}`",
-                    GithubCheckConclusion.ACTION_REQUIRED: f":warning: Action Required to create or update PR Environment `{self.pr_environment_name}` :warning:\n\n{failure_msg}",
-                }
-                summary = conclusion_to_summary.get(
-                    conclusion, f":interrobang: Got an unexpected conclusion: {conclusion.value}"
-                )
-                if plan_flags:
-                    plan_flags_section = self._generate_plan_flags_section(plan_flags)
-                    summary += f"\n\n{plan_flags_section}"
-
+            summary = self.get_pr_environment_summary(conclusion, exception)
             self._append_output("pr_environment_name", self.pr_environment_name)
             return conclusion, check_title, summary