Add latest version check (#1553)

RELEASE-NOTES.md

@@ -43,6 +43,7 @@
 * Added `--package-entity-id` and `--app-entity-id` options to `snow app` commands to allow targeting specific entities when the `definition_version` in `snowflake.yml` is `2` or higher and it contains multiple `application package` or `application` entities.
 * Added templates expansion of arbitrary files for Native Apps through `templates` processor.
 * Added `SNOWFLAKE_..._PRIVATE_KEY_RAW` environment variable to pass private key as a raw string.
+* Added periodic check for newest version of Snowflake CLI. When new version is available, user will be notified.
 
 ## Fixes and improvements
 * Fixed problem with whitespaces in `snow connection add` command.

src/snowflake/cli/_app/cli_app.py

@@ -38,6 +38,10 @@
 )
 from snowflake.cli._app.main_typer import SnowCliMainTyper
 from snowflake.cli._app.printing import MessageResult, print_result
+from snowflake.cli._app.version_check import (
+ get_new_version_msg,
+ show_new_version_banner_callback,
+)
 from snowflake.cli.api import Api, api_provider
 from snowflake.cli.api.config import config_init
 from snowflake.cli.api.output.formats import OutputFormat
@@ -145,8 +149,13 @@ def _info_callback(value: bool):
 
 def app_factory() -> SnowCliMainTyper:
  app = SnowCliMainTyper()
+ new_version_msg = get_new_version_msg()
 
- @app.callback(invoke_without_command=True)
+ @app.callback(
+ invoke_without_command=True,
+ epilog=new_version_msg,
+ result_callback=show_new_version_banner_callback(new_version_msg),
+ )
  def default(
  ctx: typer.Context,
  version: bool = typer.Option(

src/snowflake/cli/_app/version_check.py

@@ -0,0 +1,74 @@
+import json
+import time
+
+import requests
+from packaging.version import Version
+from snowflake.cli.__about__ import VERSION
+from snowflake.cli.api.console import cli_console
+from snowflake.cli.api.secure_path import SecurePath
+from snowflake.connector.config_manager import CONFIG_MANAGER
+
+
+def get_new_version_msg() -> str | None:
+ last = _VersionCache().get_last_version()
+ current = Version(VERSION)
+ if last and last > current:
+ return f"\nNew version of Snowflake CLI available. Newest: {last}, current: {VERSION}\n"
+ return None
+
+
+def show_new_version_banner_callback(msg):
+ def _callback(*args, **kwargs):
+ if msg:
+ cli_console.message(msg)
+
+ return _callback
+
+
+class _VersionCache:
+ _last_time = "last_time_check"
+ _version = "version"
+ _version_cache_file = SecurePath(
+ CONFIG_MANAGER.file_path.parent / ".cli_version.cache"
+ )
+
+ def __init__(self):
+ self._cache_file = _VersionCache._version_cache_file
+
+ def _save_latest_version(self, version: str):
+ data = {
+ _VersionCache._last_time: time.time(),
+ _VersionCache._version: str(version),
+ }
+ self._cache_file.write_text(json.dumps(data))
+
+ @staticmethod
+ def _get_version_from_pypi() -> str | None:
+ headers = {"Content-Type": "application/vnd.pypi.simple.v1+json"}
+ response = requests.get(
+ "https://pypi.org/pypi/snowflake-cli-labs/json", headers=headers, timeout=3
+ )
+ response.raise_for_status()
+ return response.json()["info"]["version"]
+
+ def _update_latest_version(self) -> Version | None:
+ version = self._get_version_from_pypi()
+ if version is None:
+ return None
+ self._save_latest_version(version)
+ return Version(version)
+
+ def _read_latest_version(self) -> Version | None:
+ if self._cache_file.exists():
+ data = json.loads(self._cache_file.read_text())
+ now = time.time()
+ if data[_VersionCache._last_time] > now - 60 * 60:
+ return Version(data[_VersionCache._version])
+
+ return self._update_latest_version()
+
+ def get_last_version(self) -> Version | None:
+ try:
+ return self._read_latest_version()
+ except: # anything, this it not crucial feature
+ return None

tests/app/test_version_check.py

@@ -0,0 +1,135 @@
+from io import BytesIO
+from unittest.mock import patch
+
+from packaging.version import Version
+from requests import Response
+from snowflake.cli._app.version_check import _VersionCache, get_new_version_msg
+
+
+@patch("snowflake.cli._app.version_check.VERSION", "1.0.0")
+@patch(
+ "snowflake.cli._app.version_check._VersionCache.get_last_version",
+ lambda _: Version("2.0.0"),
+)
+def test_banner_shows_up_in_help(build_runner):
+ runner = build_runner()
+ result = runner.invoke(["--help"])
+ msg = "New version of Snowflake CLI available. Newest: 2.0.0, current: 1.0.0"
+ assert msg in result.output
+
+
+@patch("snowflake.cli._app.version_check.VERSION", "1.0.0")
+@patch(
+ "snowflake.cli._app.version_check._VersionCache.get_last_version",
+ lambda _: Version("2.0.0"),
+)
+def test_banner_shows_up_in_command_invocation(build_runner):
+ runner = build_runner()
+ result = runner.invoke(["connection", "set-default", "default"])
+ msg = "New version of Snowflake CLI available. Newest: 2.0.0, current: 1.0.0"
+ assert msg in result.output
+
+
+@patch("snowflake.cli._app.version_check.VERSION", "1.0.0")
+@patch(
+ "snowflake.cli._app.version_check._VersionCache.get_last_version",
+ lambda _: Version("2.0.0"),
+)
+def test_banner_do_not_shows_up_if_silent(build_runner):
+ runner = build_runner()
+ result = runner.invoke(["connection", "set-default", "default", "--silent"])
+ msg = "New version of Snowflake CLI available. Newest: 2.0.0, current: 1.0.0"
+ assert msg not in result.output
+
+
+@patch("snowflake.cli._app.version_check._VersionCache._read_latest_version")
+def test_version_check_exception_are_handled_safely(
+ mock_read_latest_version, build_runner
+):
+ mock_read_latest_version.side_effect = Exception("Error")
+ runner = build_runner()
+ result = runner.invoke(["connection", "set-default", "default"])
+
+ msg = "New version of Snowflake CLI available. Newest: 2.0.0, current: 1.0.0"
+ assert result.exit_code == 0
+ assert msg not in result.output
+
+
+@patch("snowflake.cli._app.version_check.VERSION", "1.0.0")
+@patch(
+ "snowflake.cli._app.version_check._VersionCache.get_last_version",
+ lambda _: Version("2.0.0"),
+)
+def test_get_new_version_msg_message_if_new_version_available():
+ msg = get_new_version_msg()
+ assert (
+ msg.strip()
+ == "New version of Snowflake CLI available. Newest: 2.0.0, current: 1.0.0"
+ )
+
+
+@patch("snowflake.cli._app.version_check.VERSION", "1.0.0")
+@patch(
+ "snowflake.cli._app.version_check._VersionCache.get_last_version", lambda _: None
+)
+def test_get_new_version_msg_does_not_show_message_if_no_new_version():
+ assert get_new_version_msg() is None
+
+
+@patch("snowflake.cli._app.version_check.VERSION", "3.0.0")
+@patch(
+ "snowflake.cli._app.version_check._VersionCache.get_last_version",
+ lambda _: Version("2.0.0"),
+)
+def test_new_version_banner_does_not_show_message_if_local_version_is_newer():
+ assert get_new_version_msg() is None
+
+
+@patch("snowflake.cli._app.version_check.requests.get")
+def test_get_version_from_pypi(mock_get):
+ r = Response()
+ r.status_code = 200
+ r.raw = BytesIO(b'{"info": {"version": "1.2.3"}}')
+ mock_get.return_value = r
+ assert _VersionCache()._get_version_from_pypi() == "1.2.3" # noqa
+ mock_get.assert_called_once_with(
+ "https://pypi.org/pypi/snowflake-cli-labs/json",
+ headers={"Content-Type": "application/vnd.pypi.simple.v1+json"},
+ timeout=3,
+ )
+
+
+@patch("snowflake.cli._app.version_check.time.time", lambda: 0.0)
+def test_saves_latest_version(named_temporary_file):
+ with named_temporary_file() as f:
+ vc = _VersionCache()
+ vc._cache_file = f # noqa
+ vc._save_latest_version("1.2.3") # noqa
+ data = f.read_text()
+ assert data == '{"last_time_check": 0.0, "version": "1.2.3"}'
+
+
+@patch("snowflake.cli._app.version_check.time.time", lambda: 60)
+def test_read_last_version(named_temporary_file):
+ with named_temporary_file() as f:
+ vc = _VersionCache()
+ vc._cache_file = f # noqa
+ f.write_text('{"last_time_check": 0.0, "version": "4.2.3"}')
+ assert vc._read_latest_version() == Version("4.2.3") # noqa
+
+
+@patch(
+ "snowflake.cli._app.version_check._VersionCache._get_version_from_pypi",
+ lambda _: "8.0.0",
+)
+@patch("snowflake.cli._app.version_check.time.time")
+def test_read_last_version_and_updates_it(mock_time, named_temporary_file):
+ mock_time.side_effect = [2 * 60 * 60, 120]
+
+ with named_temporary_file() as f:
+ vc = _VersionCache()
+ vc._cache_file = f # noqa
+ f.write_text('{"last_time_check": 0.0, "version": "1.2.3"}')
+ assert vc._read_latest_version() == Version("8.0.0") # noqa
+ data = f.read_text()
+ assert data == '{"last_time_check": 120, "version": "8.0.0"}'

tests/testing_utils/fixtures.py

@@ -238,6 +238,15 @@ def runner(test_snowcli_config):
  yield SnowCLIRunner(app, test_snowcli_config)
 
 
+@pytest.fixture(scope="function")
+def build_runner(test_snowcli_config):
+ def func():
+ app = app_factory()
+ return SnowCLIRunner(app, test_snowcli_config)
+
+ return func
+
+
 @pytest.fixture
 def temp_dir():
  initial_dir = os.getcwd()