mirror of
https://github.com/DRYTRIX/TimeTracker.git
synced 2026-05-18 04:08:48 -05:00
Dries Peeters
905f6fbd37
Switch product and installation telemetry to OTLP/Grafana across runtime config, CI injection, docs, and tests to unify telemetry transport and simplify privacy-focused opt-in behavior.
277 lines
11 KiB
Python
277 lines
11 KiB
Python
"""
|
|
Tests for telemetry functionality
|
|
"""
|
|
|
|
import pytest
|
|
import os
|
|
import json
|
|
import tempfile
|
|
from unittest.mock import patch, MagicMock
|
|
from app.utils.telemetry import (
|
|
get_telemetry_fingerprint,
|
|
is_telemetry_enabled,
|
|
send_telemetry_ping,
|
|
send_install_ping,
|
|
send_update_ping,
|
|
send_health_ping,
|
|
should_send_telemetry,
|
|
mark_telemetry_sent,
|
|
check_and_send_telemetry,
|
|
)
|
|
|
|
|
|
class TestTelemetryFingerprint:
|
|
"""Tests for telemetry fingerprint generation"""
|
|
|
|
def test_fingerprint_is_consistent(self):
|
|
"""Test that fingerprint is consistent for same inputs"""
|
|
with patch.dict(os.environ, {"TELE_SALT": "test-salt"}):
|
|
fp1 = get_telemetry_fingerprint()
|
|
fp2 = get_telemetry_fingerprint()
|
|
assert fp1 == fp2
|
|
|
|
def test_fingerprint_changes_with_salt(self):
|
|
"""Test that fingerprint changes when salt changes"""
|
|
# Mock the installation config to force fallback to environment variable
|
|
with patch("app.utils.telemetry.get_installation_config") as mock_config:
|
|
mock_config.side_effect = Exception("Force fallback to env var")
|
|
|
|
with patch.dict(os.environ, {"TELE_SALT": "salt1"}):
|
|
fp1 = get_telemetry_fingerprint()
|
|
|
|
with patch.dict(os.environ, {"TELE_SALT": "salt2"}):
|
|
fp2 = get_telemetry_fingerprint()
|
|
|
|
assert fp1 != fp2
|
|
|
|
def test_fingerprint_is_sha256_hash(self):
|
|
"""Test that fingerprint is a valid SHA-256 hash"""
|
|
fp = get_telemetry_fingerprint()
|
|
assert len(fp) == 64 # SHA-256 produces 64 hex characters
|
|
assert all(c in "0123456789abcdef" for c in fp)
|
|
|
|
|
|
class TestTelemetryEnabled:
|
|
"""Tests for telemetry enabled check"""
|
|
|
|
@pytest.mark.parametrize(
|
|
"value,expected",
|
|
[
|
|
("true", True),
|
|
("True", True),
|
|
("TRUE", True),
|
|
("1", True),
|
|
("yes", True),
|
|
("on", True),
|
|
("false", False),
|
|
("False", False),
|
|
("0", False),
|
|
("no", False),
|
|
("", False),
|
|
("random", False),
|
|
],
|
|
)
|
|
def test_telemetry_enabled_values(self, value, expected):
|
|
"""Test various values for ENABLE_TELEMETRY"""
|
|
# Mock installation config to force fallback to environment variable
|
|
with patch("app.utils.telemetry.get_installation_config") as mock_config:
|
|
mock_config.side_effect = Exception("Force fallback to env var")
|
|
with patch.dict(os.environ, {"ENABLE_TELEMETRY": value}):
|
|
assert is_telemetry_enabled() == expected
|
|
|
|
def test_telemetry_disabled_by_default(self):
|
|
"""Test that telemetry is disabled by default"""
|
|
# Mock installation config to force fallback to environment variable
|
|
with patch("app.utils.telemetry.get_installation_config") as mock_config:
|
|
mock_config.side_effect = Exception("Force fallback to env var")
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
assert is_telemetry_enabled() is False
|
|
|
|
|
|
class TestSendTelemetryPing:
|
|
"""Tests for sending telemetry pings"""
|
|
|
|
@patch("app.telemetry.service._send_otlp_event")
|
|
def test_send_ping_when_enabled(self, mock_send):
|
|
"""Test sending telemetry ping when enabled"""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_TELEMETRY": "true",
|
|
"OTEL_EXPORTER_OTLP_ENDPOINT": "https://otlp.example.com",
|
|
"OTEL_EXPORTER_OTLP_TOKEN": "test-token",
|
|
"APP_VERSION": "1.0.0",
|
|
"TELE_SALT": "test-salt",
|
|
},
|
|
):
|
|
result = send_telemetry_ping("install")
|
|
assert result is True
|
|
assert mock_send.called
|
|
|
|
call_args = mock_send.call_args
|
|
assert call_args[1]["event_name"] == "telemetry.install"
|
|
assert "identity" in call_args[1]
|
|
assert "properties" in call_args[1]
|
|
|
|
@patch("app.telemetry.service._send_otlp_event")
|
|
def test_no_ping_when_disabled(self, mock_send):
|
|
"""Test that no ping is sent when telemetry is disabled"""
|
|
with patch.dict(os.environ, {"ENABLE_TELEMETRY": "false"}):
|
|
result = send_telemetry_ping("install")
|
|
assert result is False
|
|
assert not mock_send.called
|
|
|
|
@patch("app.telemetry.service._send_otlp_event")
|
|
def test_no_ping_when_no_sink_config(self, mock_send):
|
|
"""Test that no ping is sent when OTLP sink is not set."""
|
|
with patch.dict(
|
|
os.environ, {"ENABLE_TELEMETRY": "true", "OTEL_EXPORTER_OTLP_ENDPOINT": "", "OTEL_EXPORTER_OTLP_TOKEN": ""}
|
|
):
|
|
result = send_telemetry_ping("install")
|
|
assert result is False
|
|
assert not mock_send.called
|
|
|
|
@patch("app.telemetry.service._send_otlp_event")
|
|
def test_ping_forwards_extra_data(self, mock_send):
|
|
"""Test that telemetry ping forwards custom event data."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_TELEMETRY": "true",
|
|
"OTEL_EXPORTER_OTLP_ENDPOINT": "https://otlp.example.com",
|
|
"OTEL_EXPORTER_OTLP_TOKEN": "test-token",
|
|
"APP_VERSION": "1.0.0",
|
|
"TELE_SALT": "test-salt",
|
|
},
|
|
):
|
|
send_telemetry_ping("install", extra_data={"test": "value"})
|
|
|
|
call_args = mock_send.call_args
|
|
properties = call_args[1]["properties"]
|
|
assert properties["test"] == "value"
|
|
|
|
@patch("app.telemetry.service._send_otlp_event")
|
|
def test_ping_handles_network_errors_gracefully(self, mock_send):
|
|
"""Test that network errors don't crash the application"""
|
|
mock_send.side_effect = Exception("Network error")
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLE_TELEMETRY": "true",
|
|
"OTEL_EXPORTER_OTLP_ENDPOINT": "https://otlp.example.com",
|
|
"OTEL_EXPORTER_OTLP_TOKEN": "test-token",
|
|
},
|
|
):
|
|
result = send_telemetry_ping("install")
|
|
assert result is False
|
|
|
|
|
|
class TestTelemetryEventTypes:
|
|
"""Tests for different telemetry event types"""
|
|
|
|
@patch("app.utils.telemetry.send_telemetry_ping")
|
|
def test_send_install_ping(self, mock_send):
|
|
"""Test sending install ping"""
|
|
send_install_ping()
|
|
mock_send.assert_called_once_with(event_type="install")
|
|
|
|
@patch("app.utils.telemetry.send_telemetry_ping")
|
|
def test_send_update_ping(self, mock_send):
|
|
"""Test sending update ping"""
|
|
send_update_ping("1.0.0", "1.1.0")
|
|
mock_send.assert_called_once()
|
|
call_args = mock_send.call_args
|
|
assert call_args[1]["event_type"] == "update"
|
|
assert call_args[1]["extra_data"]["old_version"] == "1.0.0"
|
|
assert call_args[1]["extra_data"]["new_version"] == "1.1.0"
|
|
|
|
@patch("app.utils.telemetry.send_telemetry_ping")
|
|
def test_send_health_ping(self, mock_send):
|
|
"""Test sending health ping"""
|
|
send_health_ping()
|
|
mock_send.assert_called_once_with(event_type="health")
|
|
|
|
|
|
class TestTelemetryMarker:
|
|
"""Tests for telemetry marker file functionality"""
|
|
|
|
def test_should_send_when_no_marker(self):
|
|
"""Test that telemetry should be sent when marker doesn't exist"""
|
|
with tempfile.NamedTemporaryFile(delete=True) as tmp:
|
|
marker_path = tmp.name + "_nonexistent"
|
|
with patch.dict(os.environ, {"ENABLE_TELEMETRY": "true"}):
|
|
assert should_send_telemetry(marker_path) is True
|
|
|
|
def test_should_not_send_when_marker_exists(self):
|
|
"""Test that telemetry shouldn't be sent when marker exists"""
|
|
with tempfile.NamedTemporaryFile(delete=False) as tmp:
|
|
marker_path = tmp.name
|
|
try:
|
|
with patch.dict(os.environ, {"ENABLE_TELEMETRY": "true"}):
|
|
assert should_send_telemetry(marker_path) is False
|
|
finally:
|
|
try:
|
|
os.unlink(marker_path)
|
|
except (PermissionError, OSError):
|
|
pass # Ignore Windows file permission errors
|
|
|
|
def test_mark_telemetry_sent_creates_file(self):
|
|
"""Test that marking telemetry as sent creates marker file"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
marker_path = os.path.join(tmpdir, "test_marker")
|
|
with patch.dict(os.environ, {"APP_VERSION": "1.0.0"}):
|
|
mark_telemetry_sent(marker_path)
|
|
assert os.path.exists(marker_path)
|
|
|
|
# Verify file contents
|
|
with open(marker_path, "r") as f:
|
|
data = json.load(f)
|
|
assert "version" in data
|
|
assert "fingerprint" in data
|
|
|
|
|
|
class TestCheckAndSendTelemetry:
|
|
"""Tests for the convenience function"""
|
|
|
|
@patch("app.utils.telemetry.send_install_ping")
|
|
@patch("app.utils.telemetry.mark_telemetry_sent")
|
|
def test_check_and_send_when_appropriate(self, mock_mark, mock_send):
|
|
"""Test that telemetry is sent and marked when appropriate"""
|
|
mock_send.return_value = True
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
marker_path = os.path.join(tmpdir, "telemetry_sent")
|
|
with patch.dict(os.environ, {"ENABLE_TELEMETRY": "true", "TELEMETRY_MARKER_FILE": marker_path}):
|
|
result = check_and_send_telemetry()
|
|
assert result is True
|
|
mock_send.assert_called_once()
|
|
mock_mark.assert_called_once()
|
|
|
|
@patch("app.utils.telemetry.send_install_ping")
|
|
def test_no_send_when_disabled(self, mock_send):
|
|
"""Test that telemetry is not sent when disabled"""
|
|
# Mock installation config to force fallback to environment variable
|
|
with patch("app.utils.telemetry.get_installation_config") as mock_config:
|
|
mock_config.side_effect = Exception("Force fallback to env var")
|
|
with patch.dict(os.environ, {"ENABLE_TELEMETRY": "false"}):
|
|
result = check_and_send_telemetry()
|
|
assert result is False
|
|
assert not mock_send.called
|
|
|
|
@patch("app.utils.telemetry.send_install_ping")
|
|
def test_no_send_when_already_sent(self, mock_send):
|
|
"""Test that telemetry is not sent when already marked as sent"""
|
|
with tempfile.NamedTemporaryFile(delete=False) as tmp:
|
|
marker_path = tmp.name
|
|
try:
|
|
with patch.dict(os.environ, {"ENABLE_TELEMETRY": "true", "TELEMETRY_MARKER_FILE": marker_path}):
|
|
result = check_and_send_telemetry()
|
|
assert result is False
|
|
assert not mock_send.called
|
|
finally:
|
|
try:
|
|
os.unlink(marker_path)
|
|
except (PermissionError, OSError):
|
|
pass # Ignore Windows file permission errors
|