extract validate logic to entity

src/snowflake/cli/_plugins/nativeapp/manager.py

@@ -39,7 +39,6 @@
 from snowflake.cli._plugins.nativeapp.exceptions import (
     ApplicationPackageDoesNotExistError,
     NoEventTableForAccount,
-    SetupScriptFailedValidation,
 )
 from snowflake.cli._plugins.nativeapp.project_model import (
     NativeAppProjectModel,
@@ -382,23 +381,26 @@ def deploy(
         return diff
 
     def validate(self, use_scratch_stage: bool = False):
-        """Validates Native App setup script SQL."""
-        with cc.phase(f"Validating Snowflake Native App setup script."):
-            validation_result = self.get_validation_result(use_scratch_stage)
-
-            # First print warnings, regardless of the outcome of validation
-            for warning in validation_result.get("warnings", []):
-                cc.warning(_validation_item_to_str(warning))
-
-            # Then print errors
-            for error in validation_result.get("errors", []):
-                # Print them as warnings for now since we're going to be
-                # revamping CLI output soon
-                cc.warning(_validation_item_to_str(error))
-
-            # Then raise an exception if validation failed
-            if validation_result["status"] == "FAIL":
-                raise SetupScriptFailedValidation()
+        def deploy_to_scratch_stage_fn():
+            bundle_map = self.build_bundle()
+            self.deploy(
+                bundle_map=bundle_map,
+                prune=True,
+                recursive=True,
+                stage_fqn=self.scratch_stage_fqn,
+                validate=False,
+                print_diff=False,
+            )
+
+        return ApplicationPackageEntity.validate(
+            console=cc,
+            package_name=self.package_name,
+            package_role=self.package_role,
+            stage_fqn=self.stage_fqn,
+            use_scratch_stage=use_scratch_stage,
+            scratch_stage_fqn=self.scratch_stage_fqn,
+            deploy_to_scratch_stage_fn=deploy_to_scratch_stage_fn,
+        )
 
     def get_validation_result(self, use_scratch_stage: bool):
         """Call system$validate_native_app_setup() to validate deployed Native App setup script."""

src/snowflake/cli/api/entities/application_package_entity.py

@@ -1,7 +1,8 @@
+import json
 from contextlib import contextmanager
 from pathlib import Path
 from textwrap import dedent
-from typing import List, Optional
+from typing import Callable, List, Optional
 
 from click import ClickException
 from snowflake.cli._plugins.nativeapp.artifacts import build_bundle
@@ -16,7 +17,10 @@
 )
 from snowflake.cli._plugins.nativeapp.exceptions import (
     ApplicationPackageAlreadyExistsError,
+    ApplicationPackageDoesNotExistError,
+    SetupScriptFailedValidation,
 )
+from snowflake.cli._plugins.stage.manager import StageManager
 from snowflake.cli._plugins.workspace.action_context import ActionContext
 from snowflake.cli.api.console.abc import AbstractConsole
 from snowflake.cli.api.entities.common import EntityBase, get_sql_executor
@@ -25,6 +29,10 @@
     generic_sql_error_handler,
     render_script_templates,
     sync_deploy_root_with_stage,
+    validation_item_to_str,
+)
+from snowflake.cli.api.errno import (
+    DOES_NOT_EXIST_OR_NOT_AUTHORIZED,
 )
 from snowflake.cli.api.exceptions import SnowflakeSQLExecutionError
 from snowflake.cli.api.project.schemas.entities.application_package_entity_model import (
@@ -59,6 +67,11 @@ def action_bundle(self, ctx: ActionContext):
         compiler.compile_artifacts()
         return bundle_map
 
+    def action_validate(self, ctx: ActionContext):
+        model = self._entity_model
+        model.artifacts
+        pass
+
     def action_deploy(
         self,
         ctx: ActionContext,
@@ -306,3 +319,76 @@ def create_app_package(
                 """
                 )
             )
+
+    @classmethod
+    def validate(
+        cls,
+        console: AbstractConsole,
+        package_name: str,
+        package_role: str,
+        stage_fqn: str,
+        use_scratch_stage: bool,
+        scratch_stage_fqn: str,
+        deploy_to_scratch_stage_fn: Callable,
+    ):
+        """Validates Native App setup script SQL."""
+        with console.phase(f"Validating Snowflake Native App setup script."):
+            validation_result = cls.get_validation_result(
+                console=console,
+                package_name=package_name,
+                package_role=package_role,
+                stage_fqn=stage_fqn,
+                use_scratch_stage=use_scratch_stage,
+                scratch_stage_fqn=scratch_stage_fqn,
+                deploy_to_scratch_stage_fn=deploy_to_scratch_stage_fn,
+            )
+
+            # First print warnings, regardless of the outcome of validation
+            for warning in validation_result.get("warnings", []):
+                console.warning(validation_item_to_str(warning))
+
+            # Then print errors
+            for error in validation_result.get("errors", []):
+                # Print them as warnings for now since we're going to be
+                # revamping CLI output soon
+                console.warning(validation_item_to_str(error))
+
+            # Then raise an exception if validation failed
+            if validation_result["status"] == "FAIL":
+                raise SetupScriptFailedValidation()
+
+    @staticmethod
+    def get_validation_result(
+        console: AbstractConsole,
+        package_name: str,
+        package_role: str,
+        stage_fqn: str,
+        use_scratch_stage: bool,
+        scratch_stage_fqn: str,
+        deploy_to_scratch_stage_fn: Callable,
+    ):
+        """Call system$validate_native_app_setup() to validate deployed Native App setup script."""
+        if use_scratch_stage:
+            stage_fqn = scratch_stage_fqn
+            deploy_to_scratch_stage_fn()
+        prefixed_stage_fqn = StageManager.get_standard_stage_prefix(stage_fqn)
+        sql_executor = get_sql_executor()
+        try:
+            cursor = sql_executor.execute_query(
+                f"call system$validate_native_app_setup('{prefixed_stage_fqn}')"
+            )
+        except ProgrammingError as err:
+            if err.errno == DOES_NOT_EXIST_OR_NOT_AUTHORIZED:
+                raise ApplicationPackageDoesNotExistError(package_name)
+            generic_sql_error_handler(err)
+        else:
+            if not cursor.rowcount:
+                raise SnowflakeSQLExecutionError()
+            return json.loads(cursor.fetchone()[0])
+        finally:
+            if use_scratch_stage:
+                console.step(f"Dropping stage {scratch_stage_fqn}.")
+                with sql_executor.use_role(package_role):
+                    sql_executor.execute_query(
+                        f"drop stage if exists {scratch_stage_fqn}"
+                    )

src/snowflake/cli/api/entities/utils.py

@@ -319,3 +319,10 @@ def render_script_templates(
             raise InvalidScriptError(relpath, e) from e
 
     return scripts_contents
+
+
+def validation_item_to_str(item: dict[str, str | int]):
+    s = item["message"]
+    if item["errorCode"]:
+        s = f"{s} (error code {item['errorCode']})"
+    return s