feat(#370): install mosaicstack-telemetry in Coordinator
- Add mosaicstack-telemetry>=0.1.0 to pyproject.toml dependencies - Configure Gitea PyPI registry via pip.conf (extra-index-url) - Integrate TelemetryClient in FastAPI lifespan (start_async/stop_async) - Store client on app.state.mosaic_telemetry for downstream access - Create mosaic_telemetry.py helper module with: - get_telemetry_client(): retrieve client from app state - build_task_event(): construct TaskCompletionEvent with coordinator defaults - create_telemetry_config(): create config from MOSAIC_TELEMETRY_* env vars - Add 28 unit tests covering config, helpers, disabled mode, and lifespan - New module has 100% test coverage Refs #370 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
426
apps/coordinator/tests/test_mosaic_telemetry.py
Normal file
426
apps/coordinator/tests/test_mosaic_telemetry.py
Normal file
@@ -0,0 +1,426 @@
|
||||
"""Tests for Mosaic Stack telemetry integration (mosaic_telemetry module).
|
||||
|
||||
These tests cover the mosaicstack-telemetry SDK integration, NOT the
|
||||
OpenTelemetry distributed tracing (which is tested in test_telemetry.py).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from mosaicstack_telemetry import (
|
||||
Complexity,
|
||||
Harness,
|
||||
Outcome,
|
||||
Provider,
|
||||
QualityGate,
|
||||
TaskCompletionEvent,
|
||||
TaskType,
|
||||
TelemetryClient,
|
||||
TelemetryConfig,
|
||||
)
|
||||
|
||||
from src.mosaic_telemetry import (
|
||||
build_task_event,
|
||||
create_telemetry_config,
|
||||
get_telemetry_client,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TelemetryConfig creation from environment variables
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCreateTelemetryConfig:
|
||||
"""Tests for create_telemetry_config helper."""
|
||||
|
||||
def test_config_reads_enabled_from_env(self) -> None:
|
||||
"""TelemetryConfig should read MOSAIC_TELEMETRY_ENABLED from env."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_ENABLED": "true"},
|
||||
clear=False,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
assert config.enabled is True
|
||||
|
||||
def test_config_disabled_from_env(self) -> None:
|
||||
"""TelemetryConfig should be disabled when env var is false."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_ENABLED": "false"},
|
||||
clear=False,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
assert config.enabled is False
|
||||
|
||||
def test_config_reads_server_url_from_env(self) -> None:
|
||||
"""TelemetryConfig should read MOSAIC_TELEMETRY_SERVER_URL from env."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_SERVER_URL": "https://telemetry.example.com"},
|
||||
clear=False,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
assert config.server_url == "https://telemetry.example.com"
|
||||
|
||||
def test_config_reads_api_key_from_env(self) -> None:
|
||||
"""TelemetryConfig should read MOSAIC_TELEMETRY_API_KEY from env."""
|
||||
api_key = "a" * 64 # 64-char hex string
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_API_KEY": api_key},
|
||||
clear=False,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
assert config.api_key == api_key
|
||||
|
||||
def test_config_reads_instance_id_from_env(self) -> None:
|
||||
"""TelemetryConfig should read MOSAIC_TELEMETRY_INSTANCE_ID from env."""
|
||||
instance_id = "12345678-1234-1234-1234-123456789abc"
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_INSTANCE_ID": instance_id},
|
||||
clear=False,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
assert config.instance_id == instance_id
|
||||
|
||||
def test_config_defaults_to_enabled(self) -> None:
|
||||
"""TelemetryConfig should default to enabled when env var is not set."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{},
|
||||
clear=True,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
assert config.enabled is True
|
||||
|
||||
def test_config_logs_validation_warnings_when_enabled(self) -> None:
|
||||
"""Config creation should log warnings for validation errors when enabled."""
|
||||
with (
|
||||
patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_ENABLED": "true"},
|
||||
clear=True,
|
||||
),
|
||||
patch("src.mosaic_telemetry.logger") as mock_logger,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
# server_url, api_key, and instance_id are all empty = validation errors
|
||||
assert config.enabled is True
|
||||
mock_logger.warning.assert_called_once()
|
||||
warning_msg = mock_logger.warning.call_args[0][0]
|
||||
assert "validation issues" in warning_msg
|
||||
|
||||
def test_config_no_warnings_when_disabled(self) -> None:
|
||||
"""Config creation should not log warnings when telemetry is disabled."""
|
||||
with (
|
||||
patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_ENABLED": "false"},
|
||||
clear=True,
|
||||
),
|
||||
patch("src.mosaic_telemetry.logger") as mock_logger,
|
||||
):
|
||||
create_telemetry_config()
|
||||
mock_logger.warning.assert_not_called()
|
||||
|
||||
def test_config_strips_trailing_slashes(self) -> None:
|
||||
"""TelemetryConfig should strip trailing slashes from server_url."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{"MOSAIC_TELEMETRY_SERVER_URL": "https://telemetry.example.com/"},
|
||||
clear=False,
|
||||
):
|
||||
config = create_telemetry_config()
|
||||
assert config.server_url == "https://telemetry.example.com"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_telemetry_client from app state
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetTelemetryClient:
|
||||
"""Tests for get_telemetry_client helper."""
|
||||
|
||||
def test_returns_client_when_set(self) -> None:
|
||||
"""Should return the telemetry client from app state."""
|
||||
app = FastAPI()
|
||||
mock_client = MagicMock(spec=TelemetryClient)
|
||||
app.state.mosaic_telemetry = mock_client
|
||||
|
||||
result = get_telemetry_client(app)
|
||||
assert result is mock_client
|
||||
|
||||
def test_returns_none_when_not_set(self) -> None:
|
||||
"""Should return None when mosaic_telemetry is not in app state."""
|
||||
app = FastAPI()
|
||||
# Do not set app.state.mosaic_telemetry
|
||||
|
||||
result = get_telemetry_client(app)
|
||||
assert result is None
|
||||
|
||||
def test_returns_none_when_explicitly_none(self) -> None:
|
||||
"""Should return None when mosaic_telemetry is explicitly set to None."""
|
||||
app = FastAPI()
|
||||
app.state.mosaic_telemetry = None
|
||||
|
||||
result = get_telemetry_client(app)
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# build_task_event helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildTaskEvent:
|
||||
"""Tests for build_task_event helper."""
|
||||
|
||||
VALID_INSTANCE_ID = "12345678-1234-1234-1234-123456789abc"
|
||||
|
||||
def test_builds_event_with_defaults(self) -> None:
|
||||
"""Should build a TaskCompletionEvent with default values."""
|
||||
event = build_task_event(instance_id=self.VALID_INSTANCE_ID)
|
||||
|
||||
assert isinstance(event, TaskCompletionEvent)
|
||||
assert str(event.instance_id) == self.VALID_INSTANCE_ID
|
||||
assert event.task_type == TaskType.IMPLEMENTATION
|
||||
assert event.complexity == Complexity.MEDIUM
|
||||
assert event.outcome == Outcome.SUCCESS
|
||||
assert event.harness == Harness.CLAUDE_CODE
|
||||
assert event.provider == Provider.ANTHROPIC
|
||||
assert event.language == "typescript"
|
||||
|
||||
def test_builds_event_with_custom_task_type(self) -> None:
|
||||
"""Should respect custom task_type parameter."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
task_type=TaskType.TESTING,
|
||||
)
|
||||
assert event.task_type == TaskType.TESTING
|
||||
|
||||
def test_builds_event_with_custom_outcome(self) -> None:
|
||||
"""Should respect custom outcome parameter."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
outcome=Outcome.FAILURE,
|
||||
)
|
||||
assert event.outcome == Outcome.FAILURE
|
||||
|
||||
def test_builds_event_with_duration(self) -> None:
|
||||
"""Should set duration_ms correctly."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
duration_ms=45000,
|
||||
)
|
||||
assert event.task_duration_ms == 45000
|
||||
|
||||
def test_builds_event_with_token_counts(self) -> None:
|
||||
"""Should set all token counts correctly."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
estimated_input_tokens=1000,
|
||||
estimated_output_tokens=500,
|
||||
actual_input_tokens=1100,
|
||||
actual_output_tokens=480,
|
||||
)
|
||||
assert event.estimated_input_tokens == 1000
|
||||
assert event.estimated_output_tokens == 500
|
||||
assert event.actual_input_tokens == 1100
|
||||
assert event.actual_output_tokens == 480
|
||||
|
||||
def test_builds_event_with_cost(self) -> None:
|
||||
"""Should set cost fields correctly."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
estimated_cost_micros=50000,
|
||||
actual_cost_micros=48000,
|
||||
)
|
||||
assert event.estimated_cost_usd_micros == 50000
|
||||
assert event.actual_cost_usd_micros == 48000
|
||||
|
||||
def test_builds_event_with_quality_gates(self) -> None:
|
||||
"""Should set quality gate information correctly."""
|
||||
gates_run = [QualityGate.LINT, QualityGate.TEST, QualityGate.BUILD]
|
||||
gates_failed = [QualityGate.TEST]
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
quality_passed=False,
|
||||
quality_gates_run=gates_run,
|
||||
quality_gates_failed=gates_failed,
|
||||
)
|
||||
assert event.quality_gate_passed is False
|
||||
assert event.quality_gates_run == gates_run
|
||||
assert event.quality_gates_failed == gates_failed
|
||||
|
||||
def test_builds_event_with_context_info(self) -> None:
|
||||
"""Should set context compaction/rotation/utilization correctly."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
context_compactions=2,
|
||||
context_rotations=1,
|
||||
context_utilization=0.75,
|
||||
)
|
||||
assert event.context_compactions == 2
|
||||
assert event.context_rotations == 1
|
||||
assert event.context_utilization_final == 0.75
|
||||
|
||||
def test_builds_event_with_retry_count(self) -> None:
|
||||
"""Should set retry count correctly."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
retry_count=3,
|
||||
)
|
||||
assert event.retry_count == 3
|
||||
|
||||
def test_builds_event_with_custom_language(self) -> None:
|
||||
"""Should allow overriding the default language."""
|
||||
event = build_task_event(
|
||||
instance_id=self.VALID_INSTANCE_ID,
|
||||
language="python",
|
||||
)
|
||||
assert event.language == "python"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TelemetryClient lifecycle (disabled mode)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTelemetryDisabledMode:
|
||||
"""Tests for disabled telemetry mode (no HTTP calls)."""
|
||||
|
||||
def test_disabled_client_does_not_start(self) -> None:
|
||||
"""Client start_async should be a no-op when disabled."""
|
||||
config = TelemetryConfig(enabled=False)
|
||||
client = TelemetryClient(config)
|
||||
# Should not raise
|
||||
assert client.is_running is False
|
||||
|
||||
def test_disabled_client_track_is_noop(self) -> None:
|
||||
"""Tracking events when disabled should silently drop them."""
|
||||
config = TelemetryConfig(enabled=False)
|
||||
client = TelemetryClient(config)
|
||||
|
||||
event = build_task_event(
|
||||
instance_id="12345678-1234-1234-1234-123456789abc",
|
||||
)
|
||||
# Should not raise, should silently drop
|
||||
client.track(event)
|
||||
assert client.queue_size == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_disabled_client_start_stop_async(self) -> None:
|
||||
"""Async start/stop should be safe when disabled."""
|
||||
config = TelemetryConfig(enabled=False)
|
||||
client = TelemetryClient(config)
|
||||
|
||||
await client.start_async()
|
||||
assert client.is_running is False
|
||||
await client.stop_async()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lifespan integration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestLifespanIntegration:
|
||||
"""Tests for Mosaic telemetry in the FastAPI lifespan."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lifespan_sets_mosaic_telemetry_on_app_state(self) -> None:
|
||||
"""Lifespan should store mosaic_telemetry client on app.state."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"GITEA_WEBHOOK_SECRET": "test-secret",
|
||||
"GITEA_URL": "https://git.mosaicstack.dev",
|
||||
"ANTHROPIC_API_KEY": "test-key",
|
||||
"MOSAIC_TELEMETRY_ENABLED": "true",
|
||||
"MOSAIC_TELEMETRY_SERVER_URL": "https://telemetry.example.com",
|
||||
"MOSAIC_TELEMETRY_API_KEY": "a" * 64,
|
||||
"MOSAIC_TELEMETRY_INSTANCE_ID": "12345678-1234-1234-1234-123456789abc",
|
||||
"OTEL_ENABLED": "false",
|
||||
"COORDINATOR_ENABLED": "false",
|
||||
},
|
||||
):
|
||||
# Reload config to pick up test env vars
|
||||
import importlib
|
||||
|
||||
from src import config
|
||||
importlib.reload(config)
|
||||
|
||||
from src.main import lifespan
|
||||
|
||||
app = FastAPI()
|
||||
async with lifespan(app) as _state:
|
||||
client = getattr(app.state, "mosaic_telemetry", None)
|
||||
assert client is not None
|
||||
assert isinstance(client, TelemetryClient)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lifespan_sets_none_when_disabled(self) -> None:
|
||||
"""Lifespan should set mosaic_telemetry to None when disabled."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"GITEA_WEBHOOK_SECRET": "test-secret",
|
||||
"GITEA_URL": "https://git.mosaicstack.dev",
|
||||
"ANTHROPIC_API_KEY": "test-key",
|
||||
"MOSAIC_TELEMETRY_ENABLED": "false",
|
||||
"OTEL_ENABLED": "false",
|
||||
"COORDINATOR_ENABLED": "false",
|
||||
},
|
||||
):
|
||||
import importlib
|
||||
|
||||
from src import config
|
||||
importlib.reload(config)
|
||||
|
||||
from src.main import lifespan
|
||||
|
||||
app = FastAPI()
|
||||
async with lifespan(app) as _state:
|
||||
client = getattr(app.state, "mosaic_telemetry", None)
|
||||
assert client is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lifespan_stops_client_on_shutdown(self) -> None:
|
||||
"""Lifespan should call stop_async on shutdown."""
|
||||
with patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"GITEA_WEBHOOK_SECRET": "test-secret",
|
||||
"GITEA_URL": "https://git.mosaicstack.dev",
|
||||
"ANTHROPIC_API_KEY": "test-key",
|
||||
"MOSAIC_TELEMETRY_ENABLED": "true",
|
||||
"MOSAIC_TELEMETRY_SERVER_URL": "https://telemetry.example.com",
|
||||
"MOSAIC_TELEMETRY_API_KEY": "a" * 64,
|
||||
"MOSAIC_TELEMETRY_INSTANCE_ID": "12345678-1234-1234-1234-123456789abc",
|
||||
"OTEL_ENABLED": "false",
|
||||
"COORDINATOR_ENABLED": "false",
|
||||
},
|
||||
):
|
||||
import importlib
|
||||
|
||||
from src import config
|
||||
importlib.reload(config)
|
||||
|
||||
from src.main import lifespan
|
||||
|
||||
app = FastAPI()
|
||||
async with lifespan(app) as _state:
|
||||
client = app.state.mosaic_telemetry
|
||||
assert isinstance(client, TelemetryClient)
|
||||
# Client was started
|
||||
# After context manager exits, stop_async should have been called
|
||||
|
||||
# After lifespan exits, client should no longer be running
|
||||
# (stop_async was called in the shutdown section)
|
||||
assert not client.is_running
|
||||
Reference in New Issue
Block a user