SNOW-1706990 Update/remove tests that use NativeAppRunProcessor

src/snowflake/cli/_plugins/nativeapp/entities/application.py

@@ -185,52 +185,14 @@ def deploy_package():
  )
 
  def drop_application_before_upgrade(cascade: bool = False):
- if cascade:
- try:
- if application_objects := self.get_objects_owned_by_application(
- app_name, app_role
- ):
- application_objects_str = self.application_objects_to_str(
- application_objects
- )
- workspace_ctx.console.message(
- f"The following objects are owned by application {app_name} and need to be dropped:\n{application_objects_str}"
- )
- except ProgrammingError as err:
- if err.errno != APPLICATION_NO_LONGER_AVAILABLE:
- generic_sql_error_handler(err)
- workspace_ctx.console.warning(
- "The application owns other objects but they could not be determined."
- )
- user_prompt = "Do you want the Snowflake CLI to drop these objects, then drop the existing application object and recreate it?"
- else:
- user_prompt = "Do you want the Snowflake CLI to drop the existing application object and recreate it?"
-
- if not policy.should_proceed(user_prompt):
- if is_interactive:
- workspace_ctx.console.message(
- "Not upgrading the application object."
- )
- raise typer.Exit(0)
- else:
- workspace_ctx.console.message(
- "Cannot upgrade the application object non-interactively without --force."
- )
- raise typer.Exit(1)
- try:
- cascade_msg = " (cascade)" if cascade else ""
- workspace_ctx.console.step(
- f"Dropping application object {app_name}{cascade_msg}."
- )
- cascade_sql = " cascade" if cascade else ""
- sql_executor = get_sql_executor()
- sql_executor.execute_query(f"drop application {app_name}{cascade_sql}")
- except ProgrammingError as err:
- if err.errno == APPLICATION_OWNS_EXTERNAL_OBJECTS and not cascade:
- # We need to cascade the deletion, let's try again (only if we didn't try with cascade already)
- return drop_application_before_upgrade(cascade=True)
- else:
- generic_sql_error_handler(err)
+ self.drop_application_before_upgrade(
+ console=workspace_ctx.console,
+ app_name=app_name,
+ app_role=app_role,
+ policy=policy,
+ is_interactive=is_interactive,
+ cascade=cascade,
+ )
 
  self.deploy(
  console=workspace_ctx.console,
@@ -800,6 +762,66 @@ def get_existing_app_info_static(app_name: str, app_role: str) -> Optional[dict]
  "applications", app_name, name_col=NAME_COL
  )
 
+ @classmethod
+ def drop_application_before_upgrade(
+ cls,
+ console: AbstractConsole,
+ app_name: str,
+ app_role: str,
+ policy: PolicyBase,
+ is_interactive: bool,
+ cascade: bool = False,
+ ):
+ if cascade:
+ try:
+ if application_objects := cls.get_objects_owned_by_application(
+ app_name, app_role
+ ):
+ application_objects_str = cls.application_objects_to_str(
+ application_objects
+ )
+ console.message(
+ f"The following objects are owned by application {app_name} and need to be dropped:\n{application_objects_str}"
+ )
+ except ProgrammingError as err:
+ if err.errno != APPLICATION_NO_LONGER_AVAILABLE:
+ generic_sql_error_handler(err)
+ console.warning(
+ "The application owns other objects but they could not be determined."
+ )
+ user_prompt = "Do you want the Snowflake CLI to drop these objects, then drop the existing application object and recreate it?"
+ else:
+ user_prompt = "Do you want the Snowflake CLI to drop the existing application object and recreate it?"
+
+ if not policy.should_proceed(user_prompt):
+ if is_interactive:
+ console.message("Not upgrading the application object.")
+ raise typer.Exit(0)
+ else:
+ console.message(
+ "Cannot upgrade the application object non-interactively without --force."
+ )
+ raise typer.Exit(1)
+ try:
+ cascade_msg = " (cascade)" if cascade else ""
+ console.step(f"Dropping application object {app_name}{cascade_msg}.")
+ cascade_sql = " cascade" if cascade else ""
+ sql_executor = get_sql_executor()
+ sql_executor.execute_query(f"drop application {app_name}{cascade_sql}")
+ except ProgrammingError as err:
+ if err.errno == APPLICATION_OWNS_EXTERNAL_OBJECTS and not cascade:
+ # We need to cascade the deletion, let's try again (only if we didn't try with cascade already)
+ return cls.drop_application_before_upgrade(
+ console=console,
+ app_name=app_name,
+ app_role=app_role,
+ policy=policy,
+ is_interactive=is_interactive,
+ cascade=True,
+ )
+ else:
+ generic_sql_error_handler(err)
+
  @classmethod
  def get_events(
  cls,

tests/nativeapp/factories.py

@@ -118,6 +118,30 @@ def _create(cls, model_class, *args, **kwargs):
  return cls._build(model_class, *args, **kwargs)
 
 
+class MetaFieldFactory(factory.DictFactory):
+ post_deploy = factory.List([])
+
+
+class EntityModelBaseFactory(factory.DictFactory):
+ meta = factory.SubFactory(MetaFieldFactory)
+
+
+class ApplicationPackageEntityModelFactory(EntityModelBaseFactory):
+ type = "application package" # noqa: A003
+ manifest = "manifest.yml"
+ artifacts = factory.List(
+ ["setup.sql", "README.md", "manifest.yml"], list_factory=ArtifactFactory
+ )
+
+
+class ApplicationEntityModelFactory(EntityModelBaseFactory):
+ type = "application" # noqa: A003
+ fromm = factory.Dict({"target": "pkg"})
+
+ class Meta:
+ rename = {"fromm": "from"}
+
+
 @dataclass
 class PdfFactoryResult:
  yml: dict
@@ -127,41 +151,24 @@ def as_json_str(self):
  return json.dumps(self.yml)
 
 
-class PdfV10Factory(factory.DictFactory):
+class _PdfFactory(factory.DictFactory):
  """
- Prepare PDF V1 dict and write to file.
+ Base class to prepare PDF dict and write to file.
 
  Returns:
  PdfFactoryResult
-
- Usage:
- Create a pdf dict with definition_version: "1", native_app with faker-generated name and an empty artifacts list and
- write to snowflake.yml in current directory:
- - PdfV10Factory()
-
- Create snowflake.local.yml and write to file
- - PdfV10Factory.with_filename("snowflake.local.yml")(native_app__name="my_local_name")
-
- Build and return yml but do not write to file:
- - PdfV10Factory.build(
- native_app__name="my_app",
- native_app__artifacts=["setup.sql", "README.md"],
- native_app__package__role="pkg_role"
- )
  """
 
- definition_version = "1"
- native_app = factory.SubFactory(NativeAppFactory)
  env = factory.SubFactory(FactoryNoEmptyDict)
  _filename = "snowflake.yml"
 
  # for snowflake.local.yml
  @classmethod
  def with_filename(cls, filename):
- class PdfV10FactoryWithFilename(cls):
+ class _PdfFactoryWithFilename(cls):
  _filename = filename
 
- return PdfV10FactoryWithFilename
+ return _PdfFactoryWithFilename
 
  @classmethod
  def _build(cls, model_class, *args, **kwargs):
@@ -185,10 +192,83 @@ def _create(cls, model_class, *args, **kwargs) -> PdfFactoryResult:
  )
 
 
+class PdfV10Factory(_PdfFactory):
+ """
+ Prepare PDF 1.0 dict and write to file.
+
+ Returns:
+ PdfFactoryResult
+
+ Usage:
+ Create a PDF dict with definition_version: "1", native_app with faker-generated name and an empty artifacts list and
+ write to snowflake.yml in current directory:
+ - PdfV10Factory()
+
+ Create snowflake.local.yml and write to file
+ - PdfV10Factory.with_filename("snowflake.local.yml")(native_app__name="my_local_name")
+
+ Build and return yml but do not write to file:
+ - PdfV10Factory.build(
+ native_app__name="my_app",
+ native_app__artifacts=["setup.sql", "README.md"],
+ native_app__package__role="pkg_role"
+ )
+ """
+
+ definition_version = "1"
+ native_app = factory.SubFactory(NativeAppFactory)
+
+
 class PdfV11Factory(PdfV10Factory):
+ """Override of Pdfv10Factory to set definition_version to 1.1"""
+
  definition_version = "1.1"
 
 
+class PdfV2Factory(_PdfFactory):
+ """
+ Prepare PDF 2 dict and write to file.
+
+ Returns:
+ PdfFactoryResult
+
+ Usage:
+ Create a PDF dict with definition_version: "2" with empty list of entities and
+ write to snowflake.yml in current directory:
+ - PdfV2Factory()
+
+ Create snowflake.local.yml with some entities and write to file
+ - PdfV2Factory.with_filename("snowflake.local.yml")(
+ entities=dict(
+ pkg=ApplicationPackageEntityModelFactory(
+ identifier="myapp_pkg",
+ ),
+ app=ApplicationEntityModelFactory(
+ identifier="myapp",
+ fromm__target="pkg",
+ ),
+ )
+ )
+
+ Build and return yml but do not write to file:
+ - PdfV2Factory.build(
+ entities=dict(
+ pkg=ApplicationPackageEntityModelFactory(
+ identifier="myapp_pkg",
+ ),
+ app=ApplicationEntityModelFactory(
+ identifier="myapp",
+ fromm__target="pkg",
+ ),
+ )
+ )
+ """
+
+ definition_version = "2"
+ entities = factory.Dict({})
+ env = factory.Dict({})
+
+
 @dataclass
 class FileModel:
  filename: Union[str, Path]
@@ -226,16 +306,15 @@ class ProjectFactoryModel:
 ProjectFiles = dict[str | Path, str]
 
 
-class ProjectV10Factory(factory.Factory):
+class _ProjectFactory(factory.Factory):
  """
  Factory to create PDF dict, and write in working directory PDF to snowflake.yml file, and other optional files.
  """
 
  class Meta:
  model = ProjectFactoryModel
 
- pdf = factory.SubFactory(PdfV10Factory)
-
+ pdf = factory.SubFactory(_PdfFactory)
  files: ProjectFiles = {}
 
  @classmethod
@@ -245,6 +324,13 @@ def _create(cls, model_class, *args, **kwargs):
  return super()._create(model_class, *args, **kwargs)
 
 
-class ProjectV11Factory(ProjectV10Factory):
+class ProjectV10Factory(_ProjectFactory):
+ pdf = factory.SubFactory(PdfV10Factory)
 
+
+class ProjectV11Factory(ProjectV10Factory):
  pdf = factory.SubFactory(PdfV11Factory)
+
+
+class ProjectV2Factory(_ProjectFactory):
+ pdf = factory.SubFactory(PdfV2Factory)

tests/nativeapp/test_manager.py

@@ -86,19 +86,6 @@
 from tests.testing_utils.files_and_dirs import create_named_file
 from tests.testing_utils.fixtures import MockConnectionCtx
 
-mock_project_definition_override = {
- "native_app": {
- "application": {
- "name": "sample_application_name",
- "role": "sample_application_role",
- },
- "package": {
- "name": "sample_package_name",
- "role": "sample_package_role",
- },
- }
-}
-
 
 def _get_dm(working_dir: Optional[str] = None):
  return DefinitionManager(working_dir)