From 96509d094fba06ad5eb91d5c3b73432e563102cf Mon Sep 17 00:00:00 2001
From: Francois Campbell <francois.campbell@snowflake.com>
Date: Thu, 19 Sep 2024 17:18:16 -0400
Subject: [PATCH] SNOW-1653909 If we can't find an app entity to convert, only
 fail if it's required, update `snow app teardown` to tear down multiple apps
 (#1557)

Followup to #1546 to fix a bug where it would complain if we had a PDv2 with multiple app entities for commands that didn't care about the app at all.

Also updates `snow app teardown` to me PDFv2-aware so we can teardown all the apps created from one package.
---
 .../cli/_plugins/nativeapp/commands.py        |  86 ++++++--
 .../v2_conversions/v2_to_v1_decorator.py      | 198 +++++++++++-------
 .../_plugins/nativeapp/version/commands.py    |   6 +-
 tests/__snapshots__/test_help_messages.ambr   |   5 -
 tests/nativeapp/test_v2_to_v1.py              | 112 ++++++++--
 .../integration_templated_v2/snowflake.yml    |   5 +
 .../projects/integration_v2/snowflake.yml     |   5 +
 tests_integration/nativeapp/conftest.py       |  27 ++-
 tests_integration/nativeapp/test_init_run.py  |   9 +-
 tests_integration/nativeapp/test_teardown.py  | 121 +++++++++++
 10 files changed, 445 insertions(+), 129 deletions(-)

diff --git a/src/snowflake/cli/_plugins/nativeapp/commands.py b/src/snowflake/cli/_plugins/nativeapp/commands.py
index 8c1430ebed..ba7a888678 100644
--- a/src/snowflake/cli/_plugins/nativeapp/commands.py
+++ b/src/snowflake/cli/_plugins/nativeapp/commands.py
@@ -22,6 +22,12 @@
 from typing import Generator, Iterable, List, Optional, cast
 
 import typer
+from snowflake.cli._plugins.nativeapp.application_entity_model import (
+    ApplicationEntityModel,
+)
+from snowflake.cli._plugins.nativeapp.application_package_entity_model import (
+    ApplicationPackageEntityModel,
+)
 from snowflake.cli._plugins.nativeapp.common_flags import (
     ForceOption,
     InteractiveOption,
@@ -46,6 +52,7 @@
     shallow_git_clone,
 )
 from snowflake.cli._plugins.nativeapp.v2_conversions.v2_to_v1_decorator import (
+    find_entity,
     nativeapp_definition_v2_to_v1,
 )
 from snowflake.cli._plugins.nativeapp.version.commands import app as versions_app
@@ -54,11 +61,13 @@
     compute_stage_diff,
 )
 from snowflake.cli._plugins.stage.utils import print_diff_to_console
+from snowflake.cli._plugins.workspace.manager import WorkspaceManager
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.commands.decorators import (
     with_project_definition,
 )
 from snowflake.cli.api.commands.snow_typer import SnowTyperFactory
+from snowflake.cli.api.entities.common import EntityActions
 from snowflake.cli.api.exceptions import IncompatibleParametersError
 from snowflake.cli.api.output.formats import OutputFormat
 from snowflake.cli.api.output.types import (
@@ -69,6 +78,7 @@
     StreamResult,
 )
 from snowflake.cli.api.project.project_verification import assert_project_type
+from snowflake.cli.api.project.schemas.project_definition import ProjectDefinitionV1
 from snowflake.cli.api.secure_path import SecurePath
 from typing_extensions import Annotated
 
@@ -160,7 +170,7 @@ def app_list_templates(**options) -> CommandResult:
 
 @app.command("bundle")
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1()
 def app_bundle(
     **options,
 ) -> CommandResult:
@@ -181,7 +191,7 @@ def app_bundle(
 
 @app.command("diff", requires_connection=True, hidden=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1()
 def app_diff(
     **options,
 ) -> CommandResult:
@@ -208,7 +218,7 @@ def app_diff(
 
 @app.command("run", requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1(app_required=True)
 def app_run(
     version: Optional[str] = typer.Option(
         None,
@@ -272,7 +282,7 @@ def app_run(
 
 @app.command("open", requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1(app_required=True)
 def app_open(
     **options,
 ) -> CommandResult:
@@ -299,7 +309,9 @@ def app_open(
 
 @app.command("teardown", requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+# This command doesn't use @nativeapp_definition_v2_to_v1 because it needs to
+# be aware of PDFv2 definitions that have multiple apps created from the same package,
+# which all need to be torn down.
 def app_teardown(
     force: Optional[bool] = ForceOption,
     cascade: Optional[bool] = typer.Option(
@@ -308,26 +320,70 @@ def app_teardown(
         show_default=False,
     ),
     interactive: bool = InteractiveOption,
+    # Same as the param auto-added by @nativeapp_definition_v2_to_v1
+    package_entity_id: Optional[str] = typer.Option(
+        default="",
+        help="The ID of the package entity on which to operate when definition_version is 2 or higher.",
+    ),
     **options,
 ) -> CommandResult:
     """
     Attempts to drop both the application object and application package as defined in the project definition file.
     """
+    cli_context = get_cli_context()
+    project = cli_context.project_definition
+    if isinstance(project, ProjectDefinitionV1):
+        # Old behaviour, not multi-app aware so we can use the old processor
+        processor = NativeAppTeardownProcessor(
+            project_definition=cli_context.project_definition.native_app,
+            project_root=cli_context.project_root,
+        )
+        processor.process(interactive, force, cascade)
+    else:
+        # New behaviour, multi-app aware so teardown all the apps created from the package
+
+        # Determine the package entity to drop, there must be one
+        app_package_entity = find_entity(
+            project,
+            ApplicationPackageEntityModel,
+            package_entity_id,
+            disambiguation_option="--package-entity-id",
+            required=True,
+        )
+        assert app_package_entity is not None  # satisfy mypy
 
-    assert_project_type("native_app")
+        # Same implementation as `snow ws drop`
+        ws = WorkspaceManager(
+            project_definition=cli_context.project_definition,
+            project_root=cli_context.project_root,
+        )
+        for app_entity in project.get_entities_by_type(
+            ApplicationEntityModel.get_type()
+        ).values():
+            # Drop each app
+            if app_entity.from_.target == app_package_entity.entity_id:
+                ws.perform_action(
+                    app_entity.entity_id,
+                    EntityActions.DROP,
+                    force_drop=force,
+                    interactive=interactive,
+                    cascade=cascade,
+                )
+        # Then drop the package
+        ws.perform_action(
+            app_package_entity.entity_id,
+            EntityActions.DROP,
+            force_drop=force,
+            interactive=interactive,
+            cascade=cascade,
+        )
 
-    cli_context = get_cli_context()
-    processor = NativeAppTeardownProcessor(
-        project_definition=cli_context.project_definition.native_app,
-        project_root=cli_context.project_root,
-    )
-    processor.process(interactive, force, cascade)
     return MessageResult(f"Teardown is now complete.")
 
 
 @app.command("deploy", requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1()
 def app_deploy(
     prune: Optional[bool] = typer.Option(
         default=None,
@@ -395,7 +451,7 @@ def app_deploy(
 
 @app.command("validate", requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1()
 def app_validate(**options):
     """
     Validates a deployed Snowflake Native App's setup script.
@@ -428,7 +484,7 @@ class RecordType(Enum):
 
 @app.command("events", requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1(app_required=True)
 def app_events(
     since: str = typer.Option(
         default="",
diff --git a/src/snowflake/cli/_plugins/nativeapp/v2_conversions/v2_to_v1_decorator.py b/src/snowflake/cli/_plugins/nativeapp/v2_conversions/v2_to_v1_decorator.py
index 3429a8e69c..ef4ea739c3 100644
--- a/src/snowflake/cli/_plugins/nativeapp/v2_conversions/v2_to_v1_decorator.py
+++ b/src/snowflake/cli/_plugins/nativeapp/v2_conversions/v2_to_v1_decorator.py
@@ -16,7 +16,7 @@
 
 import inspect
 from functools import wraps
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, Optional, Type, TypeVar, Union
 
 import typer
 from click import ClickException
@@ -31,6 +31,7 @@
     get_cli_context_manager,
 )
 from snowflake.cli.api.commands.decorators import _options_decorator_factory
+from snowflake.cli.api.project.schemas.entities.common import EntityModelBase
 from snowflake.cli.api.project.schemas.project_definition import (
     DefinitionV11,
     DefinitionV20,
@@ -55,33 +56,18 @@ def _pdf_v2_to_v1(
     v2_definition: DefinitionV20,
     package_entity_id: str = "",
     app_entity_id: str = "",
+    app_required: bool = False,
 ) -> DefinitionV11:
     pdfv1: Dict[str, Any] = {"definition_version": "1.1", "native_app": {}}
 
-    app_package_definition: Optional[ApplicationPackageEntityModel] = None
-    app_definition: Optional[ApplicationEntityModel] = None
-
-    # Enumerate all application package and application entities in the project definition
-    packages: dict[
-        str, ApplicationPackageEntityModel
-    ] = v2_definition.get_entities_by_type(ApplicationPackageEntityModel.get_type())
-    apps: dict[str, ApplicationEntityModel] = v2_definition.get_entities_by_type(
-        ApplicationEntityModel.get_type()
-    )
-
     # Determine the application entity to convert, there can be zero or one
-    if app_entity_id:
-        # If the user specified an app entity ID, use that one directly
-        app_definition = apps.get(app_entity_id)
-    elif len(apps) == 1:
-        # Otherwise, if there is only one app entity, fall back to that one
-        app_definition = next(iter(apps.values()))
-    elif len(apps) > 1:
-        # If there are multiple app entities, the user must specify which one to use
-        raise ClickException(
-            "More than one application entity exists in the project definition file, "
-            "specify --app-entity-id to choose which one to operate on."
-        )
+    app_definition = find_entity(
+        v2_definition,
+        ApplicationEntityModel,
+        app_entity_id,
+        disambiguation_option="--app-entity-id",
+        required=app_required,
+    )
 
     # Infer or verify the package if we have an app entity to convert
     if app_definition:
@@ -89,7 +75,8 @@ def _pdf_v2_to_v1(
         if package_entity_id:
             # If the user specified a package entity ID,
             # check that the app entity targets the user-specified package entity
-            if target_package != package_entity_id:
+            # if the app entity is used by the command being run
+            if target_package != package_entity_id and app_required:
                 raise ClickException(
                     f"The application entity {app_definition.entity_id} does not "
                     f"target the application package entity {package_entity_id}. Either"
@@ -97,30 +84,21 @@ def _pdf_v2_to_v1(
                     f"or omit the --package-entity-id flag to automatically use the package entity "
                     f"that the application entity targets."
                 )
-        elif target_package in packages:
+        elif target_package in v2_definition.get_entities_by_type(
+            ApplicationPackageEntityModel.get_type()
+        ):
             # If the user didn't target a specific package entity, use the one the app entity targets
             package_entity_id = target_package
 
     # Determine the package entity to convert, there must be one
-    if package_entity_id:
-        # If the user specified a package entity ID (or we inferred one from the app entity), use that one directly
-        app_package_definition = packages.get(package_entity_id)
-    elif len(packages) == 1:
-        # Otherwise, if there is only one package entity, fall back to that one
-        app_package_definition = next(iter(packages.values()))
-    elif len(packages) > 1:
-        # If there are multiple package entities, the user must specify which one to use
-        raise ClickException(
-            "More than one application package entity exists in the project definition file, "
-            "specify --package-entity-id to choose which one to operate on."
-        )
-
-    # If we don't have a package entity to convert, error out since it's not optional
-    if not app_package_definition:
-        with_id = f'with ID "{package_entity_id}" ' if package_entity_id else ""
-        raise ClickException(
-            f"Could not find an application package entity {with_id}in the project definition file."
-        )
+    app_package_definition = find_entity(
+        v2_definition,
+        ApplicationPackageEntityModel,
+        package_entity_id,
+        disambiguation_option="--package-entity-id",
+        required=True,
+    )
+    assert app_package_definition is not None  # satisfy mypy
 
     # NativeApp
     if app_definition and app_definition.fqn.identifier:
@@ -180,7 +158,60 @@ def _pdf_v2_to_v1(
     return result.project_definition
 
 
-def nativeapp_definition_v2_to_v1(func):
+T = TypeVar("T", bound=EntityModelBase)
+
+
+def find_entity(
+    project_definition: DefinitionV20,
+    entity_class: Type[T],
+    entity_id: str,
+    disambiguation_option: str,
+    required: bool,
+) -> T | None:
+    """
+    Find an entity of the specified type in the project definition file.
+
+    If an ID is passed, only that entity will be considered,
+    otherwise look for a single entity of the specified type.
+
+    If there are multiple entities of the specified type,
+    the user must specify which one to use using the CLI option
+    named in the disambiguation_option parameter.
+
+    If no entity is found, an error is raised if required is True,
+    otherwise None is returned.
+    """
+
+    entity_type = entity_class.get_type()
+    entities = project_definition.get_entities_by_type(entity_type)
+
+    entity: Optional[T] = None
+
+    # Determine the package entity to convert, there must be one
+    if entity_id:
+        # If the user specified a package entity ID (or we inferred one from the app entity), use that one directly
+        entity = entities.get(entity_id)
+    elif len(entities) == 1:
+        # Otherwise, if there is only one package entity, fall back to that one
+        entity = next(iter(entities.values()))
+    elif len(entities) > 1:
+        # If there are multiple package entities, the user must specify which one to use
+        raise ClickException(
+            f"More than one {entity_type} entity exists in the project definition file, "
+            f"specify {disambiguation_option} to choose which one to operate on."
+        )
+
+    # If we don't have a package entity to convert, error out since it's not optional
+    if not entity and required:
+        with_id = f'with ID "{entity_id}" ' if entity_id else ""
+        raise ClickException(
+            f"Could not find an {entity_type} entity {with_id}in the project definition file."
+        )
+
+    return entity
+
+
+def nativeapp_definition_v2_to_v1(*, app_required: bool = False):
     """
     A command decorator that attempts to automatically convert a native app project from
     definition v2 to v1.1. Assumes with_project_definition() has already been called.
@@ -189,40 +220,45 @@ def nativeapp_definition_v2_to_v1(func):
     entity type is expected.
     """
 
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        original_pdf: Optional[DefinitionV20] = get_cli_context().project_definition
-        if not original_pdf:
-            raise ValueError(
-                "Project definition could not be found. The nativeapp_definition_v2_to_v1 command decorator assumes with_project_definition() was called before it."
-            )
-        if original_pdf.definition_version == "2":
-            package_entity_id = kwargs.get("package_entity_id", "")
-            app_entity_id = kwargs.get("app_entity_id", "")
-            pdfv1 = _pdf_v2_to_v1(original_pdf, package_entity_id, app_entity_id)
-            get_cli_context_manager().override_project_definition = pdfv1
-        return func(*args, **kwargs)
-
-    return _options_decorator_factory(
-        wrapper,
-        additional_options=[
-            inspect.Parameter(
-                "package_entity_id",
-                inspect.Parameter.KEYWORD_ONLY,
-                annotation=Optional[str],
-                default=typer.Option(
-                    default="",
-                    help="The ID of the package entity on which to operate when definition_version is 2 or higher.",
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            original_pdf: Optional[DefinitionV20] = get_cli_context().project_definition
+            if not original_pdf:
+                raise ValueError(
+                    "Project definition could not be found. The nativeapp_definition_v2_to_v1 command decorator assumes with_project_definition() was called before it."
+                )
+            if original_pdf.definition_version == "2":
+                package_entity_id = kwargs.get("package_entity_id", "")
+                app_entity_id = kwargs.get("app_entity_id", "")
+                pdfv1 = _pdf_v2_to_v1(
+                    original_pdf, package_entity_id, app_entity_id, app_required
+                )
+                get_cli_context_manager().override_project_definition = pdfv1
+            return func(*args, **kwargs)
+
+        return _options_decorator_factory(
+            wrapper,
+            additional_options=[
+                inspect.Parameter(
+                    "package_entity_id",
+                    inspect.Parameter.KEYWORD_ONLY,
+                    annotation=Optional[str],
+                    default=typer.Option(
+                        default="",
+                        help="The ID of the package entity on which to operate when definition_version is 2 or higher.",
+                    ),
                 ),
-            ),
-            inspect.Parameter(
-                "app_entity_id",
-                inspect.Parameter.KEYWORD_ONLY,
-                annotation=Optional[str],
-                default=typer.Option(
-                    default="",
-                    help="The ID of the application entity on which to operate when definition_version is 2 or higher.",
+                inspect.Parameter(
+                    "app_entity_id",
+                    inspect.Parameter.KEYWORD_ONLY,
+                    annotation=Optional[str],
+                    default=typer.Option(
+                        default="",
+                        help="The ID of the application entity on which to operate when definition_version is 2 or higher.",
+                    ),
                 ),
-            ),
-        ],
-    )
+            ],
+        )
+
+    return decorator
diff --git a/src/snowflake/cli/_plugins/nativeapp/version/commands.py b/src/snowflake/cli/_plugins/nativeapp/version/commands.py
index 69ee6794d2..8a3dcc61e5 100644
--- a/src/snowflake/cli/_plugins/nativeapp/version/commands.py
+++ b/src/snowflake/cli/_plugins/nativeapp/version/commands.py
@@ -51,7 +51,7 @@
 
 @app.command(requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1()
 def create(
     version: Optional[str] = typer.Argument(
         None,
@@ -117,7 +117,7 @@ def create(
 
 @app.command("list", requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1()
 def version_list(
     **options,
 ) -> CommandResult:
@@ -138,7 +138,7 @@ def version_list(
 
 @app.command(requires_connection=True)
 @with_project_definition()
-@nativeapp_definition_v2_to_v1
+@nativeapp_definition_v2_to_v1()
 def drop(
     version: Optional[str] = typer.Argument(
         None,
diff --git a/tests/__snapshots__/test_help_messages.ambr b/tests/__snapshots__/test_help_messages.ambr
index 915a1dd223..9d12c4ccc2 100644
--- a/tests/__snapshots__/test_help_messages.ambr
+++ b/tests/__snapshots__/test_help_messages.ambr
@@ -789,11 +789,6 @@
   |                                                    operate when              |
   |                                                    definition_version is 2   |
   |                                                    or higher.                |
-  | --app-entity-id                              TEXT  The ID of the application |
-  |                                                    entity on which to        |
-  |                                                    operate when              |
-  |                                                    definition_version is 2   |
-  |                                                    or higher.                |
   | --project            -p                      TEXT  Path where Snowflake      |
   |                                                    project resides. Defaults |
   |                                                    to current working        |
diff --git a/tests/nativeapp/test_v2_to_v1.py b/tests/nativeapp/test_v2_to_v1.py
index c0b5f90089..37849b9835 100644
--- a/tests/nativeapp/test_v2_to_v1.py
+++ b/tests/nativeapp/test_v2_to_v1.py
@@ -28,6 +28,7 @@
     DefinitionV11,
     DefinitionV20,
 )
+from snowflake.cli.api.utils.definition_rendering import render_definition_template
 
 
 def package_v2(entity_id: str):
@@ -75,7 +76,7 @@ def app_v2(entity_id: str, from_pkg: str):
 
 
 def native_app_v1(name: str, pkg: str, app: str):
-    return {
+    napp = {
         "name": name,
         "artifacts": [{"src": "app/*", "dest": "./"}],
         "source_stage": "app.stage",
@@ -93,7 +94,9 @@ def native_app_v1(name: str, pkg: str, app: str):
                 {"sql_script": "scripts/script2.sql"},
             ],
         },
-        "application": {
+    }
+    if app:
+        napp["application"] = {
             "name": app,
             "role": "app_role",
             "debug": True,
@@ -102,8 +105,8 @@ def native_app_v1(name: str, pkg: str, app: str):
                 {"sql_script": "scripts/script3.sql"},
                 {"sql_script": "scripts/script4.sql"},
             ],
-        },
-    }
+        }
+    return napp
 
 
 @pytest.mark.parametrize(
@@ -113,24 +116,25 @@ def native_app_v1(name: str, pkg: str, app: str):
             {
                 "definition_version": "2",
                 "entities": {
-                    **package_v2("pkg1"),
-                    **package_v2("pkg2"),
+                    **package_v2("pkg"),
                 },
             },
+            {
+                "definition_version": "1.1",
+                "native_app": native_app_v1("pkg", "pkg", ""),
+            },
             None,
-            "More than one application package entity exists",
         ],
         [
             {
                 "definition_version": "2",
                 "entities": {
-                    **package_v2("pkg"),
-                    **app_v2("app1", "pkg"),
-                    **app_v2("app2", "pkg"),
+                    **package_v2("pkg1"),
+                    **package_v2("pkg2"),
                 },
             },
             None,
-            "More than one application entity exists",
+            "More than one application package entity exists",
         ],
         [
             {
@@ -170,14 +174,16 @@ def test_v2_to_v1_conversions(pdfv2_input, expected_pdfv1, expected_error):
             _pdf_v2_to_v1(pdfv2)
     else:
         pdfv1_actual = vars(_pdf_v2_to_v1(pdfv2))
-        pdfv1_expected = vars(DefinitionV11(**expected_pdfv1))
+        pdfv1_expected = vars(
+            render_definition_template(expected_pdfv1, {}).project_definition
+        )
 
         # Assert that the expected dict is a subset of the actual dict
         assert {**pdfv1_actual, **pdfv1_expected} == pdfv1_actual
 
 
 @pytest.mark.parametrize(
-    "pdfv2_input, target_pkg, target_app, expected_pdfv1, expected_error",
+    "pdfv2_input, target_pkg, target_app, app_required, expected_pdfv1, expected_error",
     [
         [
             {
@@ -190,6 +196,7 @@ def test_v2_to_v1_conversions(pdfv2_input, expected_pdfv1, expected_error):
             },
             "",
             "",
+            True,
             None,
             "More than one application entity exists in the project definition file, "
             "specify --app-entity-id to choose which one to operate on.",
@@ -205,6 +212,7 @@ def test_v2_to_v1_conversions(pdfv2_input, expected_pdfv1, expected_error):
             },
             "pkg2",
             "app2",
+            True,
             None,
             "The application entity app2 does not "
             "target the application package entity pkg2.",
@@ -219,6 +227,56 @@ def test_v2_to_v1_conversions(pdfv2_input, expected_pdfv1, expected_error):
             },
             "",
             "",
+            True,
+            None,
+            "Could not find an application entity in the project definition file.",
+        ],
+        [
+            {
+                "definition_version": "2",
+                "entities": {
+                    **package_v2("pkg"),
+                    **app_v2("app1", "pkg"),
+                    **app_v2("app2", "pkg"),
+                },
+            },
+            "",
+            "",
+            True,
+            None,
+            "More than one application entity exists in the project definition file, "
+            "specify --app-entity-id to choose which one to operate on.",
+        ],
+        [
+            {
+                "definition_version": "2",
+                "entities": {
+                    **package_v2("pkg1"),
+                    **app_v2("app1", "pkg1"),
+                    **package_v2("pkg2"),
+                    **app_v2("app2", "pkg2"),
+                },
+            },
+            "pkg2",
+            "app2",
+            True,
+            {
+                "definition_version": "1.1",
+                "native_app": native_app_v1("app2", "pkg2", "app2"),
+            },
+            None,
+        ],
+        [
+            {
+                "definition_version": "2",
+                "entities": {
+                    **package_v2("pkg1"),
+                    **package_v2("pkg2"),
+                },
+            },
+            "",
+            "",
+            False,
             None,
             "More than one application package entity exists in the project definition file, "
             "specify --package-entity-id to choose which one to operate on.",
@@ -233,8 +291,9 @@ def test_v2_to_v1_conversions(pdfv2_input, expected_pdfv1, expected_error):
             },
             "pkg3",
             "",
+            False,
             None,
-            f'Could not find an application package entity with ID "pkg3" in the project definition file.',
+            'Could not find an application package entity with ID "pkg3" in the project definition file.',
         ],
         [
             {
@@ -248,6 +307,7 @@ def test_v2_to_v1_conversions(pdfv2_input, expected_pdfv1, expected_error):
             },
             "pkg2",
             "app2",
+            False,
             {
                 "definition_version": "1.1",
                 "native_app": native_app_v1("app2", "pkg2", "app2"),
@@ -257,17 +317,29 @@ def test_v2_to_v1_conversions(pdfv2_input, expected_pdfv1, expected_error):
     ],
 )
 def test_v2_to_v1_conversions_with_multiple_entities(
-    pdfv2_input, target_pkg, target_app, expected_pdfv1, expected_error
+    pdfv2_input, target_pkg, target_app, app_required, expected_pdfv1, expected_error
 ):
     pdfv2 = DefinitionV20(**pdfv2_input)
     if expected_error:
         with pytest.raises(ClickException, match=expected_error) as err:
-            _pdf_v2_to_v1(pdfv2, package_entity_id=target_pkg, app_entity_id=target_app)
+            _pdf_v2_to_v1(
+                pdfv2,
+                package_entity_id=target_pkg,
+                app_entity_id=target_app,
+                app_required=app_required,
+            )
     else:
         pdfv1_actual = vars(
-            _pdf_v2_to_v1(pdfv2, package_entity_id=target_pkg, app_entity_id=target_app)
+            _pdf_v2_to_v1(
+                pdfv2,
+                package_entity_id=target_pkg,
+                app_entity_id=target_app,
+                app_required=app_required,
+            )
+        )
+        pdfv1_expected = vars(
+            render_definition_template(expected_pdfv1, {}).project_definition
         )
-        pdfv1_expected = vars(DefinitionV11(**expected_pdfv1))
 
         # Assert that the expected dict is a subset of the actual dict
         assert {**pdfv1_actual, **pdfv1_expected} == pdfv1_actual
@@ -275,7 +347,7 @@ def test_v2_to_v1_conversions_with_multiple_entities(
 
 def test_decorator_error_when_no_project_exists():
     with pytest.raises(ValueError, match="Project definition could not be found"):
-        nativeapp_definition_v2_to_v1(lambda *args: None)()
+        nativeapp_definition_v2_to_v1()(lambda *args: None)()
 
 
 @pytest.mark.parametrize(
@@ -337,7 +409,7 @@ def test_decorator_skips_when_project_is_not_v2(mock_pdf_v2_to_v1):
     )
     get_cli_context_manager().override_project_definition = pdfv1
 
-    nativeapp_definition_v2_to_v1(lambda *args: None)()
+    nativeapp_definition_v2_to_v1()(lambda *args: None)()
 
     mock_pdf_v2_to_v1.launch.assert_not_called()
     assert get_cli_context().project_definition == pdfv1
diff --git a/tests/test_data/projects/integration_templated_v2/snowflake.yml b/tests/test_data/projects/integration_templated_v2/snowflake.yml
index 0242b42357..b3fb8ac074 100644
--- a/tests/test_data/projects/integration_templated_v2/snowflake.yml
+++ b/tests/test_data/projects/integration_templated_v2/snowflake.yml
@@ -12,6 +12,11 @@ entities:
       post_deploy:
         - sql_script: package/001-shared.sql
         - sql_script: package/002-shared.sql
+  app:
+    type: application
+    identifier: integration_<% ctx.env.INTERMEDIATE_CI_ENV %>_<% ctx.env.USER %>
+    from:
+      target: pkg
 env:
   INTERMEDIATE_CI_ENV: '<% ctx.env.CI_ENV %>'
   CI_ENV: 'dev'
diff --git a/tests/test_data/projects/integration_v2/snowflake.yml b/tests/test_data/projects/integration_v2/snowflake.yml
index 61cbeaa13d..04d363cb23 100644
--- a/tests/test_data/projects/integration_v2/snowflake.yml
+++ b/tests/test_data/projects/integration_v2/snowflake.yml
@@ -12,3 +12,8 @@ entities:
       post_deploy:
         - sql_script: package/001-shared.sql
         - sql_script: package/002-shared.sql
+  app:
+    type: application
+    identifier: integration_<% ctx.env.USER %>
+    from:
+      target: pkg
diff --git a/tests_integration/nativeapp/conftest.py b/tests_integration/nativeapp/conftest.py
index 498c9b6f4f..2ca4ac2bb3 100644
--- a/tests_integration/nativeapp/conftest.py
+++ b/tests_integration/nativeapp/conftest.py
@@ -5,6 +5,7 @@
 from typing import Any
 
 import pytest
+import yaml
 
 from tests_integration.conftest import SnowCLIRunner
 
@@ -60,7 +61,29 @@ def _nativeapp_teardown(
             kwargs: dict[str, Any] = {}
             if env:
                 kwargs["env"] = env
-            result = runner.invoke_with_connection(["app", "teardown", *args], **kwargs)
-            assert result.exit_code == 0
+
+            # `snow app teardown` can only teardown one package at a time for safety,
+            # so when cleaning up PDFv2 tests, we need to iterate all the package entities
+            # and teardown each one individually.
+            snowflake_yml = (project_dir or Path.cwd()) / "snowflake.yml"
+            with open(snowflake_yml, "r") as f:
+                project_yml = yaml.safe_load(f)
+            packages = [
+                entity_id
+                for entity_id, entity in project_yml.get("entities", {}).items()
+                if entity["type"] == "application package"
+            ]
+            if packages:
+                for package in packages:
+                    result = runner.invoke_with_connection(
+                        ["app", "teardown", *args, "--package-entity-id", package],
+                        **kwargs,
+                    )
+                    assert result.exit_code == 0
+            else:
+                result = runner.invoke_with_connection(
+                    ["app", "teardown", *args], **kwargs
+                )
+                assert result.exit_code == 0
 
     return _nativeapp_teardown
diff --git a/tests_integration/nativeapp/test_init_run.py b/tests_integration/nativeapp/test_init_run.py
index dec50703c9..2e6b2340bf 100644
--- a/tests_integration/nativeapp/test_init_run.py
+++ b/tests_integration/nativeapp/test_init_run.py
@@ -81,9 +81,12 @@ def test_nativeapp_init_run_multiple_pdf_entities(
     resource_suffix,
 ):
     project_name = "myapp"
-    entity_id_selector = ["--package-entity-id", "pkg2", "--app-entity-id", "app2"]
-    with nativeapp_project_directory(test_project, teardown_args=entity_id_selector):
-        result = runner.invoke_with_connection_json(["app", "run"] + entity_id_selector)
+    with nativeapp_project_directory(
+        test_project, teardown_args=["--package-entity-id", "pkg2"]
+    ):
+        result = runner.invoke_with_connection_json(
+            ["app", "run", "--package-entity-id", "pkg2", "--app-entity-id", "app2"]
+        )
         assert result.exit_code == 0
 
         # app + package exist
diff --git a/tests_integration/nativeapp/test_teardown.py b/tests_integration/nativeapp/test_teardown.py
index dbc65c2cc8..21399058f0 100644
--- a/tests_integration/nativeapp/test_teardown.py
+++ b/tests_integration/nativeapp/test_teardown.py
@@ -13,6 +13,8 @@
 # limitations under the License.
 from shlex import split
 
+import yaml
+
 from tests.project.fixtures import *
 from tests_integration.test_utils import (
     contains_row_with,
@@ -261,3 +263,122 @@ def test_nativeapp_teardown_pkg_versions(
         # either way, we can now tear down the application package
         result = runner.invoke_with_connection(split(command) + teardown_args)
         assert result.exit_code == 0
+
+
+def test_nativeapp_teardown_multiple_apps_using_snow_app(
+    runner,
+    nativeapp_project_directory,
+    snowflake_session,
+    default_username,
+    resource_suffix,
+):
+    test_project = "napp_init_v2"
+    project_name = "myapp"
+    pkg_name = f"{project_name}_pkg_{default_username}{resource_suffix}"
+    app_name_1 = f"{project_name}_{default_username}{resource_suffix}".upper()
+    app_name_2 = f"{project_name}2_{default_username}{resource_suffix}".upper()
+
+    with nativeapp_project_directory(test_project) as project_dir:
+        # Add a second app to the project
+        snowflake_yml = project_dir / "snowflake.yml"
+        with open(snowflake_yml, "r") as file:
+            project_yml = yaml.safe_load(file)
+        project_yml["entities"]["app2"] = project_yml["entities"]["app"] | dict(
+            identifier="myapp2_<% ctx.env.USER %>"
+        )
+        with open(snowflake_yml, "w") as file:
+            yaml.dump(project_yml, file)
+
+        # Create the package and both apps
+        result = runner.invoke_with_connection_json(
+            ["app", "run", "--app-entity-id", "app"]
+        )
+        assert result.exit_code == 0, result.output
+
+        result = runner.invoke_with_connection_json(
+            ["app", "run", "--app-entity-id", "app2"]
+        )
+        assert result.exit_code == 0, result.output
+
+        # Run the teardown command
+        result = runner.invoke_with_connection_json(["app", "teardown"])
+        assert result.exit_code == 0, result.output
+
+        # Verify the package is dropped
+        assert not_contains_row_with(
+            row_from_snowflake_session(
+                snowflake_session.execute_string(
+                    f"show application packages like '{pkg_name}'",
+                )
+            ),
+            dict(name=pkg_name),
+        )
+
+        # Verify the apps are dropped
+        for app_name in [app_name_1, app_name_2]:
+            assert not_contains_row_with(
+                row_from_snowflake_session(
+                    snowflake_session.execute_string(
+                        f"show applications like '{app_name}'",
+                    )
+                ),
+                dict(name=app_name),
+            )
+
+
+def test_nativeapp_teardown_multiple_packages_using_snow_app_must_choose(
+    runner,
+    nativeapp_project_directory,
+    snowflake_session,
+    default_username,
+    resource_suffix,
+):
+    test_project = "napp_init_v2"
+    project_name = "myapp"
+    pkgs = {
+        "pkg": f"{project_name}_pkg_{default_username}{resource_suffix}",
+        "pkg2": f"{project_name}_pkg2_{default_username}{resource_suffix}",
+    }
+
+    with nativeapp_project_directory(test_project) as project_dir:
+        # Add a second package to the project
+        snowflake_yml = project_dir / "snowflake.yml"
+        with open(snowflake_yml, "r") as file:
+            project_yml = yaml.safe_load(file)
+        project_yml["entities"]["pkg2"] = project_yml["entities"]["pkg"] | dict(
+            identifier="myapp_pkg2_<% ctx.env.USER %>"
+        )
+        with open(snowflake_yml, "w") as file:
+            yaml.dump(project_yml, file)
+
+        # Create both packages
+        for entity_id in pkgs:
+            result = runner.invoke_with_connection_json(
+                ["app", "deploy", "--package-entity-id", entity_id]
+            )
+            assert result.exit_code == 0, result.output
+
+        # Run the teardown command without specifying a package, it should fail
+        result = runner.invoke_with_connection_json(["app", "teardown"])
+        assert result.exit_code == 1, result.output
+        assert (
+            "More than one application package entity exists in the project definition"
+            in result.output
+        )
+
+        # Run the teardown command on each package
+        for entity_id, pkg_name in pkgs.items():
+            result = runner.invoke_with_connection_json(
+                ["app", "teardown", "--package-entity-id", entity_id]
+            )
+            assert result.exit_code == 0, result.output
+
+            # Verify the package is dropped
+            assert not_contains_row_with(
+                row_from_snowflake_session(
+                    snowflake_session.execute_string(
+                        f"show application packages like '{pkg_name}'",
+                    )
+                ),
+                dict(name=pkg_name),
+            )