Jason Woltje
f02207e33c
Standalone Python package (mosaicstack-telemetry) for reporting task-completion telemetry and querying predictions from the Mosaic Stack Telemetry server. - Sync/async TelemetryClient with context manager support - Thread-safe EventQueue with bounded deque - BatchSubmitter with httpx, exponential backoff, Retry-After - PredictionCache with TTL - EventBuilder convenience class - All types standalone (no server dependency) - 55 tests, 90% coverage, mypy strict clean Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
"""Tests for EventBuilder."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID
|
|
|
|
from mosaicstack_telemetry.event_builder import EventBuilder
|
|
from mosaicstack_telemetry.types.events import (
|
|
Complexity,
|
|
Harness,
|
|
Outcome,
|
|
Provider,
|
|
QualityGate,
|
|
RepoSizeCategory,
|
|
TaskType,
|
|
)
|
|
|
|
TEST_INSTANCE_ID = "12345678-1234-1234-1234-123456789abc"
|
|
|
|
|
|
class TestEventBuilder:
|
|
"""Tests for the fluent event builder."""
|
|
|
|
def test_build_complete_event(self) -> None:
|
|
"""Build an event with all fields set."""
|
|
event = (
|
|
EventBuilder(instance_id=TEST_INSTANCE_ID)
|
|
.task_type(TaskType.IMPLEMENTATION)
|
|
.model("claude-sonnet-4-20250514")
|
|
.provider(Provider.ANTHROPIC)
|
|
.harness_type(Harness.CLAUDE_CODE)
|
|
.complexity_level(Complexity.HIGH)
|
|
.outcome_value(Outcome.SUCCESS)
|
|
.duration_ms(45000)
|
|
.tokens(estimated_in=5000, estimated_out=2000, actual_in=5200, actual_out=1800)
|
|
.cost(estimated=50000, actual=48000)
|
|
.quality(
|
|
passed=True,
|
|
gates_run=[QualityGate.LINT, QualityGate.TEST],
|
|
gates_failed=[],
|
|
)
|
|
.context(compactions=1, rotations=0, utilization=0.4)
|
|
.retry_count(0)
|
|
.language("python")
|
|
.repo_size(RepoSizeCategory.MEDIUM)
|
|
.build()
|
|
)
|
|
|
|
assert event.instance_id == UUID(TEST_INSTANCE_ID)
|
|
assert event.task_type == TaskType.IMPLEMENTATION
|
|
assert event.model == "claude-sonnet-4-20250514"
|
|
assert event.provider == Provider.ANTHROPIC
|
|
assert event.harness == Harness.CLAUDE_CODE
|
|
assert event.complexity == Complexity.HIGH
|
|
assert event.outcome == Outcome.SUCCESS
|
|
assert event.task_duration_ms == 45000
|
|
assert event.estimated_input_tokens == 5000
|
|
assert event.estimated_output_tokens == 2000
|
|
assert event.actual_input_tokens == 5200
|
|
assert event.actual_output_tokens == 1800
|
|
assert event.estimated_cost_usd_micros == 50000
|
|
assert event.actual_cost_usd_micros == 48000
|
|
assert event.quality_gate_passed is True
|
|
assert event.quality_gates_run == [QualityGate.LINT, QualityGate.TEST]
|
|
assert event.quality_gates_failed == []
|
|
assert event.context_compactions == 1
|
|
assert event.context_rotations == 0
|
|
assert event.context_utilization_final == 0.4
|
|
assert event.retry_count == 0
|
|
assert event.language == "python"
|
|
assert event.repo_size_category == RepoSizeCategory.MEDIUM
|
|
|
|
def test_auto_generated_defaults(self) -> None:
|
|
"""event_id and timestamp are auto-generated."""
|
|
event = (
|
|
EventBuilder(instance_id=TEST_INSTANCE_ID)
|
|
.task_type(TaskType.DEBUGGING)
|
|
.model("gpt-4o")
|
|
.provider(Provider.OPENAI)
|
|
.build()
|
|
)
|
|
|
|
assert event.event_id is not None
|
|
assert event.timestamp is not None
|
|
assert event.timestamp.tzinfo is not None
|
|
|
|
def test_custom_event_id(self) -> None:
|
|
"""Custom event_id can be set."""
|
|
custom_id = "abcdef12-1234-1234-1234-123456789abc"
|
|
event = (
|
|
EventBuilder(instance_id=TEST_INSTANCE_ID)
|
|
.event_id(custom_id)
|
|
.model("test-model")
|
|
.build()
|
|
)
|
|
|
|
assert event.event_id == UUID(custom_id)
|
|
|
|
def test_custom_timestamp(self) -> None:
|
|
"""Custom timestamp can be set."""
|
|
ts = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
event = EventBuilder(instance_id=TEST_INSTANCE_ID).timestamp(ts).model("test-model").build()
|
|
|
|
assert event.timestamp == ts
|
|
|
|
def test_minimal_event_defaults(self) -> None:
|
|
"""Minimal event has sensible defaults."""
|
|
event = EventBuilder(instance_id=TEST_INSTANCE_ID).model("test-model").build()
|
|
|
|
assert event.task_type == TaskType.UNKNOWN
|
|
assert event.complexity == Complexity.MEDIUM
|
|
assert event.harness == Harness.UNKNOWN
|
|
assert event.provider == Provider.UNKNOWN
|
|
assert event.outcome == Outcome.FAILURE
|
|
assert event.task_duration_ms == 0
|
|
assert event.retry_count == 0
|
|
assert event.language is None
|
|
assert event.repo_size_category is None
|
|
|
|
def test_quality_defaults_to_empty_lists(self) -> None:
|
|
"""Quality gate lists default to empty."""
|
|
event = EventBuilder(instance_id=TEST_INSTANCE_ID).model("m").build()
|
|
|
|
assert event.quality_gates_run == []
|
|
assert event.quality_gates_failed == []
|
|
assert event.quality_gate_passed is False
|
|
|
|
def test_schema_version(self) -> None:
|
|
"""Schema version defaults to 1.0."""
|
|
event = EventBuilder(instance_id=TEST_INSTANCE_ID).model("m").build()
|
|
assert event.schema_version == "1.0"
|