Skip to content

Commit

Permalink
Add unit test
Browse files Browse the repository at this point in the history
  • Loading branch information
@sfc-gh-fcampbell
sfc-gh-fcampbell committed Jun 20, 2024
1 parent e823b38 commit 71190c5
Showing 1 changed file with 274 additions and 0 deletions.
274 changes: 274 additions & 0 deletions tests/nativeapp/test_run_processor.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
from click import UsageError
from snowflake.cli.api.project.definition_manager import DefinitionManager
from snowflake.cli.plugins.nativeapp.constants import (
ERROR_MESSAGE_093079,
ERROR_MESSAGE_093128,
LOOSE_FILES_MAGIC_VERSION,
SPECIAL_COMMENT,
)
Expand Down Expand Up @@ -693,6 +695,278 @@ def test_create_dev_app_create_new_quoted_override(
assert mock_execute.mock_calls == expected


# Test run existing app info
# AND app package has been dropped
# AND user wants to drop app
# AND drop succeeds
# AND app is created successfully.
@mock.patch(NATIVEAPP_MANAGER_EXECUTE)
@mock.patch(RUN_PROCESSOR_GET_EXISTING_APP_INFO)
@mock_connection()
def test_create_dev_app_recreate_app_when_orphaned(
mock_conn,
mock_get_existing_app_info,
mock_execute,
temp_dir,
mock_cursor,
):
mock_get_existing_app_info.return_value = {
"name": "myapp",
"comment": SPECIAL_COMMENT,
"owner": "app_role",
"version": LOOSE_FILES_MAGIC_VERSION,
}
side_effects, expected = mock_execute_helper(
[
(
mock_cursor([{"CURRENT_ROLE()": "old_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(None, mock.call("use role app_role")),
(None, mock.call("use warehouse app_warehouse")),
(
ProgrammingError(
msg=ERROR_MESSAGE_093079,
errno=93079,
),
mock.call(
"alter application myapp upgrade using @app_pkg.app_src.stage"
),
),
(None, mock.call("drop application myapp")),
(
mock_cursor([{"CURRENT_ROLE()": "app_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(None, mock.call("use role package_role")),
(None, mock.call("use role app_role")),
(
None,
mock.call(
dedent(
f"""\
create application myapp
from application package app_pkg
using @app_pkg.app_src.stage
debug_mode = True
comment = {SPECIAL_COMMENT}
"""
)
),
),
(None, mock.call("use role old_role")),
]
)
mock_conn.return_value = MockConnectionCtx()
mock_execute.side_effect = side_effects

current_working_directory = os.getcwd()
create_named_file(
file_name="snowflake.yml",
dir_name=current_working_directory,
contents=[mock_snowflake_yml_file],
)

run_processor = _get_na_run_processor()
run_processor._create_dev_app(allow_always_policy) # noqa: SLF001
assert mock_execute.mock_calls == expected


# Test run existing app info
# AND app package has been dropped
# AND user wants to drop app
# AND drop requires cascade
# AND drop succeeds
# AND app is created successfully.
@mock.patch(NATIVEAPP_MANAGER_EXECUTE)
@mock.patch(RUN_PROCESSOR_GET_EXISTING_APP_INFO)
@mock_connection()
def test_create_dev_app_recreate_app_when_orphaned_requires_cascade(
mock_conn,
mock_get_existing_app_info,
mock_execute,
temp_dir,
mock_cursor,
):
mock_get_existing_app_info.return_value = {
"name": "myapp",
"comment": SPECIAL_COMMENT,
"owner": "app_role",
"version": LOOSE_FILES_MAGIC_VERSION,
}
side_effects, expected = mock_execute_helper(
[
(
mock_cursor([{"CURRENT_ROLE()": "old_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(None, mock.call("use role app_role")),
(None, mock.call("use warehouse app_warehouse")),
(
ProgrammingError(
msg=ERROR_MESSAGE_093079,
errno=93079,
),
mock.call(
"alter application myapp upgrade using @app_pkg.app_src.stage"
),
),
(
ProgrammingError(
msg=ERROR_MESSAGE_093128,
errno=93128,
),
mock.call("drop application myapp"),
),
(
mock_cursor([{"CURRENT_ROLE()": "app_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(
mock_cursor(
[
[None, "mypool", "COMPUTE_POOL"],
],
[],
),
mock.call("show objects owned by application myapp"),
),
(None, mock.call("drop application myapp cascade")),
(
mock_cursor([{"CURRENT_ROLE()": "app_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(None, mock.call("use role package_role")),
(None, mock.call("use role app_role")),
(
None,
mock.call(
dedent(
f"""\
create application myapp
from application package app_pkg
using @app_pkg.app_src.stage
debug_mode = True
comment = {SPECIAL_COMMENT}
"""
)
),
),
(None, mock.call("use role old_role")),
]
)
mock_conn.return_value = MockConnectionCtx()
mock_execute.side_effect = side_effects

current_working_directory = os.getcwd()
create_named_file(
file_name="snowflake.yml",
dir_name=current_working_directory,
contents=[mock_snowflake_yml_file],
)

run_processor = _get_na_run_processor()
run_processor._create_dev_app(allow_always_policy) # noqa: SLF001
assert mock_execute.mock_calls == expected


# Test run existing app info
# AND app package has been dropped
# AND user wants to drop app
# AND drop requires cascade
# AND we can't see which objects are owned by the app
# AND drop succeeds
# AND app is created successfully.
@mock.patch(NATIVEAPP_MANAGER_EXECUTE)
@mock.patch(RUN_PROCESSOR_GET_EXISTING_APP_INFO)
@mock_connection()
def test_create_dev_app_recreate_app_when_orphaned_requires_cascade_unknown_objects(
mock_conn,
mock_get_existing_app_info,
mock_execute,
temp_dir,
mock_cursor,
):
mock_get_existing_app_info.return_value = {
"name": "myapp",
"comment": SPECIAL_COMMENT,
"owner": "app_role",
"version": LOOSE_FILES_MAGIC_VERSION,
}
side_effects, expected = mock_execute_helper(
[
(
mock_cursor([{"CURRENT_ROLE()": "old_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(None, mock.call("use role app_role")),
(None, mock.call("use warehouse app_warehouse")),
(
ProgrammingError(
msg=ERROR_MESSAGE_093079,
errno=93079,
),
mock.call(
"alter application myapp upgrade using @app_pkg.app_src.stage"
),
),
(
ProgrammingError(
msg=ERROR_MESSAGE_093128,
errno=93128,
),
mock.call("drop application myapp"),
),
(
mock_cursor([{"CURRENT_ROLE()": "app_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(
ProgrammingError(
msg=ERROR_MESSAGE_093079,
errno=93079,
),
mock.call("show objects owned by application myapp"),
),
(None, mock.call("drop application myapp cascade")),
(
mock_cursor([{"CURRENT_ROLE()": "app_role"}], []),
mock.call("select current_role()", cursor_class=DictCursor),
),
(None, mock.call("use role package_role")),
(None, mock.call("use role app_role")),
(
None,
mock.call(
dedent(
f"""\
create application myapp
from application package app_pkg
using @app_pkg.app_src.stage
debug_mode = True
comment = {SPECIAL_COMMENT}
"""
)
),
),
(None, mock.call("use role old_role")),
]
)
mock_conn.return_value = MockConnectionCtx()
mock_execute.side_effect = side_effects

current_working_directory = os.getcwd()
create_named_file(
file_name="snowflake.yml",
dir_name=current_working_directory,
contents=[mock_snowflake_yml_file],
)

run_processor = _get_na_run_processor()
run_processor._create_dev_app(allow_always_policy) # noqa: SLF001
assert mock_execute.mock_calls == expected


# Test upgrade app method for release directives AND throws warehouse error
@mock.patch(NATIVEAPP_MANAGER_EXECUTE)
@mock_connection()
Expand Down

0 comments on commit 71190c5

Please sign in to comment.