Merge branch 'main' into SNOW-2226057-moving-driver-to-key-pair

Author: sfc-gh-pcyrek

test/integ/test_connection.py

@@ -193,6 +193,17 @@ def test_keep_alive_heartbeat_frequency_min(conn_cnx):
         assert cnx.client_session_keep_alive_heartbeat_frequency == 900
 
 
+@pytest.mark.skipolddriver
+def test_platform_detection_timeout(conn_cnx):
+    """Tests platform detection timeout.
+
+    Creates a connection with platform_detection_timeout parameter.
+    """
+    
+    with conn_cnx(platform_detection_timeout_seconds=2.5) as cnx:
+        assert cnx.platform_detection_timeout_seconds == 2.5
+
+
 def test_bad_db(conn_cnx):
     """Attempts to use a bad DB."""
     with conn_cnx(database="baddb") as cnx:
@@ -1399,3 +1410,80 @@ def test_file_utils_sanity_check():
     conn = create_connection("default")
     assert hasattr(conn._file_operation_parser, "parse_file_operation")
     assert hasattr(conn._stream_downloader, "download_as_stream")
+
+
+@pytest.mark.skipolddriver
+@pytest.mark.skipif(IS_WINDOWS, reason="chmod doesn't work on Windows")
+def test_unsafe_skip_file_permissions_check_skips_config_permissions_check(
+    db_parameters, tmp_path
+):
+    """Test that unsafe_skip_file_permissions_check flag bypasses permission checks on config files."""
+    # Write config file and set unsafe permissions (readable by others)
+    tmp_config_file = tmp_path / "config.toml"
+    tmp_config_file.write_text("[log]\n" "save_logs = false\n" 'level = "INFO"\n')
+    tmp_config_file.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IROTH)
+
+    def _run_select_1(unsafe_skip_file_permissions_check: bool):
+        warnings.simplefilter("always")
+        # Connect directly with db_parameters, using custom config file path
+        # We need to modify CONFIG_MANAGER to point to our test file
+        from snowflake.connector.config_manager import CONFIG_MANAGER
+
+        original_file_path = CONFIG_MANAGER.file_path
+        try:
+            CONFIG_MANAGER.file_path = tmp_config_file
+            CONFIG_MANAGER.conf_file_cache = None  # Force re-read
+            with snowflake.connector.connect(
+                **db_parameters,
+                unsafe_skip_file_permissions_check=unsafe_skip_file_permissions_check,
+            ) as conn:
+                with conn.cursor() as cur:
+                    result = cur.execute("select 1;").fetchall()
+                    assert result == [(1,)]
+        finally:
+            CONFIG_MANAGER.file_path = original_file_path
+            CONFIG_MANAGER.conf_file_cache = None
+
+    # Without the flag - should trigger permission warnings
+    with warnings.catch_warnings(record=True) as warning_list:
+        _run_select_1(unsafe_skip_file_permissions_check=False)
+    permission_warnings = [
+        w for w in warning_list if "Bad owner or permissions" in str(w.message)
+    ]
+    assert (
+        len(permission_warnings) > 0
+    ), "Expected permission warning when unsafe_skip_file_permissions_check=False"
+
+    # With the flag - should bypass permission checks and not show warnings
+    with warnings.catch_warnings(record=True) as warning_list:
+        _run_select_1(unsafe_skip_file_permissions_check=True)
+    permission_warnings = [
+        w for w in warning_list if "Bad owner or permissions" in str(w.message)
+    ]
+    assert (
+        len(permission_warnings) == 0
+    ), "Expected no permission warning when unsafe_skip_file_permissions_check=True"
+
+
+# The property snowflake_version is newly introduced and therefore should not be tested on old drivers.
+@pytest.mark.skipolddriver
+def test_snowflake_version():
+    import re
+
+    conn = create_connection("default")
+    # Assert that conn has a snowflake_version attribute
+    assert hasattr(
+        conn, "snowflake_version"
+    ), "conn should have a snowflake_version attribute"
+
+    # Assert that conn.snowflake_version is a string.
+    assert isinstance(
+        conn.snowflake_version, str
+    ), f"snowflake_version should be a string, but got {type(conn.snowflake_version)}"
+
+    # Assert that conn.snowflake_version is in the format of "x.y.z", where
+    # x, y and z are numbers.
+    version_pattern = r"^\d+\.\d+\.\d+$"
+    assert re.match(
+        version_pattern, conn.snowflake_version
+    ), f"snowflake_version should match pattern 'x.y.z', but got '{conn.snowflake_version}'"