Created new test enviroments for execute command

tests_integration/test_data/projects/execute_streamlit/environment.yml

@@ -0,0 +1,6 @@
+name: sf_env
+channels:
+ - snowflake
+dependencies:
+ - streamlit
+ - snowflake-snowpark-python

tests_integration/test_data/projects/execute_streamlit/snowflake.yml

@@ -0,0 +1,10 @@
+definition_version: "1.1"
+streamlit:
+ name: "vmaleki_cli_app"
+ title: "Vida Uploaded File"
+ stage: "ST_DB.SIS.MY_STREAMLIT_STAGE"
+ database: "ST_DB"
+ schema: "SIS"
+ query_warehouse: "regress"
+ main_file: "streamlit_app.py"
+ env_file: "environment.yml"

tests_integration/test_streamlit.py

@@ -370,57 +370,85 @@ def _new_streamlit_role(snowflake_session, test_database):
  )
 
 
-def _execute_streamlit(runner, streamlit_name):
+def _execute_streamlit(runner, streamlit_name, database, schema):
  result = runner.invoke_with_connection_json(
- ["streamlit", "execute", streamlit_name, "--format", "json"]
+ [
+ "streamlit",
+ "execute",
+ streamlit_name,
+ "--database",
+ database,
+ "--schema",
+ schema,
+ "--format",
+ "json",
+ ]
  )
  assert result.exit_code == 0
  assert result.json == {"message": f"Streamlit {streamlit_name} executed."}
 
 
-def _execute_streamlit_failure(runner, streamlit_name):
- result = runner.invoke_with_connection(["streamlit", "execute", streamlit_name])
+def _execute_streamlit_failure(runner, streamlit_name, database, schema):
+ result = runner.invoke_with_connection(
+ [
+ "streamlit",
+ "execute",
+ streamlit_name,
+ "--database",
+ database,
+ "--schema",
+ schema,
+ ]
+ )
  assert result.exit_code == 1
  assert f"NameError: name {streamlit_name} is not defined" in result.output
 
 
 @pytest.mark.integration
 def test_execute_streamlit(runner, snowflake_session, test_database):
- stage_name = "streamlit_stage"
+ stage_name = "ST_DB.SIS.MY_STREAMLIT_STAGE"
+ database = "ST_DB"
+ schema = "SIS" # Update schema to match your snowflake.yml
+
  snowflake_session.execute_string(f"create stage {stage_name};")
 
- # Create a Streamlit app and upload it to the stage
+ # Create and deploy a Streamlit app
  streamlit_app = Path(
- "tests_integration/test_data/projects/streamlit_v2/streamlit_app.py"
+ "tests_integration/test_data/projects/execute_streamlit/streamlit_app.py"
  )
 
- _create_streamlit(
+ _deploy_streamlit(
  local_streamlit_file=streamlit_app,
  runner=runner,
  snowflake_session=snowflake_session,
  stage_name=stage_name,
+ replace=True,
  )
 
  # Test successful and failure executions
- _execute_streamlit(runner, streamlit_app.stem)
- _execute_streamlit_failure(runner, "nonexistent_app")
+ _execute_streamlit(runner, streamlit_app.stem, database, schema)
+ _execute_streamlit_failure(runner, "nonexistent_app", database, schema)
 
 
-def _create_streamlit(local_streamlit_file, runner, snowflake_session, stage_name):
+def _deploy_streamlit(
+ local_streamlit_file, runner, snowflake_session, stage_name, replace=True
+):
  streamlit_name = local_streamlit_file.stem
  stage_path = f"@{stage_name}/{local_streamlit_file.name}"
  snowflake_session.execute_string(
  f"put file://{local_streamlit_file.absolute()} @{stage_name} AUTO_COMPRESS=FALSE;"
  )
- command = (
+ command = [
  "streamlit",
- "create",
- streamlit_name,
+ "deploy",
  "--streamlit-file",
  stage_path,
  "--format",
  "json",
- )
+ ]
+ if replace:
+ command.append("--replace")
+
  result = runner.invoke_with_connection_json(command)
  assert result.exit_code == 0
  message: str = result.json.get("message", "")