Aligned kedro-viz telemetry with kedro-telemetry

RELEASE.md

@@ -5,6 +5,14 @@ Please follow the established format:
 - Use present tense (e.g. 'Add new feature')
 - Include the ID number for the related PR (or PRs) in parentheses
 -->
+
+# Release 10.1.0
+
+## Major features and improvements
+
+- Update Kedro-Viz telemetry for opt-out model (#2022)
+
+
 # Release 10.0.0
 
 ## Major features and improvements

package/kedro_viz/integrations/kedro/telemetry.py

@@ -1,14 +1,15 @@
 """`kedro_viz.integrations.kedro.telemetry` helps integrate Kedro-Viz with Kedro-Telemetry
 """
-import hashlib
-import socket
+
 from pathlib import Path
 from typing import Optional
 
-import yaml
-
 try:
-    from kedro_telemetry.plugin import _get_heap_app_id, _is_valid_syntax
+    from kedro_telemetry.plugin import (
+        _check_for_telemetry_consent,
+        _get_heap_app_id,
+        _get_or_create_uuid,
+    )
 
     _IS_TELEMETRY_INSTALLED = True
 except ImportError:  # pragma: no cover
@@ -19,23 +20,17 @@ def get_heap_app_id(project_path: Path) -> Optional[str]:
     """Return the Heap App ID used for Kedro telemetry if user has given consent."""
     if not _IS_TELEMETRY_INSTALLED:  # pragma: no cover
         return None
-    telemetry_file_path = project_path / ".telemetry"
-    if not telemetry_file_path.exists():
-        return None
-    with open(
-        telemetry_file_path, encoding="utf8"
-    ) as telemetry_file:  # pylint: disable: unspecified-encoding
-        telemetry = yaml.safe_load(telemetry_file)
-        if _is_valid_syntax(telemetry) and telemetry["consent"]:
-            return _get_heap_app_id()
+
+    if _check_for_telemetry_consent(project_path):
+        return _get_heap_app_id()
     return None
 
 
 def get_heap_identity() -> Optional[str]:  # pragma: no cover
-    """Return the user ID in heap identical to the id used by kedro-telemetry plugin."""
+    """Reads a UUID from a configuration file or generates and saves a new one if not present."""
     if not _IS_TELEMETRY_INSTALLED:
         return None
     try:
-        return hashlib.sha512(bytes(socket.gethostname(), encoding="utf8")).hexdigest()
-    except socket.timeout:
+        return _get_or_create_uuid()
+    except Exception:
         return None

package/requirements.txt

@@ -4,6 +4,7 @@ fastapi>=0.100.0,<0.200.0
 fsspec>=2021.4
 ipython>=7.0.0, <9.0
 kedro>=0.18.0
+kedro-telemetry>=0.6.0
 networkx>=2.5
 orjson>=3.9, <4.0
 packaging>=23.0

package/tests/test_api/test_apps.py

@@ -10,7 +10,7 @@ class TestIndexEndpoint:
     def test_index(self, client):
         response = client.get("/")
         assert response.status_code == 200
-        assert "heap" not in response.text
+        assert "heap" in response.text
         assert "checkReloadStatus" not in response.text
 
     @mock.patch("kedro_viz.integrations.kedro.telemetry.get_heap_app_id")

package/tests/test_integrations/test_telemetry.py

@@ -5,13 +5,13 @@
 
 
 def test_get_heap_app_id_no_telemetry_file():
-    assert kedro_telemetry.get_heap_app_id(Path.cwd()) is None
+    assert kedro_telemetry.get_heap_app_id(Path.cwd()) is not None
 
 
 def test_get_heap_app_id_invalid_telemetry_file(tmpdir):
     telemetry_file = tmpdir / ".telemetry"
     telemetry_file.write_text("foo", encoding="utf-8")
-    assert kedro_telemetry.get_heap_app_id(tmpdir) is None
+    assert kedro_telemetry.get_heap_app_id(tmpdir) is not None
 
 
 def test_get_heap_app_id_no_consent(tmpdir):
@@ -21,8 +21,22 @@ def test_get_heap_app_id_no_consent(tmpdir):
 
 
 @mock.patch("kedro_viz.integrations.kedro.telemetry._get_heap_app_id")
-def test_get_heap_app_id_with_consent(original_get_heap_app_id, tmpdir):
-    original_get_heap_app_id.return_value = "my_heap_id"
+@mock.patch("kedro_viz.integrations.kedro.telemetry._check_for_telemetry_consent")
+def test_get_heap_app_id_with_consent(
+    mock_check_for_telemetry_consent, mock_get_heap_app_id, tmpdir
+):
+    mock_check_for_telemetry_consent.return_value = True
+    mock_get_heap_app_id.return_value = "my_heap_id"
     telemetry_file = tmpdir / ".telemetry"
     telemetry_file.write_text("consent: true", encoding="utf-8")
     assert kedro_telemetry.get_heap_app_id(tmpdir) == "my_heap_id"
+
+
+@mock.patch("kedro_viz.integrations.kedro.telemetry._check_for_telemetry_consent")
+def test_get_heap_app_id_with_env_var(mock_check_for_telemetry_consent, tmpdir):
+    mock_check_for_telemetry_consent.return_value = False
+    with mock.patch.dict("os.environ", {"DO_NOT_TRACK": "1"}):
+        assert kedro_telemetry.get_heap_app_id(tmpdir) is None
+
+    with mock.patch.dict("os.environ", {"KEDRO_DISABLE_TELEMETRY": "1"}):
+        assert kedro_telemetry.get_heap_app_id(tmpdir) is None