diff --git a/src/snowflake/cli/_plugins/nativeapp/manager.py b/src/snowflake/cli/_plugins/nativeapp/manager.py index f8d5f453d1..d911fdd2b3 100644 --- a/src/snowflake/cli/_plugins/nativeapp/manager.py +++ b/src/snowflake/cli/_plugins/nativeapp/manager.py @@ -20,7 +20,7 @@ from functools import cached_property from pathlib import Path from textwrap import dedent -from typing import Generator, List, Optional, TypedDict +from typing import Generator, List, Optional from snowflake.cli._plugins.connection.util import make_snowsight_url from snowflake.cli._plugins.nativeapp.artifacts import ( @@ -38,6 +38,7 @@ from snowflake.cli.api.console import cli_console as cc from snowflake.cli.api.entities.application_entity import ( ApplicationEntity, + ApplicationOwnedObject, ) from snowflake.cli.api.entities.application_package_entity import ( ApplicationPackageEntity, @@ -57,8 +58,6 @@ from snowflake.cli.api.sql_execution import SqlExecutionMixin from snowflake.connector import DictCursor, ProgrammingError -ApplicationOwnedObject = TypedDict("ApplicationOwnedObject", {"name": str, "type": str}) - class NativeAppCommandProcessor(ABC): @abstractmethod @@ -246,32 +245,19 @@ def get_existing_app_pkg_info(self) -> Optional[dict]: package_role=self.package_role, ) - def get_objects_owned_by_application(self) -> List[ApplicationOwnedObject]: - """ - Returns all application objects owned by this application. - """ - with self.use_role(self.app_role): - results = self._execute_query( - f"show objects owned by application {self.app_name}" - ).fetchall() - return [{"name": row[1], "type": row[2]} for row in results] + def get_objects_owned_by_application(self): + return ApplicationEntity.get_objects_owned_by_application( + app_name=self.app_name, + app_role=self.app_role, + ) def _application_objects_to_str( self, application_objects: list[ApplicationOwnedObject] ) -> str: - """ - Returns a list in an "(Object Type) Object Name" format. Database-level and schema-level object names are fully qualified: - (COMPUTE_POOL) POOL_NAME - (DATABASE) DB_NAME - (SCHEMA) DB_NAME.PUBLIC - ... - """ - return "\n".join( - [self._application_object_to_str(obj) for obj in application_objects] - ) + return ApplicationEntity.application_objects_to_str(application_objects) - def _application_object_to_str(self, obj: ApplicationOwnedObject) -> str: - return f"({obj['type']}) {obj['name']}" + def _application_object_to_str(self, obj: ApplicationOwnedObject): + return ApplicationEntity.application_object_to_str(obj) def get_snowsight_url(self) -> str: """Returns the URL that can be used to visit this app via Snowsight.""" diff --git a/src/snowflake/cli/_plugins/nativeapp/teardown_processor.py b/src/snowflake/cli/_plugins/nativeapp/teardown_processor.py index 74b168cf2f..139ca417db 100644 --- a/src/snowflake/cli/_plugins/nativeapp/teardown_processor.py +++ b/src/snowflake/cli/_plugins/nativeapp/teardown_processor.py @@ -15,179 +15,36 @@ from __future__ import annotations from pathlib import Path -from textwrap import dedent from typing import Dict, Optional -import typer -from snowflake.cli._plugins.nativeapp.constants import ( - ALLOWED_SPECIAL_COMMENTS, - COMMENT_COL, - OWNER_COL, -) from snowflake.cli._plugins.nativeapp.manager import ( NativeAppCommandProcessor, NativeAppManager, ) -from snowflake.cli._plugins.nativeapp.utils import ( - needs_confirmation, -) from snowflake.cli.api.console import cli_console as cc +from snowflake.cli.api.entities.application_entity import ( + ApplicationEntity, +) from snowflake.cli.api.entities.application_package_entity import ( ApplicationPackageEntity, ) -from snowflake.cli.api.entities.utils import ( - drop_generic_object, - ensure_correct_owner, -) -from snowflake.cli.api.errno import APPLICATION_NO_LONGER_AVAILABLE -from snowflake.connector import ProgrammingError class NativeAppTeardownProcessor(NativeAppManager, NativeAppCommandProcessor): def __init__(self, project_definition: Dict, project_root: Path): super().__init__(project_definition, project_root) - def drop_generic_object( - self, object_type: str, object_name: str, role: str, cascade: bool = False - ): - return drop_generic_object( - console=cc, - object_type=object_type, - object_name=object_name, - role=role, - cascade=cascade, - ) - def drop_application( self, auto_yes: bool, interactive: bool = False, cascade: Optional[bool] = None ): - """ - Attempts to drop the application object if all validations and user prompts allow so. - """ - - needs_confirm = True - - # 1. If existing application is not found, exit gracefully - show_obj_row = self.get_existing_app_info() - if show_obj_row is None: - cc.warning( - f"Role {self.app_role} does not own any application object with the name {self.app_name}, or the application object does not exist." - ) - return - - # 2. Check for the right owner - ensure_correct_owner( - row=show_obj_row, role=self.app_role, obj_name=self.app_name - ) - - # 3. Check if created by the Snowflake CLI - row_comment = show_obj_row[COMMENT_COL] - if row_comment not in ALLOWED_SPECIAL_COMMENTS and needs_confirmation( - needs_confirm, auto_yes - ): - should_drop_object = typer.confirm( - dedent( - f"""\ - Application object {self.app_name} was not created by Snowflake CLI. - Application object details: - Name: {self.app_name} - Created on: {show_obj_row["created_on"]} - Source: {show_obj_row["source"]} - Owner: {show_obj_row[OWNER_COL]} - Comment: {show_obj_row[COMMENT_COL]} - Version: {show_obj_row["version"]} - Patch: {show_obj_row["patch"]} - Are you sure you want to drop it? - """ - ) - ) - if not should_drop_object: - cc.message(f"Did not drop application object {self.app_name}.") - # The user desires to keep the app, therefore we can't proceed since it would - # leave behind an orphan app when we get to dropping the package - raise typer.Abort() - - # 4. Check for application objects owned by the application - # This query will fail if the application package has already been dropped, so handle this case gracefully - has_objects_to_drop = False - message_prefix = "" - cascade_true_message = "" - cascade_false_message = "" - interactive_prompt = "" - non_interactive_abort = "" - try: - if application_objects := self.get_objects_owned_by_application(): - has_objects_to_drop = True - message_prefix = ( - f"The following objects are owned by application {self.app_name}" - ) - cascade_true_message = f"{message_prefix} and will be dropped:" - cascade_false_message = f"{message_prefix} and will NOT be dropped:" - interactive_prompt = "Would you like to drop these objects in addition to the application? [y/n/ABORT]" - non_interactive_abort = "Re-run teardown again with --cascade or --no-cascade to specify whether these objects should be dropped along with the application" - except ProgrammingError as e: - if e.errno != APPLICATION_NO_LONGER_AVAILABLE: - raise - application_objects = [] - message_prefix = f"Could not determine which objects are owned by application {self.app_name}" - has_objects_to_drop = True # potentially, but we don't know what they are - cascade_true_message = ( - f"{message_prefix}, an unknown number of objects will be dropped." - ) - cascade_false_message = f"{message_prefix}, they will NOT be dropped." - interactive_prompt = f"Would you like to drop an unknown set of objects in addition to the application? [y/n/ABORT]" - non_interactive_abort = f"Re-run teardown again with --cascade or --no-cascade to specify whether any objects should be dropped along with the application." - - if has_objects_to_drop: - if cascade is True: - # If the user explicitly passed the --cascade flag - cc.message(cascade_true_message) - with cc.indented(): - for obj in application_objects: - cc.message(self._application_object_to_str(obj)) - elif cascade is False: - # If the user explicitly passed the --no-cascade flag - cc.message(cascade_false_message) - with cc.indented(): - for obj in application_objects: - cc.message(self._application_object_to_str(obj)) - elif interactive: - # If the user didn't pass any cascade flag and the session is interactive - cc.message(message_prefix) - with cc.indented(): - for obj in application_objects: - cc.message(self._application_object_to_str(obj)) - user_response = typer.prompt( - interactive_prompt, - show_default=False, - default="ABORT", - ).lower() - if user_response in ["y", "yes"]: - cascade = True - elif user_response in ["n", "no"]: - cascade = False - else: - raise typer.Abort() - else: - # Else abort since we don't know what to do and can't ask the user - cc.message(message_prefix) - with cc.indented(): - for obj in application_objects: - cc.message(self._application_object_to_str(obj)) - cc.message(non_interactive_abort) - raise typer.Abort() - elif cascade is None: - # If there's nothing to drop, set cascade to an explicit False value - cascade = False - - # 5. All validations have passed, drop object - self.drop_generic_object( - object_type="application", - object_name=self.app_name, - role=self.app_role, + return ApplicationEntity.drop( + console=cc, + app_name=self.app_name, + app_role=self.app_role, + auto_yes=auto_yes, + interactive=interactive, cascade=cascade, ) - return # The application object was successfully dropped, therefore exit gracefully def drop_package(self, auto_yes: bool): return ApplicationPackageEntity.drop( diff --git a/src/snowflake/cli/_plugins/workspace/commands.py b/src/snowflake/cli/_plugins/workspace/commands.py index 0fccb9a4f7..2b2625a0eb 100644 --- a/src/snowflake/cli/_plugins/workspace/commands.py +++ b/src/snowflake/cli/_plugins/workspace/commands.py @@ -178,7 +178,13 @@ def drop( help=f"""The ID of the entity you want to drop.""", ), # TODO The following options should be generated automatically, depending on the specified entity type + interactive: bool = InteractiveOption, force: Optional[bool] = ForceOption, + cascade: Optional[bool] = typer.Option( + None, + help=f"""Whether to drop all application objects owned by the application within the account. Default: false.""", + show_default=False, + ), **options, ): """ @@ -194,6 +200,8 @@ def drop( entity_id, EntityActions.DROP, force_drop=force, + interactive=interactive, + cascade=cascade, ) diff --git a/src/snowflake/cli/api/entities/application_entity.py b/src/snowflake/cli/api/entities/application_entity.py index b578ba1c65..302406a56b 100644 --- a/src/snowflake/cli/api/entities/application_entity.py +++ b/src/snowflake/cli/api/entities/application_entity.py @@ -1,8 +1,9 @@ from contextlib import contextmanager from pathlib import Path from textwrap import dedent -from typing import Callable, List, Optional +from typing import Callable, List, Optional, TypedDict +import typer from click import ClickException, UsageError from snowflake.cli._plugins.nativeapp.common_flags import ( ForceOption, @@ -10,7 +11,10 @@ ValidateOption, ) from snowflake.cli._plugins.nativeapp.constants import ( + ALLOWED_SPECIAL_COMMENTS, + COMMENT_COL, NAME_COL, + OWNER_COL, PATCH_COL, SPECIAL_COMMENT, VERSION_COL, @@ -27,6 +31,9 @@ from snowflake.cli._plugins.nativeapp.same_account_install_method import ( SameAccountInstallMethod, ) +from snowflake.cli._plugins.nativeapp.utils import ( + needs_confirmation, +) from snowflake.cli._plugins.workspace.action_context import ActionContext from snowflake.cli.api.console.abc import AbstractConsole from snowflake.cli.api.entities.application_package_entity import ( @@ -34,6 +41,8 @@ ) from snowflake.cli.api.entities.common import EntityBase, get_sql_executor from snowflake.cli.api.entities.utils import ( + drop_generic_object, + ensure_correct_owner, execute_post_deploy_hooks, generic_sql_error_handler, print_messages, @@ -71,6 +80,8 @@ APPLICATION_NO_LONGER_AVAILABLE, } +ApplicationOwnedObject = TypedDict("ApplicationOwnedObject", {"name": str, "type": str}) + class ApplicationEntity(EntityBase[ApplicationEntityModel]): """ @@ -159,6 +170,209 @@ def deploy_package(): deploy_package=deploy_package, ) + def action_drop( + self, + ctx: ActionContext, + interactive: bool, + force_drop: bool = False, + cascade: Optional[bool] = None, + *args, + **kwargs, + ): + model = self._entity_model + app_name = model.fqn.identifier + if model.meta and model.meta.role: + app_role = model.meta.role + else: + app_role = ctx.default_role + self.drop( + console=ctx.console, + app_name=app_name, + app_role=app_role, + auto_yes=force_drop, + interactive=interactive, + cascade=cascade, + ) + + @classmethod + def drop( + cls, + console: AbstractConsole, + app_name: str, + app_role: str, + auto_yes: bool, + interactive: bool = False, + cascade: Optional[bool] = None, + ): + """ + Attempts to drop the application object if all validations and user prompts allow so. + """ + + needs_confirm = True + + # 1. If existing application is not found, exit gracefully + show_obj_row = cls.get_existing_app_info( + app_name=app_name, + app_role=app_role, + ) + if show_obj_row is None: + console.warning( + f"Role {app_role} does not own any application object with the name {app_name}, or the application object does not exist." + ) + return + + # 2. Check for the right owner + ensure_correct_owner(row=show_obj_row, role=app_role, obj_name=app_name) + + # 3. Check if created by the Snowflake CLI + row_comment = show_obj_row[COMMENT_COL] + if row_comment not in ALLOWED_SPECIAL_COMMENTS and needs_confirmation( + needs_confirm, auto_yes + ): + should_drop_object = typer.confirm( + dedent( + f"""\ + Application object {app_name} was not created by Snowflake CLI. + Application object details: + Name: {app_name} + Created on: {show_obj_row["created_on"]} + Source: {show_obj_row["source"]} + Owner: {show_obj_row[OWNER_COL]} + Comment: {show_obj_row[COMMENT_COL]} + Version: {show_obj_row["version"]} + Patch: {show_obj_row["patch"]} + Are you sure you want to drop it? + """ + ) + ) + if not should_drop_object: + console.message(f"Did not drop application object {app_name}.") + # The user desires to keep the app, therefore we can't proceed since it would + # leave behind an orphan app when we get to dropping the package + raise typer.Abort() + + # 4. Check for application objects owned by the application + # This query will fail if the application package has already been dropped, so handle this case gracefully + has_objects_to_drop = False + message_prefix = "" + cascade_true_message = "" + cascade_false_message = "" + interactive_prompt = "" + non_interactive_abort = "" + try: + if application_objects := cls.get_objects_owned_by_application( + app_name=app_name, + app_role=app_role, + ): + has_objects_to_drop = True + message_prefix = ( + f"The following objects are owned by application {app_name}" + ) + cascade_true_message = f"{message_prefix} and will be dropped:" + cascade_false_message = f"{message_prefix} and will NOT be dropped:" + interactive_prompt = "Would you like to drop these objects in addition to the application? [y/n/ABORT]" + non_interactive_abort = "Re-run teardown again with --cascade or --no-cascade to specify whether these objects should be dropped along with the application" + except ProgrammingError as e: + if e.errno != APPLICATION_NO_LONGER_AVAILABLE: + raise + application_objects = [] + message_prefix = ( + f"Could not determine which objects are owned by application {app_name}" + ) + has_objects_to_drop = True # potentially, but we don't know what they are + cascade_true_message = ( + f"{message_prefix}, an unknown number of objects will be dropped." + ) + cascade_false_message = f"{message_prefix}, they will NOT be dropped." + interactive_prompt = f"Would you like to drop an unknown set of objects in addition to the application? [y/n/ABORT]" + non_interactive_abort = f"Re-run teardown again with --cascade or --no-cascade to specify whether any objects should be dropped along with the application." + + if has_objects_to_drop: + if cascade is True: + # If the user explicitly passed the --cascade flag + console.message(cascade_true_message) + with console.indented(): + for obj in application_objects: + console.message(cls.application_object_to_str(obj)) + elif cascade is False: + # If the user explicitly passed the --no-cascade flag + console.message(cascade_false_message) + with console.indented(): + for obj in application_objects: + console.message(cls.application_object_to_str(obj)) + elif interactive: + # If the user didn't pass any cascade flag and the session is interactive + console.message(message_prefix) + with console.indented(): + for obj in application_objects: + console.message(cls.application_object_to_str(obj)) + user_response = typer.prompt( + interactive_prompt, + show_default=False, + default="ABORT", + ).lower() + if user_response in ["y", "yes"]: + cascade = True + elif user_response in ["n", "no"]: + cascade = False + else: + raise typer.Abort() + else: + # Else abort since we don't know what to do and can't ask the user + console.message(message_prefix) + with console.indented(): + for obj in application_objects: + console.message(cls.application_object_to_str(obj)) + console.message(non_interactive_abort) + raise typer.Abort() + elif cascade is None: + # If there's nothing to drop, set cascade to an explicit False value + cascade = False + + # 5. All validations have passed, drop object + drop_generic_object( + console=console, + object_type="application", + object_name=app_name, + role=app_role, + cascade=cascade, + ) + return # The application object was successfully dropped, therefore exit gracefully + + @staticmethod + def get_objects_owned_by_application( + app_name: str, + app_role: str, + ) -> List[ApplicationOwnedObject]: + """ + Returns all application objects owned by this application. + """ + sql_executor = get_sql_executor() + with sql_executor.use_role(app_role): + results = sql_executor.execute_query( + f"show objects owned by application {app_name}" + ).fetchall() + return [{"name": row[1], "type": row[2]} for row in results] + + @classmethod + def application_objects_to_str( + cls, application_objects: list[ApplicationOwnedObject] + ) -> str: + """ + Returns a list in an "(Object Type) Object Name" format. Database-level and schema-level object names are fully qualified: + (COMPUTE_POOL) POOL_NAME + (DATABASE) DB_NAME + (SCHEMA) DB_NAME.PUBLIC + ... + """ + return "\n".join( + [cls.application_object_to_str(obj) for obj in application_objects] + ) + + @staticmethod + def application_object_to_str(obj: ApplicationOwnedObject) -> str: + return f"({obj['type']}) {obj['name']}" + @classmethod def deploy( cls, diff --git a/tests/nativeapp/test_teardown_processor.py b/tests/nativeapp/test_teardown_processor.py index 437612c133..ee1aeb8a22 100644 --- a/tests/nativeapp/test_teardown_processor.py +++ b/tests/nativeapp/test_teardown_processor.py @@ -29,6 +29,7 @@ from snowflake.cli._plugins.nativeapp.teardown_processor import ( NativeAppTeardownProcessor, ) +from snowflake.cli.api.entities.utils import drop_generic_object from snowflake.cli.api.errno import ( APPLICATION_NO_LONGER_AVAILABLE, DOES_NOT_EXIST_OR_CANNOT_BE_PERFORMED, @@ -40,17 +41,17 @@ from tests.nativeapp.patch_utils import mock_get_app_pkg_distribution_in_sf from tests.nativeapp.utils import ( + APP_ENTITY_DROP_GENERIC_OBJECT, + APP_ENTITY_GET_EXISTING_APP_INFO, + APP_ENTITY_GET_OBJECTS_OWNED_BY_APPLICATION, + APP_ENTITY_IS_CORRECT_OWNER, + APP_ENTITY_MODULE, APP_PACKAGE_ENTITY_DROP_GENERIC_OBJECT, APP_PACKAGE_ENTITY_GET_EXISTING_APP_PKG_INFO, APP_PACKAGE_ENTITY_IS_CORRECT_OWNER, APP_PACKAGE_ENTITY_IS_DISTRIBUTION_SAME, - NATIVEAPP_MANAGER_EXECUTE, - NATIVEAPP_MANAGER_GET_OBJECTS_OWNED_BY_APPLICATION, SQL_EXECUTOR_EXECUTE, TEARDOWN_MODULE, - TEARDOWN_PROCESSOR_DROP_GENERIC_OBJECT, - TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO, - TEARDOWN_PROCESSOR_IS_CORRECT_OWNER, TYPER_CONFIRM, TYPER_PROMPT, mock_execute_helper, @@ -91,9 +92,11 @@ def test_drop_generic_object_success(mock_execute, temp_dir, mock_cursor): contents=[mock_snowflake_yml_file], ) - teardown_processor = _get_na_teardown_processor() - teardown_processor.drop_generic_object( - object_type="application", object_name="myapp", role="app_role" + drop_generic_object( + console=mock.Mock(), + object_type="application", + object_name="myapp", + role="app_role", ) assert mock_execute.mock_calls == expected @@ -131,9 +134,9 @@ def test_drop_generic_object_failure_w_exception( contents=[mock_snowflake_yml_file], ) - teardown_processor = _get_na_teardown_processor() with pytest.raises(SnowflakeSQLExecutionError): - teardown_processor.drop_generic_object( + drop_generic_object( + console=mock.Mock(), object_type="application package", object_name="app_pkg", role="package_role", @@ -142,7 +145,7 @@ def test_drop_generic_object_failure_w_exception( # Test drop_application() when no application exists -@mock.patch(TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO, return_value=None) +@mock.patch(APP_ENTITY_GET_EXISTING_APP_INFO, return_value=None) @mock.patch(f"{TEARDOWN_MODULE}.cc.warning") @pytest.mark.parametrize( "auto_yes_param", @@ -167,7 +170,7 @@ def test_drop_application_no_existing_application( # Test drop_application() when it has a different owner role -@mock.patch(TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO) +@mock.patch(APP_ENTITY_GET_EXISTING_APP_INFO) @pytest.mark.parametrize( "auto_yes_param", [True, False], # This should have no effect on the test @@ -195,10 +198,10 @@ def test_drop_application_incorrect_owner( # Test drop_application() successfully when it has special comment -@mock.patch(TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO) -@mock.patch(TEARDOWN_PROCESSOR_IS_CORRECT_OWNER, return_value=True) -@mock.patch(TEARDOWN_PROCESSOR_DROP_GENERIC_OBJECT, return_value=None) -@mock.patch(NATIVEAPP_MANAGER_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) +@mock.patch(APP_ENTITY_GET_EXISTING_APP_INFO) +@mock.patch(APP_ENTITY_IS_CORRECT_OWNER, return_value=True) +@mock.patch(APP_ENTITY_DROP_GENERIC_OBJECT, return_value=None) +@mock.patch(APP_ENTITY_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) @pytest.mark.parametrize( "auto_yes_param, special_comment", # auto_yes should have no effect on the test [ @@ -239,7 +242,7 @@ def test_drop_application_has_special_comment( # Test drop_application() successfully when it has special comment but is a quoted string @mock.patch(SQL_EXECUTOR_EXECUTE) -@mock.patch(NATIVEAPP_MANAGER_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) +@mock.patch(APP_ENTITY_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) @pytest.mark.parametrize( "auto_yes_param, special_comment", # auto_yes should have no effect on the test [ @@ -312,10 +315,10 @@ def test_drop_application_has_special_comment_and_quoted_name( # Test drop_application() without special comment AND auto_yes is False AND should_drop is False -@mock.patch(TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO) -@mock.patch(TEARDOWN_PROCESSOR_IS_CORRECT_OWNER, return_value=True) -@mock.patch(TEARDOWN_PROCESSOR_DROP_GENERIC_OBJECT, return_value=None) -@mock.patch(f"{TEARDOWN_MODULE}.{TYPER_CONFIRM}", return_value=False) +@mock.patch(APP_ENTITY_GET_EXISTING_APP_INFO) +@mock.patch(APP_ENTITY_IS_CORRECT_OWNER, return_value=True) +@mock.patch(APP_ENTITY_DROP_GENERIC_OBJECT, return_value=None) +@mock.patch(f"{APP_ENTITY_MODULE}.{TYPER_CONFIRM}", return_value=False) @mock.patch(f"{TEARDOWN_MODULE}.cc.message") def test_drop_application_user_prohibits_drop( mock_message, @@ -353,12 +356,12 @@ def test_drop_application_user_prohibits_drop( # Test drop_application() without special comment AND auto_yes is False AND should_drop is True # Test drop_application() without special comment AND auto_yes is True -@mock.patch(TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO) -@mock.patch(TEARDOWN_PROCESSOR_IS_CORRECT_OWNER, return_value=True) -@mock.patch(TEARDOWN_PROCESSOR_DROP_GENERIC_OBJECT, return_value=None) -@mock.patch(NATIVEAPP_MANAGER_EXECUTE) -@mock.patch(f"{TEARDOWN_MODULE}.{TYPER_CONFIRM}", return_value=True) -@mock.patch(NATIVEAPP_MANAGER_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) +@mock.patch(APP_ENTITY_GET_EXISTING_APP_INFO) +@mock.patch(APP_ENTITY_IS_CORRECT_OWNER, return_value=True) +@mock.patch(APP_ENTITY_DROP_GENERIC_OBJECT, return_value=None) +@mock.patch(SQL_EXECUTOR_EXECUTE) +@mock.patch(f"{APP_ENTITY_MODULE}.{TYPER_CONFIRM}", return_value=True) +@mock.patch(APP_ENTITY_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) @pytest.mark.parametrize( "auto_yes_param", [False, True], @@ -411,10 +414,10 @@ def test_drop_application_user_allows_drop( # Test idempotent drop_application() -@mock.patch(TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO) -@mock.patch(TEARDOWN_PROCESSOR_IS_CORRECT_OWNER, return_value=True) -@mock.patch(TEARDOWN_PROCESSOR_DROP_GENERIC_OBJECT, return_value=None) -@mock.patch(NATIVEAPP_MANAGER_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) +@mock.patch(APP_ENTITY_GET_EXISTING_APP_INFO) +@mock.patch(APP_ENTITY_IS_CORRECT_OWNER, return_value=True) +@mock.patch(APP_ENTITY_DROP_GENERIC_OBJECT, return_value=None) +@mock.patch(APP_ENTITY_GET_OBJECTS_OWNED_BY_APPLICATION, return_value=[]) @pytest.mark.parametrize( "auto_yes_param", [False, True], # This should have no effect on the test @@ -560,7 +563,7 @@ def test_show_versions_failure_w_exception( @mock.patch(SQL_EXECUTOR_EXECUTE) @mock_get_app_pkg_distribution_in_sf() @mock.patch(APP_PACKAGE_ENTITY_IS_DISTRIBUTION_SAME, return_value=True) -@mock.patch(f"{TEARDOWN_MODULE}.{TYPER_CONFIRM}", return_value=False) +@mock.patch(f"{APP_ENTITY_MODULE}.{TYPER_CONFIRM}", return_value=False) def test_drop_package_no_mismatch_no_drop( mock_confirm, mock_is_distribution_same, @@ -617,7 +620,7 @@ def test_drop_package_no_mismatch_no_drop( @mock.patch(f"{TEARDOWN_MODULE}.cc.warning") @mock_get_app_pkg_distribution_in_sf() @mock.patch(APP_PACKAGE_ENTITY_IS_DISTRIBUTION_SAME) -@mock.patch(f"{TEARDOWN_MODULE}.{TYPER_CONFIRM}", return_value=True) +@mock.patch(f"{APP_ENTITY_MODULE}.{TYPER_CONFIRM}", return_value=True) @mock.patch(APP_PACKAGE_ENTITY_DROP_GENERIC_OBJECT, return_value=None) @pytest.mark.parametrize( "auto_yes_param, is_pkg_distribution_same", @@ -849,7 +852,7 @@ def test_drop_package_variable_mistmatch_w_special_comment_quoted_name_auto_drop @mock.patch(SQL_EXECUTOR_EXECUTE) @mock_get_app_pkg_distribution_in_sf() @mock.patch(APP_PACKAGE_ENTITY_IS_DISTRIBUTION_SAME) -@mock.patch(f"{TEARDOWN_MODULE}.{TYPER_CONFIRM}", return_value=False) +@mock.patch(f"{APP_ENTITY_MODULE}.{TYPER_CONFIRM}", return_value=False) @mock.patch(f"{TEARDOWN_MODULE}.cc.warning") @pytest.mark.parametrize("is_pkg_distribution_same", [True, False]) def test_drop_package_variable_mistmatch_no_special_comment_user_prohibits_drop( @@ -915,7 +918,7 @@ def test_drop_package_variable_mistmatch_no_special_comment_user_prohibits_drop( @mock.patch(SQL_EXECUTOR_EXECUTE) @mock_get_app_pkg_distribution_in_sf() @mock.patch(APP_PACKAGE_ENTITY_IS_DISTRIBUTION_SAME) -@mock.patch(f"{TEARDOWN_MODULE}.{TYPER_CONFIRM}", return_value=True) +@mock.patch(f"{APP_ENTITY_MODULE}.{TYPER_CONFIRM}", return_value=True) @mock.patch(APP_PACKAGE_ENTITY_DROP_GENERIC_OBJECT, return_value=None) @pytest.mark.parametrize( "auto_yes_param, is_pkg_distribution_same", # auto_yes_param should have no effect on the test @@ -1039,11 +1042,11 @@ def test_drop_package_idempotent( mock_execute.mock_calls == expected -@mock.patch(f"{TEARDOWN_MODULE}.{TYPER_PROMPT}") -@mock.patch(TEARDOWN_PROCESSOR_GET_EXISTING_APP_INFO) -@mock.patch(TEARDOWN_PROCESSOR_IS_CORRECT_OWNER, return_value=True) -@mock.patch(TEARDOWN_PROCESSOR_DROP_GENERIC_OBJECT, return_value=None) -@mock.patch(NATIVEAPP_MANAGER_GET_OBJECTS_OWNED_BY_APPLICATION) +@mock.patch(f"{APP_ENTITY_MODULE}.{TYPER_PROMPT}") +@mock.patch(APP_ENTITY_GET_EXISTING_APP_INFO) +@mock.patch(APP_ENTITY_IS_CORRECT_OWNER, return_value=True) +@mock.patch(APP_ENTITY_DROP_GENERIC_OBJECT, return_value=None) +@mock.patch(APP_ENTITY_GET_OBJECTS_OWNED_BY_APPLICATION) @pytest.mark.parametrize( "cascade,application_objects,interactive_response,expected_cascade", [ @@ -1107,6 +1110,7 @@ def test_drop_application_cascade( else: teardown_processor.drop_application(False, interactive, cascade) mock_drop_generic_object.assert_called_once_with( + console=mock.ANY, object_type="application", object_name="myapp", role="app_role", diff --git a/tests/nativeapp/utils.py b/tests/nativeapp/utils.py index cad8be3a3a..eb6431276d 100644 --- a/tests/nativeapp/utils.py +++ b/tests/nativeapp/utils.py @@ -67,17 +67,21 @@ TEARDOWN_PROCESSOR_GET_EXISTING_APP_PKG_INFO = ( f"{TEARDOWN_PROCESSOR}.get_existing_app_pkg_info" ) -TEARDOWN_PROCESSOR_IS_CORRECT_OWNER = f"{TEARDOWN_MODULE}.ensure_correct_owner" -TEARDOWN_PROCESSOR_DROP_GENERIC_OBJECT = f"{TEARDOWN_PROCESSOR}.drop_generic_object" RUN_PROCESSOR_GET_EXISTING_APP_INFO = f"{RUN_PROCESSOR}.get_existing_app_info" RUN_PROCESSOR_APP_POST_DEPLOY_HOOKS = f"{RUN_PROCESSOR}.app_post_deploy_hooks" FIND_VERSION_FROM_MANIFEST = f"{VERSION_MODULE}.find_version_info_in_manifest_file" -APP_ENTITY = "snowflake.cli.api.entities.application_entity.ApplicationEntity" +APP_ENTITY_MODULE = "snowflake.cli.api.entities.application_entity" +APP_ENTITY = f"{APP_ENTITY_MODULE}.ApplicationEntity" APP_ENTITY_GET_EXISTING_APP_INFO = f"{APP_ENTITY}.get_existing_app_info" APP_ENTITY_GET_EXISTING_VERSION_INFO = f"{APP_ENTITY}.get_existing_version_info" +APP_ENTITY_DROP_GENERIC_OBJECT = f"{APP_ENTITY_MODULE}.drop_generic_object" +APP_ENTITY_IS_CORRECT_OWNER = f"{APP_ENTITY_MODULE}.ensure_correct_owner" +APP_ENTITY_GET_OBJECTS_OWNED_BY_APPLICATION = ( + f"{APP_ENTITY}.get_objects_owned_by_application" +) APP_PACKAGE_ENTITY = ( "snowflake.cli.api.entities.application_package_entity.ApplicationPackageEntity" diff --git a/tests_integration/nativeapp/test_teardown.py b/tests_integration/nativeapp/test_teardown.py index b1d392aa00..dbc65c2cc8 100644 --- a/tests_integration/nativeapp/test_teardown.py +++ b/tests_integration/nativeapp/test_teardown.py @@ -22,27 +22,65 @@ @pytest.mark.integration +@pytest.mark.parametrize("orphan_app", [True, False]) @pytest.mark.parametrize( - "command,expected_error", + "test_project,command,expected_error", [ - # "snow app teardown --cascade" should drop both application and application objects - ["app teardown --cascade", None], - # "snow app teardown --force --no-cascade" should attempt to drop the application and fail + # "--cascade" should drop both application and application objects + [ + "napp_create_db_v1", + "app teardown --cascade", + None, + ], + [ + "napp_create_db_v2", + "app teardown --cascade", + None, + ], [ + "napp_create_db_v2", + "ws drop --entity-id=app --cascade", + None, + ], + # "--force --no-cascade" should attempt to drop the application and fail + [ + "napp_create_db_v1", "app teardown --force --no-cascade", "Could not successfully execute the Snowflake SQL statements", ], - # "snow app teardown" with owned application objects should abort the teardown - ["app teardown", "Aborted"], + [ + "napp_create_db_v2", + "app teardown --force --no-cascade", + "Could not successfully execute the Snowflake SQL statements", + ], + [ + "napp_create_db_v2", + "ws drop --entity-id=app --force --no-cascade", + "Could not successfully execute the Snowflake SQL statements", + ], + # teardown/drop with owned application objects should abort the teardown + [ + "napp_create_db_v1", + "app teardown", + "Aborted", + ], + [ + "napp_create_db_v2", + "app teardown", + "Aborted", + ], + [ + "napp_create_db_v2", + "ws drop --entity-id=app", + "Aborted", + ], ], ) -@pytest.mark.parametrize("orphan_app", [True, False]) -@pytest.mark.parametrize("test_project", ["napp_create_db_v1", "napp_create_db_v2"]) def test_nativeapp_teardown_cascade( - command, - expected_error, orphan_app, test_project, + command, + expected_error, nativeapp_project_directory, runner, snowflake_session, @@ -131,12 +169,20 @@ def test_nativeapp_teardown_cascade( @pytest.mark.integration @pytest.mark.parametrize("force", [True, False]) -@pytest.mark.parametrize("test_project", ["napp_init_v1", "napp_init_v2"]) +@pytest.mark.parametrize( + "command,test_project", + [ + ["app teardown", "napp_init_v1"], + ["app teardown", "napp_init_v2"], + ["ws drop --entity-id=app", "napp_init_v2"], + ], +) def test_nativeapp_teardown_unowned_app( runner, default_username, resource_suffix, force, + command, test_project, nativeapp_project_directory, ): @@ -152,10 +198,10 @@ def test_nativeapp_teardown_unowned_app( assert result.exit_code == 0 if force: - result = runner.invoke_with_connection_json(["app", "teardown", "--force"]) + result = runner.invoke_with_connection_json([*split(command), "--force"]) assert result.exit_code == 0 else: - result = runner.invoke_with_connection_json(["app", "teardown"]) + result = runner.invoke_with_connection_json(split(command)) assert result.exit_code == 1