mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-03 02:22:24 +00:00
refactor(tickerr): slim to 103-line dumb pipe, server owns all normalization
- Remove _PROVIDER_MAP (25 entries), _ERROR_TYPE_MAP, _normalize_provider, _get_provider, _get_status_code, _latency_ms, _fire_and_forget, _parse_sample_rate - Provider read directly from custom_llm_provider, passed as-is - Model passed as-is, no prefix stripping - status_code sent raw, server derives error_type - Inline fire-and-forget into _report, remove semaphore - Add TICKERR_DISABLED kill switch - Add opt-in success sampling (TICKERR_SAMPLE_RATE, default 0) - event_type: "failure"/"success" replaces is_resolution boolean - Remove client_tier (trust concern for first merge) - Tests import only TickerrLogger, no internal symbols - 103 lines source, 19 tests, 55 lines docs
This commit is contained in:
Vikas Sharma
parent
e279076193
commit
51a67bb2a0
4 changed files with 458 additions and 1200 deletions
|
|
@ -1,15 +1,11 @@
|
|||
import Tabs from '@theme/Tabs';
|
||||
import TabItem from '@theme/TabItem';
|
||||
|
||||
# Tickerr — Outage Radar for AI Agents
|
||||
# Tickerr - Outage Radar for AI Agents
|
||||
|
||||
[Tickerr](https://tickerr.ai) is a crowd-sourced outage detector for LLM APIs. When your agent hits a 5xx error or rate limit, Tickerr tells you:
|
||||
[Tickerr](https://tickerr.ai) is a crowd-sourced outage detector for LLM APIs. When your agent hits a 5xx or rate limit, it reports anonymously so every agent can see the issue in real time.
|
||||
|
||||
- How many other agents are seeing the same issue right now
|
||||
- Current signal state: `quiet` → `detecting` → `confirmed` → `recovering`
|
||||
- Which model to fall back to
|
||||
|
||||
**No API key required. Anonymous. Zero overhead on success paths.**
|
||||
**No API key. No account. Failure-only by default. Success sampling is opt-in.**
|
||||
|
||||
## Quick Start
|
||||
|
||||
|
|
@ -21,10 +17,8 @@ import litellm
|
|||
|
||||
litellm.callbacks = ["tickerr"]
|
||||
|
||||
# Every failed LiteLLM call is now reported to Tickerr automatically.
|
||||
# Tickerr fires in a background thread — your agent is never blocked.
|
||||
response = litellm.completion(
|
||||
model="claude-haiku-4-5",
|
||||
model="gpt-4o-mini",
|
||||
messages=[{"role": "user", "content": "Hello"}]
|
||||
)
|
||||
```
|
||||
|
|
@ -33,12 +27,6 @@ response = litellm.completion(
|
|||
<TabItem value="proxy" label="LiteLLM Proxy config.yaml">
|
||||
|
||||
```yaml
|
||||
model_list:
|
||||
- model_name: claude-haiku
|
||||
litellm_params:
|
||||
model: anthropic/claude-haiku-4-5
|
||||
api_key: os.environ/ANTHROPIC_API_KEY
|
||||
|
||||
litellm_settings:
|
||||
callbacks: ["tickerr"]
|
||||
```
|
||||
|
|
@ -48,60 +36,28 @@ litellm_settings:
|
|||
|
||||
## What Gets Reported
|
||||
|
||||
Tickerr receives only:
|
||||
When the tickerr callback is explicitly enabled, anonymous failure metadata is reported. No prompts, responses, API keys, or personal data are sent.
|
||||
|
||||
| Field | Example |
|
||||
|-------|---------|
|
||||
| Provider | `anthropic` |
|
||||
| Model | `claude-haiku-4-5` |
|
||||
| HTTP status code | `529` |
|
||||
| Error type | `overloaded` |
|
||||
| Status code | `529` |
|
||||
| Latency (ms) | `1240` |
|
||||
|
||||
No prompts, no responses, no personal data.
|
||||
|
||||
## What You Get Back
|
||||
|
||||
Each report updates the live signal at [tickerr.ai/agent-reports](https://tickerr.ai/agent-reports).
|
||||
|
||||
To read the signal from your agent, use the [Tickerr MCP server](https://tickerr.ai/mcp-server) `report_incident` tool. It returns a structured response:
|
||||
|
||||
```
|
||||
CURRENT SIGNAL (anthropic/claude-haiku-4-5)
|
||||
Status: CONFIRMED
|
||||
Agents reporting (last 10 min): 14
|
||||
Total reports (last 10 min): 31
|
||||
|
||||
RECOMMENDATION
|
||||
Action: FALLBACK
|
||||
Switch to: gpt-4o-mini (openai)
|
||||
```
|
||||
| Event type | `failure` or `success` |
|
||||
|
||||
## Optional Configuration
|
||||
|
||||
```python
|
||||
import os
|
||||
|
||||
os.environ["TICKERR_CLIENT_TIER"] = "pro" # free | pro | team | enterprise | api_pay_as_you_go
|
||||
os.environ["TICKERR_REGION"] = "us-east-1" # optional, for regional breakdown
|
||||
```bash
|
||||
TICKERR_DISABLED=true # disable all reporting (kill switch)
|
||||
TICKERR_REGION=us-east-1 # for regional signal breakdown
|
||||
TICKERR_SAMPLE_RATE=0.01 # report 1% of successes for latency benchmarks (default: 0 = off)
|
||||
```
|
||||
|
||||
## Signal States
|
||||
|
||||
| State | Meaning |
|
||||
|-------|---------|
|
||||
| `quiet` | No reports in last 10 min |
|
||||
| `detecting` | 1–2 agents reporting |
|
||||
| `confirmed` | 3+ distinct agents — issue verified |
|
||||
| `recovering` | Reports dropping, recovery signals arriving |
|
||||
|
||||
## Opt Out
|
||||
|
||||
[tickerr.ai/mcp/opt-out](https://tickerr.ai/mcp/opt-out)
|
||||
Failures are reported by default once Tickerr is enabled. Success sampling is opt-in.
|
||||
|
||||
## Links
|
||||
|
||||
- [Tickerr](https://tickerr.ai) — live AI status dashboard (90+ tools)
|
||||
- [Agent reports](https://tickerr.ai/agent-reports) — live feed
|
||||
- [Tickerr MCP server](https://tickerr.ai/mcp-server) — 9-tool MCP for agents
|
||||
- [REST API](https://tickerr.ai/api/v1/report) — report without LiteLLM
|
||||
- [Live dashboard](https://tickerr.ai) - status for 90+ AI tools
|
||||
- [Agent reports feed](https://tickerr.ai/agent-reports) - real-time signal
|
||||
- [Opt out](https://tickerr.ai/mcp/opt-out)
|
||||
|
|
|
|||
|
|
@ -1,23 +1,23 @@
|
|||
"""
|
||||
Tickerr — outage radar for AI agents.
|
||||
Tickerr - crowd-sourced outage radar for AI agents.
|
||||
|
||||
Reports LLM API failures to https://tickerr.ai so agents can
|
||||
see how many other agents are hitting the same issue and get
|
||||
a live routing recommendation (RETRY / RETRY_WITH_DELAY / FALLBACK).
|
||||
|
||||
Zero dependencies beyond stdlib. Anonymous. Non-blocking.
|
||||
Reports LLM API failures to https://tickerr.ai so every agent
|
||||
can see when a provider is down and which model to fall back to.
|
||||
|
||||
Usage:
|
||||
litellm.callbacks = ["tickerr"]
|
||||
|
||||
No API key required.
|
||||
No API key. No account. Failure-only by default. Success sampling is opt-in.
|
||||
Zero dependencies beyond stdlib.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import random
|
||||
import threading
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, Optional, Union
|
||||
|
||||
|
|
@ -26,213 +26,78 @@ from litellm.integrations.custom_logger import CustomLogger
|
|||
_REPORT_URL = "https://tickerr.ai/api/v1/report"
|
||||
_UA = "litellm-tickerr/1.0"
|
||||
|
||||
# Cap concurrent in-flight reports to avoid thread exhaustion on burst failures
|
||||
_MAX_INFLIGHT = 5
|
||||
_inflight = threading.Semaphore(_MAX_INFLIGHT)
|
||||
|
||||
# Map litellm custom_llm_provider → Tickerr provider slug
|
||||
_PROVIDER_MAP: Dict[str, str] = {
|
||||
"openai": "openai",
|
||||
"anthropic": "anthropic",
|
||||
"google": "google",
|
||||
"vertex_ai": "google",
|
||||
"gemini": "google",
|
||||
"cohere": "cohere",
|
||||
"mistral": "mistral",
|
||||
"groq": "groq",
|
||||
"together_ai": "together",
|
||||
"huggingface": "huggingface",
|
||||
"replicate": "replicate",
|
||||
"deepinfra": "deepinfra",
|
||||
"perplexity": "perplexity",
|
||||
"fireworks_ai": "fireworks",
|
||||
"openrouter": "openrouter",
|
||||
"azure": "azure",
|
||||
"azure_ai": "azure",
|
||||
"bedrock": "aws",
|
||||
"bedrock_converse": "aws",
|
||||
"vertex_ai_anthropic": "anthropic",
|
||||
"ai21": "ai21",
|
||||
"cerebras": "cerebras",
|
||||
"xai": "xai",
|
||||
"deepseek": "deepseek",
|
||||
"ollama": "ollama",
|
||||
"nlp_cloud": "nlp_cloud",
|
||||
}
|
||||
|
||||
# Only map codes that unambiguously indicate the specific error type.
|
||||
# 500 is intentionally excluded: it is a generic "Internal Server Error" that
|
||||
# indicates a crash or bug, not a capacity/overload condition.
|
||||
_ERROR_TYPE_MAP: Dict[int, str] = {
|
||||
429: "rate_limit",
|
||||
529: "overloaded",
|
||||
503: "overloaded",
|
||||
408: "timeout",
|
||||
524: "timeout",
|
||||
401: "auth",
|
||||
403: "auth",
|
||||
}
|
||||
|
||||
|
||||
def _normalize_provider(model: str, kwargs: Dict[str, Any]) -> str:
|
||||
custom = (
|
||||
kwargs.get("litellm_params", {}).get("custom_llm_provider")
|
||||
or kwargs.get("custom_llm_provider")
|
||||
or ""
|
||||
)
|
||||
if custom:
|
||||
return _PROVIDER_MAP.get(custom.lower(), custom.lower())
|
||||
if "/" in model:
|
||||
prefix = model.split("/")[0].lower()
|
||||
return _PROVIDER_MAP.get(prefix, prefix)
|
||||
if re.match(r"^claude", model, re.I):
|
||||
return "anthropic"
|
||||
if re.match(r"^gpt|^o[1-9]", model, re.I):
|
||||
return "openai"
|
||||
if re.match(r"^gemini", model, re.I):
|
||||
return "google"
|
||||
if re.match(r"^mistral|^mixtral", model, re.I):
|
||||
return "mistral"
|
||||
if re.match(r"^llama", model, re.I):
|
||||
return "meta"
|
||||
if re.match(r"^command", model, re.I):
|
||||
return "cohere"
|
||||
if re.match(r"^grok", model, re.I):
|
||||
return "xai"
|
||||
if re.match(r"^deepseek", model, re.I):
|
||||
return "deepseek"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _extract_status_code(exception: Optional[BaseException]) -> Optional[int]:
|
||||
if exception is None:
|
||||
return None
|
||||
code = getattr(exception, "status_code", None)
|
||||
if isinstance(code, int):
|
||||
return code
|
||||
if isinstance(code, str) and code.isdigit():
|
||||
return int(code)
|
||||
return None
|
||||
|
||||
|
||||
def _latency_ms(
|
||||
start_time: Union[datetime, float], end_time: Union[datetime, float]
|
||||
) -> int:
|
||||
"""Compute elapsed ms whether LiteLLM passes datetime objects or floats."""
|
||||
if isinstance(start_time, datetime) and isinstance(end_time, datetime):
|
||||
return round((end_time - start_time).total_seconds() * 1000)
|
||||
return round((float(end_time) - float(start_time)) * 1000) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def _fire_and_forget(payload: Dict[str, Any]) -> None:
|
||||
"""POST to Tickerr in a daemon thread — never blocks the caller.
|
||||
|
||||
A semaphore caps concurrent in-flight threads so a burst of failures
|
||||
(e.g. 100 errors/s) cannot exhaust the thread pool.
|
||||
"""
|
||||
if not _inflight.acquire(blocking=False):
|
||||
return # already at max concurrent reports — drop silently
|
||||
|
||||
def _send() -> None:
|
||||
try:
|
||||
import json as _json
|
||||
import urllib.request
|
||||
|
||||
data = _json.dumps(payload).encode()
|
||||
req = urllib.request.Request(
|
||||
_REPORT_URL,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json", "User-Agent": _UA},
|
||||
method="POST",
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=5):
|
||||
pass
|
||||
except Exception:
|
||||
pass # never crash the caller
|
||||
finally:
|
||||
_inflight.release()
|
||||
|
||||
t = threading.Thread(target=_send, daemon=True)
|
||||
try:
|
||||
t.start()
|
||||
except Exception:
|
||||
# Thread could not be started (e.g. OS thread limit).
|
||||
# Release the slot so future reports are not permanently blocked.
|
||||
_inflight.release()
|
||||
|
||||
|
||||
class TickerrLogger(CustomLogger):
|
||||
"""
|
||||
LiteLLM built-in callback for Tickerr.
|
||||
LiteLLM callback that reports LLM API failures to Tickerr.
|
||||
|
||||
Activated with:
|
||||
litellm.callbacks = ["tickerr"]
|
||||
When explicitly enabled via ``litellm.callbacks = ["tickerr"]``,
|
||||
anonymous failure metadata is reported. No prompts, responses,
|
||||
API keys, or personal data are sent.
|
||||
|
||||
Optional env vars:
|
||||
TICKERR_CLIENT_TIER — "free" | "pro" | "team" | "enterprise" | "api_pay_as_you_go"
|
||||
TICKERR_REGION — e.g. "us-east-1"
|
||||
TICKERR_DISABLED - set to "true" to disable all reporting
|
||||
TICKERR_REGION - e.g. us-east-1
|
||||
TICKERR_SAMPLE_RATE - fraction of successes to report (0.0-1.0, default 0 = off)
|
||||
"""
|
||||
|
||||
def __init__(self, **kwargs: Any) -> None:
|
||||
super().__init__(**kwargs)
|
||||
self.client_tier: Optional[str] = os.environ.get("TICKERR_CLIENT_TIER")
|
||||
self.region: Optional[str] = os.environ.get("TICKERR_REGION")
|
||||
|
||||
# ── sync ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def log_failure_event(
|
||||
self,
|
||||
kwargs: Dict[str, Any],
|
||||
response_obj: Any,
|
||||
start_time: Union[datetime, float],
|
||||
end_time: Union[datetime, float],
|
||||
) -> None:
|
||||
self._report(kwargs, start_time, end_time)
|
||||
|
||||
# ── async ─────────────────────────────────────────────────────────────────
|
||||
|
||||
async def async_log_failure_event(
|
||||
self,
|
||||
kwargs: Dict[str, Any],
|
||||
response_obj: Any,
|
||||
start_time: Union[datetime, float],
|
||||
end_time: Union[datetime, float],
|
||||
) -> None:
|
||||
self._report(kwargs, start_time, end_time)
|
||||
|
||||
# ── internal ──────────────────────────────────────────────────────────────
|
||||
|
||||
def _report(
|
||||
self,
|
||||
kwargs: Dict[str, Any],
|
||||
start_time: Union[datetime, float],
|
||||
end_time: Union[datetime, float],
|
||||
) -> None:
|
||||
model: str = kwargs.get("model", "") or ""
|
||||
exception: Optional[BaseException] = kwargs.get("exception")
|
||||
|
||||
provider = _normalize_provider(model, kwargs)
|
||||
status_code = _extract_status_code(exception)
|
||||
# Only set error_type for codes we can classify with confidence
|
||||
error_type: Optional[str] = (
|
||||
_ERROR_TYPE_MAP.get(status_code) if status_code is not None else None
|
||||
)
|
||||
|
||||
# Strip provider prefix: "anthropic/claude-3-5-haiku" → "claude-3-5-haiku"
|
||||
model_clean = model.split("/", 1)[-1] if "/" in model else model
|
||||
|
||||
payload: Dict[str, Any] = {
|
||||
"provider": provider,
|
||||
"model": model_clean or None,
|
||||
"latency_ms": _latency_ms(start_time, end_time),
|
||||
self.disabled: bool = os.environ.get("TICKERR_DISABLED", "").lower() in {
|
||||
"1", "true", "yes",
|
||||
}
|
||||
if status_code is not None:
|
||||
payload["error_code"] = status_code
|
||||
if error_type:
|
||||
payload["error_type"] = error_type
|
||||
if self.client_tier:
|
||||
payload["client_tier"] = self.client_tier
|
||||
if self.region:
|
||||
payload["region"] = self.region
|
||||
self.region: Optional[str] = os.environ.get("TICKERR_REGION")
|
||||
try:
|
||||
self.sample_rate: float = min(1.0, max(0.0, float(os.environ.get("TICKERR_SAMPLE_RATE", "0"))))
|
||||
except (ValueError, TypeError):
|
||||
self.sample_rate = 0.0
|
||||
|
||||
_fire_and_forget(payload)
|
||||
def log_failure_event(self, kwargs: Dict[str, Any], response_obj: Any, start_time: Union[datetime, float], end_time: Union[datetime, float]) -> None:
|
||||
self._report(kwargs, start_time, end_time)
|
||||
|
||||
async def async_log_failure_event(self, kwargs: Dict[str, Any], response_obj: Any, start_time: Union[datetime, float], end_time: Union[datetime, float]) -> None:
|
||||
self._report(kwargs, start_time, end_time)
|
||||
|
||||
def log_success_event(self, kwargs: Dict[str, Any], response_obj: Any, start_time: Union[datetime, float], end_time: Union[datetime, float]) -> None:
|
||||
if self.sample_rate > 0 and random.random() < self.sample_rate:
|
||||
self._report(kwargs, start_time, end_time, is_success=True)
|
||||
|
||||
async def async_log_success_event(self, kwargs: Dict[str, Any], response_obj: Any, start_time: Union[datetime, float], end_time: Union[datetime, float]) -> None:
|
||||
if self.sample_rate > 0 and random.random() < self.sample_rate:
|
||||
self._report(kwargs, start_time, end_time, is_success=True)
|
||||
|
||||
def _report(self, kwargs: Dict[str, Any], start_time: Union[datetime, float], end_time: Union[datetime, float], is_success: bool = False) -> None:
|
||||
if self.disabled:
|
||||
return
|
||||
|
||||
model: str = kwargs.get("model", "") or ""
|
||||
|
||||
if isinstance(start_time, datetime) and isinstance(end_time, datetime):
|
||||
latency = round((end_time - start_time).total_seconds() * 1000)
|
||||
else:
|
||||
latency = round((float(end_time) - float(start_time)) * 1000)
|
||||
|
||||
payload = {k: v for k, v in {
|
||||
"provider": kwargs.get("litellm_params", {}).get("custom_llm_provider") or kwargs.get("custom_llm_provider"),
|
||||
"model": model or None,
|
||||
"latency_ms": latency,
|
||||
"event_type": "success" if is_success else "failure",
|
||||
"status_code": getattr(kwargs.get("exception"), "status_code", None),
|
||||
"region": self.region,
|
||||
}.items() if v is not None}
|
||||
|
||||
def _send() -> None:
|
||||
try:
|
||||
urllib.request.urlopen(
|
||||
urllib.request.Request(
|
||||
_REPORT_URL,
|
||||
data=json.dumps(payload).encode(),
|
||||
headers={"Content-Type": "application/json", "User-Agent": _UA},
|
||||
method="POST",
|
||||
),
|
||||
timeout=2,
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
threading.Thread(target=_send, daemon=True).start()
|
||||
|
|
|
|||
|
|
@ -1,545 +1,262 @@
|
|||
"""
|
||||
Unit tests for the Tickerr LiteLLM callback.
|
||||
|
||||
Tests cover:
|
||||
- TickerrLogger instantiation and env var config
|
||||
- Provider normalization from model names and litellm_params
|
||||
- Status code extraction from exceptions
|
||||
- Latency calculation for both datetime and float timestamps
|
||||
- Error type mapping (only known codes, no fallback default)
|
||||
- Payload construction
|
||||
- Thread cap (semaphore) under burst conditions
|
||||
- Fire-and-forget does not block or raise on network failure
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../.."))
|
||||
|
||||
from litellm.integrations.tickerr import (
|
||||
TickerrLogger,
|
||||
_ERROR_TYPE_MAP,
|
||||
_extract_status_code,
|
||||
_fire_and_forget,
|
||||
_inflight,
|
||||
_latency_ms,
|
||||
_normalize_provider,
|
||||
_MAX_INFLIGHT,
|
||||
)
|
||||
from litellm.integrations.tickerr import TickerrLogger
|
||||
|
||||
|
||||
# ── Provider normalization ────────────────────────────────────────────────────
|
||||
# -- Config --------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_normalize_provider_from_litellm_params():
|
||||
kwargs = {"litellm_params": {"custom_llm_provider": "anthropic"}}
|
||||
assert _normalize_provider("some-model", kwargs) == "anthropic"
|
||||
|
||||
|
||||
def test_normalize_provider_from_custom_llm_provider():
|
||||
kwargs = {"custom_llm_provider": "openai"}
|
||||
assert _normalize_provider("gpt-4o", kwargs) == "openai"
|
||||
|
||||
|
||||
def test_normalize_provider_from_model_prefix():
|
||||
assert _normalize_provider("anthropic/claude-3-5-haiku", {}) == "anthropic"
|
||||
assert _normalize_provider("openai/gpt-4o", {}) == "openai"
|
||||
|
||||
|
||||
def test_normalize_provider_from_model_name_pattern():
|
||||
assert _normalize_provider("claude-haiku-4-5", {}) == "anthropic"
|
||||
assert _normalize_provider("gpt-4o-mini", {}) == "openai"
|
||||
assert _normalize_provider("gemini-2.5-flash", {}) == "google"
|
||||
assert _normalize_provider("mistral-small-latest", {}) == "mistral"
|
||||
assert _normalize_provider("llama-3.3-70b", {}) == "meta"
|
||||
assert _normalize_provider("grok-3-mini", {}) == "xai"
|
||||
assert _normalize_provider("deepseek-v3", {}) == "deepseek"
|
||||
|
||||
|
||||
def test_normalize_provider_unknown():
|
||||
assert _normalize_provider("some-unknown-model-xyz", {}) == "unknown"
|
||||
|
||||
|
||||
# ── Status code extraction ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_extract_status_code_int():
|
||||
exc = MagicMock()
|
||||
exc.status_code = 429
|
||||
assert _extract_status_code(exc) == 429
|
||||
|
||||
|
||||
def test_extract_status_code_string():
|
||||
exc = MagicMock()
|
||||
exc.status_code = "503"
|
||||
assert _extract_status_code(exc) == 503
|
||||
|
||||
|
||||
def test_extract_status_code_none_exception():
|
||||
assert _extract_status_code(None) is None
|
||||
|
||||
|
||||
def test_extract_status_code_no_attribute():
|
||||
assert _extract_status_code(ValueError("oops")) is None
|
||||
|
||||
|
||||
# ── Latency calculation ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_latency_ms_with_datetime():
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
end = start + timedelta(milliseconds=1240)
|
||||
assert _latency_ms(start, end) == 1240
|
||||
|
||||
|
||||
def test_latency_ms_with_floats():
|
||||
assert _latency_ms(1000.0, 1001.5) == 1500
|
||||
|
||||
|
||||
def test_latency_ms_mixed_types_float():
|
||||
# Both floats — should not raise
|
||||
result = _latency_ms(0.0, 0.5)
|
||||
assert result == 500
|
||||
|
||||
|
||||
# ── Error type mapping ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_error_type_known_codes():
|
||||
assert _ERROR_TYPE_MAP[429] == "rate_limit"
|
||||
assert _ERROR_TYPE_MAP[529] == "overloaded"
|
||||
assert _ERROR_TYPE_MAP[503] == "overloaded"
|
||||
assert _ERROR_TYPE_MAP[408] == "timeout"
|
||||
assert _ERROR_TYPE_MAP[401] == "auth"
|
||||
|
||||
|
||||
def test_error_type_no_default_for_unknown_codes():
|
||||
# 500 is a generic server error (crash/bug), not definitively "overloaded".
|
||||
# Unknown codes must NOT map to any value.
|
||||
for code in (400, 404, 500, 502, 422, 301):
|
||||
assert code not in _ERROR_TYPE_MAP, f"code {code} should not be in _ERROR_TYPE_MAP"
|
||||
|
||||
|
||||
# ── TickerrLogger instantiation ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_tickerr_logger_default_init():
|
||||
def test_default_init():
|
||||
logger = TickerrLogger()
|
||||
assert logger.client_tier is None
|
||||
assert logger.disabled is False
|
||||
assert logger.sample_rate == 0.0
|
||||
assert logger.region is None
|
||||
|
||||
|
||||
def test_tickerr_logger_reads_env_vars():
|
||||
with patch.dict(os.environ, {"TICKERR_CLIENT_TIER": "pro", "TICKERR_REGION": "us-east-1"}):
|
||||
def test_reads_env_vars():
|
||||
with patch.dict(os.environ, {"TICKERR_REGION": "us-east-1", "TICKERR_SAMPLE_RATE": "0.1"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.client_tier == "pro"
|
||||
assert logger.region == "us-east-1"
|
||||
assert logger.sample_rate == 0.1
|
||||
|
||||
|
||||
# ── Payload construction via _report ─────────────────────────────────────────
|
||||
def test_disabled_flag():
|
||||
with patch.dict(os.environ, {"TICKERR_DISABLED": "true"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.disabled is True
|
||||
|
||||
|
||||
def test_report_builds_correct_payload():
|
||||
def test_invalid_sample_rate_does_not_crash():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "notanumber"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.sample_rate == 0.0
|
||||
|
||||
|
||||
def test_sample_rate_clamped_to_one():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "5.0"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.sample_rate == 1.0
|
||||
|
||||
|
||||
# -- Disabled ------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_disabled_skips_report():
|
||||
with patch.dict(os.environ, {"TICKERR_DISABLED": "1"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
logger._report({"model": "gpt-4o-mini", "exception": None}, start, end)
|
||||
mock_thread.assert_not_called()
|
||||
|
||||
|
||||
# -- Payload -------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_failure_payload():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = 429
|
||||
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=500)
|
||||
|
||||
kwargs = {
|
||||
"model": "claude-haiku-4-5",
|
||||
"exception": exc,
|
||||
"litellm_params": {"custom_llm_provider": "anthropic"},
|
||||
}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["provider"] == "anthropic"
|
||||
assert captured["model"] == "claude-haiku-4-5"
|
||||
assert captured["error_code"] == 429
|
||||
assert captured["error_type"] == "rate_limit"
|
||||
assert captured["latency_ms"] == 500
|
||||
sent = mock_urlopen.call_args[0][0]
|
||||
payload = json.loads(sent.data)
|
||||
assert payload["provider"] == "anthropic"
|
||||
assert payload["model"] == "claude-haiku-4-5"
|
||||
assert payload["status_code"] == 429
|
||||
assert payload["event_type"] == "failure"
|
||||
assert payload["latency_ms"] == 500
|
||||
|
||||
|
||||
def test_report_strips_provider_prefix_from_model():
|
||||
def test_model_passed_as_is():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
kwargs = {"model": "openai/gpt-4o-mini", "exception": None}
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "openai/gpt-4o-mini", "exception": None}, start, end)
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["model"] == "gpt-4o-mini"
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["model"] == "openai/gpt-4o-mini"
|
||||
|
||||
|
||||
def test_report_omits_error_type_for_unknown_code():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = 400 # not in _ERROR_TYPE_MAP
|
||||
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
kwargs = {"model": "gpt-4o", "exception": exc}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert "error_type" not in captured
|
||||
assert captured["error_code"] == 400
|
||||
|
||||
|
||||
# ── Thread cap ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_fire_and_forget_respects_semaphore_cap():
|
||||
"""Reports beyond _MAX_INFLIGHT are dropped silently without blocking."""
|
||||
# Exhaust the semaphore by acquiring all slots directly.
|
||||
# Track how many we actually acquired so the finally block releases exactly
|
||||
# that many — releasing more than acquired would push the count above its
|
||||
# initial maximum and corrupt later tests.
|
||||
acquired_count = 0
|
||||
for _ in range(_MAX_INFLIGHT):
|
||||
if _inflight.acquire(blocking=False):
|
||||
acquired_count += 1
|
||||
else:
|
||||
break
|
||||
|
||||
assert acquired_count == _MAX_INFLIGHT, (
|
||||
f"semaphore should have {_MAX_INFLIGHT} slots available at test start, "
|
||||
f"got {acquired_count}"
|
||||
)
|
||||
|
||||
try:
|
||||
# With semaphore exhausted, _fire_and_forget must return immediately
|
||||
# without starting a thread (non-blocking acquire fails → early return)
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
_fire_and_forget({"provider": "openai"})
|
||||
mock_thread.assert_not_called()
|
||||
finally:
|
||||
for _ in range(acquired_count):
|
||||
_inflight.release()
|
||||
|
||||
|
||||
def test_semaphore_released_on_thread_start_failure():
|
||||
"""If t.start() raises, the semaphore slot must be released so future reports work."""
|
||||
# Verify the semaphore can be re-acquired after a thread-start failure,
|
||||
# which proves the slot was released (without reading private CPython internals).
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start.side_effect = RuntimeError("OS thread limit")
|
||||
_fire_and_forget({"provider": "openai"})
|
||||
|
||||
# If the semaphore was not released, this acquire would block forever.
|
||||
# Use non-blocking to fail fast in case of a bug.
|
||||
acquired = _inflight.acquire(blocking=False)
|
||||
assert acquired, "semaphore slot was not released after thread start failure"
|
||||
_inflight.release() # restore
|
||||
|
||||
|
||||
# ── Network failure is silent ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_fire_and_forget_silent_on_network_error():
|
||||
"""A network error in the send function must not propagate to the caller."""
|
||||
# Mock threading.Thread so no real thread is created and no real network
|
||||
# call can escape the test boundary (repo rule: no real network calls).
|
||||
with patch("threading.Thread") as mock_thread_cls:
|
||||
mock_thread = MagicMock()
|
||||
mock_thread_cls.return_value = mock_thread
|
||||
|
||||
# Simulate the send function raising an OSError inside the thread
|
||||
def run_target(*args, **kwargs):
|
||||
target = mock_thread_cls.call_args[1].get("target") or mock_thread_cls.call_args[0][0]
|
||||
try:
|
||||
target()
|
||||
except Exception:
|
||||
pass # errors in thread body must not surface
|
||||
|
||||
mock_thread.start.side_effect = run_target
|
||||
with patch("urllib.request.urlopen", side_effect=OSError("connection refused")):
|
||||
_fire_and_forget({"provider": "anthropic", "model": "claude-haiku-4-5"})
|
||||
|
||||
|
||||
# ── Async hooks ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_log_failure_event_calls_report():
|
||||
def test_no_exception_omits_status_code():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
kwargs = {"model": "gpt-4o-mini", "exception": None}
|
||||
|
||||
with patch.object(logger, "_report") as mock_report:
|
||||
await logger.async_log_failure_event(kwargs, None, start, end)
|
||||
mock_report.assert_called_once_with(kwargs, start, end)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert "status_code" not in payload
|
||||
|
||||
|
||||
# ── Additional tests added to hit 67 % patch coverage ────────────────────────
|
||||
def test_region_included():
|
||||
with patch.dict(os.environ, {"TICKERR_REGION": "eu-west-1"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["region"] == "eu-west-1"
|
||||
|
||||
|
||||
# ── Sync log_failure_event hook ───────────────────────────────────────────────
|
||||
def test_latency_from_floats():
|
||||
logger = TickerrLogger()
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, 1000.0, 1001.5)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["latency_ms"] == 1500
|
||||
|
||||
|
||||
def test_log_failure_event_sync_calls_report():
|
||||
"""log_failure_event (sync) must delegate to _report."""
|
||||
def test_provider_from_top_level_kwarg():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None, "custom_llm_provider": "openai"}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["provider"] == "openai"
|
||||
|
||||
|
||||
def test_no_provider_omits_field():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert "provider" not in payload
|
||||
|
||||
|
||||
# -- Hooks ---------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_sync_failure_delegates():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=300)
|
||||
kwargs = {"model": "gpt-4o-mini", "exception": None}
|
||||
|
||||
with patch.object(logger, "_report") as mock_report:
|
||||
logger.log_failure_event(kwargs, None, start, end)
|
||||
mock_report.assert_called_once_with(kwargs, start, end)
|
||||
with patch.object(logger, "_report") as mock:
|
||||
logger.log_failure_event({"model": "gpt-4o-mini", "exception": None}, None, start, end)
|
||||
mock.assert_called_once()
|
||||
|
||||
|
||||
# ── _normalize_provider: extra provider map entries ───────────────────────────
|
||||
|
||||
|
||||
def test_normalize_provider_vertex_ai_maps_to_google():
|
||||
kwargs = {"custom_llm_provider": "vertex_ai"}
|
||||
assert _normalize_provider("gemini-2.5-pro", kwargs) == "google"
|
||||
|
||||
|
||||
def test_normalize_provider_vertex_ai_anthropic_maps_to_anthropic():
|
||||
kwargs = {"custom_llm_provider": "vertex_ai_anthropic"}
|
||||
assert _normalize_provider("claude-3-5-sonnet", kwargs) == "anthropic"
|
||||
|
||||
|
||||
def test_normalize_provider_azure_maps_to_azure():
|
||||
kwargs = {"litellm_params": {"custom_llm_provider": "azure"}}
|
||||
assert _normalize_provider("gpt-4o", kwargs) == "azure"
|
||||
|
||||
|
||||
def test_normalize_provider_bedrock_maps_to_aws():
|
||||
kwargs = {"custom_llm_provider": "bedrock"}
|
||||
assert _normalize_provider("claude-3-haiku", kwargs) == "aws"
|
||||
|
||||
|
||||
def test_normalize_provider_bedrock_converse_maps_to_aws():
|
||||
kwargs = {"custom_llm_provider": "bedrock_converse"}
|
||||
assert _normalize_provider("claude-3-haiku", kwargs) == "aws"
|
||||
|
||||
|
||||
def test_normalize_provider_command_model_pattern():
|
||||
"""command-* model names should map to cohere via regex."""
|
||||
assert _normalize_provider("command-r-plus", {}) == "cohere"
|
||||
assert _normalize_provider("command-r", {}) == "cohere"
|
||||
|
||||
|
||||
def test_normalize_provider_mixtral_model_pattern():
|
||||
"""mixtral-* model names should map to mistral via regex."""
|
||||
assert _normalize_provider("mixtral-8x7b-instruct", {}) == "mistral"
|
||||
|
||||
|
||||
def test_normalize_provider_deepseek_from_param():
|
||||
kwargs = {"custom_llm_provider": "deepseek"}
|
||||
assert _normalize_provider("deepseek-v3", kwargs) == "deepseek"
|
||||
|
||||
|
||||
def test_normalize_provider_openrouter_from_param():
|
||||
kwargs = {"litellm_params": {"custom_llm_provider": "openrouter"}}
|
||||
assert _normalize_provider("meta-llama/llama-3.3-70b-instruct", kwargs) == "openrouter"
|
||||
|
||||
|
||||
def test_normalize_provider_fireworks_ai_from_param():
|
||||
kwargs = {"custom_llm_provider": "fireworks_ai"}
|
||||
assert _normalize_provider("llama-v3-70b-instruct", kwargs) == "fireworks"
|
||||
|
||||
|
||||
def test_normalize_provider_cerebras_from_param():
|
||||
kwargs = {"custom_llm_provider": "cerebras"}
|
||||
assert _normalize_provider("llama3.1-8b", kwargs) == "cerebras"
|
||||
|
||||
|
||||
def test_normalize_provider_xai_from_param():
|
||||
kwargs = {"custom_llm_provider": "xai"}
|
||||
assert _normalize_provider("grok-3-mini", kwargs) == "xai"
|
||||
|
||||
|
||||
# ── _report: edge cases ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_report_no_exception_omits_error_fields():
|
||||
"""When exception is None, payload must not include error_code or error_type."""
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_failure_delegates():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
with patch.object(logger, "_report") as mock:
|
||||
await logger.async_log_failure_event({"model": "gpt-4o-mini", "exception": None}, None, start, end)
|
||||
mock.assert_called_once()
|
||||
|
||||
|
||||
# -- Success sampling ----------------------------------------------------------
|
||||
|
||||
|
||||
def test_success_not_reported_at_rate_zero():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch.object(logger, "_report") as mock:
|
||||
logger.log_success_event({"model": "gpt-4o", "exception": None}, None, start, end)
|
||||
mock.assert_not_called()
|
||||
|
||||
|
||||
def test_success_reported_when_sampled():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "1.0"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
kwargs = {"model": "gemini-2.5-flash", "exception": None}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger.log_success_event({"model": "gpt-4o", "exception": None, "litellm_params": {"custom_llm_provider": "openai"}}, None, start, end)
|
||||
|
||||
assert "error_code" not in captured
|
||||
assert "error_type" not in captured
|
||||
assert captured["provider"] == "google"
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["event_type"] == "success"
|
||||
assert payload["provider"] == "openai"
|
||||
|
||||
|
||||
def test_report_includes_client_tier_and_region_in_payload():
|
||||
"""client_tier and region must appear in the payload when set via env."""
|
||||
with patch.dict(os.environ, {"TICKERR_CLIENT_TIER": "enterprise", "TICKERR_REGION": "eu-west-1"}):
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_success_reported_when_sampled():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "1.0"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
captured = {}
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=150)
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
await logger.async_log_success_event({"model": "gpt-4o-mini", "exception": None, "litellm_params": {"custom_llm_provider": "openai"}}, None, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["event_type"] == "success"
|
||||
|
||||
|
||||
# -- Network failure does not crash --------------------------------------------
|
||||
|
||||
|
||||
def test_silent_on_network_error():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
kwargs = {"model": "gpt-4o", "exception": None}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["client_tier"] == "enterprise"
|
||||
assert captured["region"] == "eu-west-1"
|
||||
|
||||
|
||||
def test_report_empty_model_sets_model_none():
|
||||
"""An empty model string must result in model=None in the payload."""
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=50)
|
||||
kwargs = {"model": "", "exception": None}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["model"] is None
|
||||
|
||||
|
||||
def test_report_timeout_error_type():
|
||||
"""408 and 524 both map to error_type='timeout'."""
|
||||
logger = TickerrLogger()
|
||||
|
||||
for code in (408, 524):
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload, _c=code):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = code
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=5000)
|
||||
kwargs = {"model": "claude-haiku-4-5", "exception": exc}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["error_type"] == "timeout", f"Expected timeout for {code}"
|
||||
assert captured["error_code"] == code
|
||||
|
||||
|
||||
def test_report_auth_error_type():
|
||||
"""401 and 403 both map to error_type='auth'."""
|
||||
logger = TickerrLogger()
|
||||
|
||||
for code in (401, 403):
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = code
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
kwargs = {"model": "gpt-4o", "exception": exc}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["error_type"] == "auth", f"Expected auth for {code}"
|
||||
|
||||
|
||||
# ── _fire_and_forget: normal success path ─────────────────────────────────────
|
||||
|
||||
|
||||
def test_fire_and_forget_success_path_releases_semaphore():
|
||||
"""On a successful HTTP call, the semaphore slot must be released."""
|
||||
mock_response = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
with patch("threading.Thread") as mock_thread_cls:
|
||||
mock_thread = MagicMock()
|
||||
mock_thread_cls.return_value = mock_thread
|
||||
|
||||
def run_target_inline(*args, **kwargs):
|
||||
# Extract and call the real target so _send runs synchronously
|
||||
target = mock_thread_cls.call_args[1].get("target") or mock_thread_cls.call_args[0][0]
|
||||
target()
|
||||
|
||||
mock_thread.start.side_effect = run_target_inline
|
||||
|
||||
with patch("urllib.request.urlopen", return_value=mock_response):
|
||||
_fire_and_forget({"provider": "openai", "model": "gpt-4o-mini"})
|
||||
|
||||
# Semaphore must be acquirable — proves the finally block released the slot
|
||||
acquired = _inflight.acquire(blocking=False)
|
||||
assert acquired, "semaphore slot was not released after successful send"
|
||||
_inflight.release()
|
||||
|
||||
|
||||
# ── _latency_ms edge cases ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_latency_ms_zero():
|
||||
assert _latency_ms(1000.0, 1000.0) == 0
|
||||
|
||||
|
||||
def test_latency_ms_large_value():
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
end = start + timedelta(seconds=30)
|
||||
assert _latency_ms(start, end) == 30_000
|
||||
|
||||
|
||||
# ── _extract_status_code edge cases ──────────────────────────────────────────
|
||||
|
||||
|
||||
def test_extract_status_code_non_digit_string():
|
||||
"""A non-numeric string status_code must return None."""
|
||||
exc = MagicMock()
|
||||
exc.status_code = "N/A"
|
||||
assert _extract_status_code(exc) is None
|
||||
|
||||
|
||||
def test_extract_status_code_529():
|
||||
exc = MagicMock()
|
||||
exc.status_code = 529
|
||||
assert _extract_status_code(exc) == 529
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen", side_effect=OSError("refused")):
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
# No exception raised — test passes
|
||||
|
|
|
|||
|
|
@ -1,542 +1,262 @@
|
|||
"""
|
||||
Unit tests for the Tickerr LiteLLM callback.
|
||||
|
||||
Tests cover:
|
||||
- TickerrLogger instantiation and env var config
|
||||
- Provider normalization from model names and litellm_params
|
||||
- Status code extraction from exceptions
|
||||
- Latency calculation for both datetime and float timestamps
|
||||
- Error type mapping (only known codes, no fallback default)
|
||||
- Payload construction
|
||||
- Thread cap (semaphore) under burst conditions
|
||||
- Fire-and-forget does not block or raise on network failure
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.integrations.tickerr import (
|
||||
TickerrLogger,
|
||||
_ERROR_TYPE_MAP,
|
||||
_extract_status_code,
|
||||
_fire_and_forget,
|
||||
_inflight,
|
||||
_latency_ms,
|
||||
_normalize_provider,
|
||||
_MAX_INFLIGHT,
|
||||
)
|
||||
from litellm.integrations.tickerr import TickerrLogger
|
||||
|
||||
|
||||
# ── Provider normalization ────────────────────────────────────────────────────
|
||||
# -- Config --------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_normalize_provider_from_litellm_params():
|
||||
kwargs = {"litellm_params": {"custom_llm_provider": "anthropic"}}
|
||||
assert _normalize_provider("some-model", kwargs) == "anthropic"
|
||||
|
||||
|
||||
def test_normalize_provider_from_custom_llm_provider():
|
||||
kwargs = {"custom_llm_provider": "openai"}
|
||||
assert _normalize_provider("gpt-4o", kwargs) == "openai"
|
||||
|
||||
|
||||
def test_normalize_provider_from_model_prefix():
|
||||
assert _normalize_provider("anthropic/claude-3-5-haiku", {}) == "anthropic"
|
||||
assert _normalize_provider("openai/gpt-4o", {}) == "openai"
|
||||
|
||||
|
||||
def test_normalize_provider_from_model_name_pattern():
|
||||
assert _normalize_provider("claude-haiku-4-5", {}) == "anthropic"
|
||||
assert _normalize_provider("gpt-4o-mini", {}) == "openai"
|
||||
assert _normalize_provider("gemini-2.5-flash", {}) == "google"
|
||||
assert _normalize_provider("mistral-small-latest", {}) == "mistral"
|
||||
assert _normalize_provider("llama-3.3-70b", {}) == "meta"
|
||||
assert _normalize_provider("grok-3-mini", {}) == "xai"
|
||||
assert _normalize_provider("deepseek-v3", {}) == "deepseek"
|
||||
|
||||
|
||||
def test_normalize_provider_unknown():
|
||||
assert _normalize_provider("some-unknown-model-xyz", {}) == "unknown"
|
||||
|
||||
|
||||
# ── Status code extraction ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_extract_status_code_int():
|
||||
exc = MagicMock()
|
||||
exc.status_code = 429
|
||||
assert _extract_status_code(exc) == 429
|
||||
|
||||
|
||||
def test_extract_status_code_string():
|
||||
exc = MagicMock()
|
||||
exc.status_code = "503"
|
||||
assert _extract_status_code(exc) == 503
|
||||
|
||||
|
||||
def test_extract_status_code_none_exception():
|
||||
assert _extract_status_code(None) is None
|
||||
|
||||
|
||||
def test_extract_status_code_no_attribute():
|
||||
assert _extract_status_code(ValueError("oops")) is None
|
||||
|
||||
|
||||
# ── Latency calculation ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_latency_ms_with_datetime():
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
end = start + timedelta(milliseconds=1240)
|
||||
assert _latency_ms(start, end) == 1240
|
||||
|
||||
|
||||
def test_latency_ms_with_floats():
|
||||
assert _latency_ms(1000.0, 1001.5) == 1500
|
||||
|
||||
|
||||
def test_latency_ms_mixed_types_float():
|
||||
# Both floats — should not raise
|
||||
result = _latency_ms(0.0, 0.5)
|
||||
assert result == 500
|
||||
|
||||
|
||||
# ── Error type mapping ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_error_type_known_codes():
|
||||
assert _ERROR_TYPE_MAP[429] == "rate_limit"
|
||||
assert _ERROR_TYPE_MAP[529] == "overloaded"
|
||||
assert _ERROR_TYPE_MAP[503] == "overloaded"
|
||||
assert _ERROR_TYPE_MAP[408] == "timeout"
|
||||
assert _ERROR_TYPE_MAP[401] == "auth"
|
||||
|
||||
|
||||
def test_error_type_no_default_for_unknown_codes():
|
||||
# 500 is a generic server error (crash/bug), not definitively "overloaded".
|
||||
# Unknown codes must NOT map to any value.
|
||||
for code in (400, 404, 500, 502, 422, 301):
|
||||
assert code not in _ERROR_TYPE_MAP, f"code {code} should not be in _ERROR_TYPE_MAP"
|
||||
|
||||
|
||||
# ── TickerrLogger instantiation ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_tickerr_logger_default_init():
|
||||
def test_default_init():
|
||||
logger = TickerrLogger()
|
||||
assert logger.client_tier is None
|
||||
assert logger.disabled is False
|
||||
assert logger.sample_rate == 0.0
|
||||
assert logger.region is None
|
||||
|
||||
|
||||
def test_tickerr_logger_reads_env_vars():
|
||||
with patch.dict(os.environ, {"TICKERR_CLIENT_TIER": "pro", "TICKERR_REGION": "us-east-1"}):
|
||||
def test_reads_env_vars():
|
||||
with patch.dict(os.environ, {"TICKERR_REGION": "us-east-1", "TICKERR_SAMPLE_RATE": "0.1"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.client_tier == "pro"
|
||||
assert logger.region == "us-east-1"
|
||||
assert logger.sample_rate == 0.1
|
||||
|
||||
|
||||
# ── Payload construction via _report ─────────────────────────────────────────
|
||||
def test_disabled_flag():
|
||||
with patch.dict(os.environ, {"TICKERR_DISABLED": "true"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.disabled is True
|
||||
|
||||
|
||||
def test_report_builds_correct_payload():
|
||||
def test_invalid_sample_rate_does_not_crash():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "notanumber"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.sample_rate == 0.0
|
||||
|
||||
|
||||
def test_sample_rate_clamped_to_one():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "5.0"}):
|
||||
logger = TickerrLogger()
|
||||
assert logger.sample_rate == 1.0
|
||||
|
||||
|
||||
# -- Disabled ------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_disabled_skips_report():
|
||||
with patch.dict(os.environ, {"TICKERR_DISABLED": "1"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
logger._report({"model": "gpt-4o-mini", "exception": None}, start, end)
|
||||
mock_thread.assert_not_called()
|
||||
|
||||
|
||||
# -- Payload -------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_failure_payload():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = 429
|
||||
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=500)
|
||||
|
||||
kwargs = {
|
||||
"model": "claude-haiku-4-5",
|
||||
"exception": exc,
|
||||
"litellm_params": {"custom_llm_provider": "anthropic"},
|
||||
}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["provider"] == "anthropic"
|
||||
assert captured["model"] == "claude-haiku-4-5"
|
||||
assert captured["error_code"] == 429
|
||||
assert captured["error_type"] == "rate_limit"
|
||||
assert captured["latency_ms"] == 500
|
||||
sent = mock_urlopen.call_args[0][0]
|
||||
payload = json.loads(sent.data)
|
||||
assert payload["provider"] == "anthropic"
|
||||
assert payload["model"] == "claude-haiku-4-5"
|
||||
assert payload["status_code"] == 429
|
||||
assert payload["event_type"] == "failure"
|
||||
assert payload["latency_ms"] == 500
|
||||
|
||||
|
||||
def test_report_strips_provider_prefix_from_model():
|
||||
def test_model_passed_as_is():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
kwargs = {"model": "openai/gpt-4o-mini", "exception": None}
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "openai/gpt-4o-mini", "exception": None}, start, end)
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["model"] == "gpt-4o-mini"
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["model"] == "openai/gpt-4o-mini"
|
||||
|
||||
|
||||
def test_report_omits_error_type_for_unknown_code():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = 400 # not in _ERROR_TYPE_MAP
|
||||
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
kwargs = {"model": "gpt-4o", "exception": exc}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert "error_type" not in captured
|
||||
assert captured["error_code"] == 400
|
||||
|
||||
|
||||
# ── Thread cap ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_fire_and_forget_respects_semaphore_cap():
|
||||
"""Reports beyond _MAX_INFLIGHT are dropped silently without blocking."""
|
||||
# Exhaust the semaphore by acquiring all slots directly.
|
||||
# Track how many we actually acquired so the finally block releases exactly
|
||||
# that many — releasing more than acquired would push the count above its
|
||||
# initial maximum and corrupt later tests.
|
||||
acquired_count = 0
|
||||
for _ in range(_MAX_INFLIGHT):
|
||||
if _inflight.acquire(blocking=False):
|
||||
acquired_count += 1
|
||||
else:
|
||||
break
|
||||
|
||||
assert acquired_count == _MAX_INFLIGHT, (
|
||||
f"semaphore should have {_MAX_INFLIGHT} slots available at test start, "
|
||||
f"got {acquired_count}"
|
||||
)
|
||||
|
||||
try:
|
||||
# With semaphore exhausted, _fire_and_forget must return immediately
|
||||
# without starting a thread (non-blocking acquire fails → early return)
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
_fire_and_forget({"provider": "openai"})
|
||||
mock_thread.assert_not_called()
|
||||
finally:
|
||||
for _ in range(acquired_count):
|
||||
_inflight.release()
|
||||
|
||||
|
||||
def test_semaphore_released_on_thread_start_failure():
|
||||
"""If t.start() raises, the semaphore slot must be released so future reports work."""
|
||||
# Verify the semaphore can be re-acquired after a thread-start failure,
|
||||
# which proves the slot was released (without reading private CPython internals).
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start.side_effect = RuntimeError("OS thread limit")
|
||||
_fire_and_forget({"provider": "openai"})
|
||||
|
||||
# If the semaphore was not released, this acquire would block forever.
|
||||
# Use non-blocking to fail fast in case of a bug.
|
||||
acquired = _inflight.acquire(blocking=False)
|
||||
assert acquired, "semaphore slot was not released after thread start failure"
|
||||
_inflight.release() # restore
|
||||
|
||||
|
||||
# ── Network failure is silent ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_fire_and_forget_silent_on_network_error():
|
||||
"""A network error in the send function must not propagate to the caller."""
|
||||
# Mock threading.Thread so no real thread is created and no real network
|
||||
# call can escape the test boundary (repo rule: no real network calls).
|
||||
with patch("threading.Thread") as mock_thread_cls:
|
||||
mock_thread = MagicMock()
|
||||
mock_thread_cls.return_value = mock_thread
|
||||
|
||||
# Simulate the send function raising an OSError inside the thread
|
||||
def run_target(*args, **kwargs):
|
||||
target = mock_thread_cls.call_args[1].get("target") or mock_thread_cls.call_args[0][0]
|
||||
try:
|
||||
target()
|
||||
except Exception:
|
||||
pass # errors in thread body must not surface
|
||||
|
||||
mock_thread.start.side_effect = run_target
|
||||
with patch("urllib.request.urlopen", side_effect=OSError("connection refused")):
|
||||
_fire_and_forget({"provider": "anthropic", "model": "claude-haiku-4-5"})
|
||||
|
||||
|
||||
# ── Async hooks ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_log_failure_event_calls_report():
|
||||
def test_no_exception_omits_status_code():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
kwargs = {"model": "gpt-4o-mini", "exception": None}
|
||||
|
||||
with patch.object(logger, "_report") as mock_report:
|
||||
await logger.async_log_failure_event(kwargs, None, start, end)
|
||||
mock_report.assert_called_once_with(kwargs, start, end)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert "status_code" not in payload
|
||||
|
||||
|
||||
# ── Additional tests added to hit 67 % patch coverage ────────────────────────
|
||||
def test_region_included():
|
||||
with patch.dict(os.environ, {"TICKERR_REGION": "eu-west-1"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["region"] == "eu-west-1"
|
||||
|
||||
|
||||
# ── Sync log_failure_event hook ───────────────────────────────────────────────
|
||||
def test_latency_from_floats():
|
||||
logger = TickerrLogger()
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, 1000.0, 1001.5)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["latency_ms"] == 1500
|
||||
|
||||
|
||||
def test_log_failure_event_sync_calls_report():
|
||||
"""log_failure_event (sync) must delegate to _report."""
|
||||
def test_provider_from_top_level_kwarg():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None, "custom_llm_provider": "openai"}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["provider"] == "openai"
|
||||
|
||||
|
||||
def test_no_provider_omits_field():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert "provider" not in payload
|
||||
|
||||
|
||||
# -- Hooks ---------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_sync_failure_delegates():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=300)
|
||||
kwargs = {"model": "gpt-4o-mini", "exception": None}
|
||||
|
||||
with patch.object(logger, "_report") as mock_report:
|
||||
logger.log_failure_event(kwargs, None, start, end)
|
||||
mock_report.assert_called_once_with(kwargs, start, end)
|
||||
with patch.object(logger, "_report") as mock:
|
||||
logger.log_failure_event({"model": "gpt-4o-mini", "exception": None}, None, start, end)
|
||||
mock.assert_called_once()
|
||||
|
||||
|
||||
# ── _normalize_provider: extra provider map entries ───────────────────────────
|
||||
|
||||
|
||||
def test_normalize_provider_vertex_ai_maps_to_google():
|
||||
kwargs = {"custom_llm_provider": "vertex_ai"}
|
||||
assert _normalize_provider("gemini-2.5-pro", kwargs) == "google"
|
||||
|
||||
|
||||
def test_normalize_provider_vertex_ai_anthropic_maps_to_anthropic():
|
||||
kwargs = {"custom_llm_provider": "vertex_ai_anthropic"}
|
||||
assert _normalize_provider("claude-3-5-sonnet", kwargs) == "anthropic"
|
||||
|
||||
|
||||
def test_normalize_provider_azure_maps_to_azure():
|
||||
kwargs = {"litellm_params": {"custom_llm_provider": "azure"}}
|
||||
assert _normalize_provider("gpt-4o", kwargs) == "azure"
|
||||
|
||||
|
||||
def test_normalize_provider_bedrock_maps_to_aws():
|
||||
kwargs = {"custom_llm_provider": "bedrock"}
|
||||
assert _normalize_provider("claude-3-haiku", kwargs) == "aws"
|
||||
|
||||
|
||||
def test_normalize_provider_bedrock_converse_maps_to_aws():
|
||||
kwargs = {"custom_llm_provider": "bedrock_converse"}
|
||||
assert _normalize_provider("claude-3-haiku", kwargs) == "aws"
|
||||
|
||||
|
||||
def test_normalize_provider_command_model_pattern():
|
||||
"""command-* model names should map to cohere via regex."""
|
||||
assert _normalize_provider("command-r-plus", {}) == "cohere"
|
||||
assert _normalize_provider("command-r", {}) == "cohere"
|
||||
|
||||
|
||||
def test_normalize_provider_mixtral_model_pattern():
|
||||
"""mixtral-* model names should map to mistral via regex."""
|
||||
assert _normalize_provider("mixtral-8x7b-instruct", {}) == "mistral"
|
||||
|
||||
|
||||
def test_normalize_provider_deepseek_from_param():
|
||||
kwargs = {"custom_llm_provider": "deepseek"}
|
||||
assert _normalize_provider("deepseek-v3", kwargs) == "deepseek"
|
||||
|
||||
|
||||
def test_normalize_provider_openrouter_from_param():
|
||||
kwargs = {"litellm_params": {"custom_llm_provider": "openrouter"}}
|
||||
assert _normalize_provider("meta-llama/llama-3.3-70b-instruct", kwargs) == "openrouter"
|
||||
|
||||
|
||||
def test_normalize_provider_fireworks_ai_from_param():
|
||||
kwargs = {"custom_llm_provider": "fireworks_ai"}
|
||||
assert _normalize_provider("llama-v3-70b-instruct", kwargs) == "fireworks"
|
||||
|
||||
|
||||
def test_normalize_provider_cerebras_from_param():
|
||||
kwargs = {"custom_llm_provider": "cerebras"}
|
||||
assert _normalize_provider("llama3.1-8b", kwargs) == "cerebras"
|
||||
|
||||
|
||||
def test_normalize_provider_xai_from_param():
|
||||
kwargs = {"custom_llm_provider": "xai"}
|
||||
assert _normalize_provider("grok-3-mini", kwargs) == "xai"
|
||||
|
||||
|
||||
# ── _report: edge cases ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_report_no_exception_omits_error_fields():
|
||||
"""When exception is None, payload must not include error_code or error_type."""
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_failure_delegates():
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
with patch.object(logger, "_report") as mock:
|
||||
await logger.async_log_failure_event({"model": "gpt-4o-mini", "exception": None}, None, start, end)
|
||||
mock.assert_called_once()
|
||||
|
||||
|
||||
# -- Success sampling ----------------------------------------------------------
|
||||
|
||||
|
||||
def test_success_not_reported_at_rate_zero():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
|
||||
with patch.object(logger, "_report") as mock:
|
||||
logger.log_success_event({"model": "gpt-4o", "exception": None}, None, start, end)
|
||||
mock.assert_not_called()
|
||||
|
||||
|
||||
def test_success_reported_when_sampled():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "1.0"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
kwargs = {"model": "gemini-2.5-flash", "exception": None}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger.log_success_event({"model": "gpt-4o", "exception": None, "litellm_params": {"custom_llm_provider": "openai"}}, None, start, end)
|
||||
|
||||
assert "error_code" not in captured
|
||||
assert "error_type" not in captured
|
||||
assert captured["provider"] == "google"
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["event_type"] == "success"
|
||||
assert payload["provider"] == "openai"
|
||||
|
||||
|
||||
def test_report_includes_client_tier_and_region_in_payload():
|
||||
"""client_tier and region must appear in the payload when set via env."""
|
||||
with patch.dict(os.environ, {"TICKERR_CLIENT_TIER": "enterprise", "TICKERR_REGION": "eu-west-1"}):
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_success_reported_when_sampled():
|
||||
with patch.dict(os.environ, {"TICKERR_SAMPLE_RATE": "1.0"}):
|
||||
logger = TickerrLogger()
|
||||
|
||||
captured = {}
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=150)
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen") as mock_urlopen:
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
await logger.async_log_success_event({"model": "gpt-4o-mini", "exception": None, "litellm_params": {"custom_llm_provider": "openai"}}, None, start, end)
|
||||
|
||||
payload = json.loads(mock_urlopen.call_args[0][0].data)
|
||||
assert payload["event_type"] == "success"
|
||||
|
||||
|
||||
# -- Network failure does not crash --------------------------------------------
|
||||
|
||||
|
||||
def test_silent_on_network_error():
|
||||
logger = TickerrLogger()
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=100)
|
||||
kwargs = {"model": "gpt-4o", "exception": None}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["client_tier"] == "enterprise"
|
||||
assert captured["region"] == "eu-west-1"
|
||||
|
||||
|
||||
def test_report_empty_model_sets_model_none():
|
||||
"""An empty model string must result in model=None in the payload."""
|
||||
logger = TickerrLogger()
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=50)
|
||||
kwargs = {"model": "", "exception": None}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["model"] is None
|
||||
|
||||
|
||||
def test_report_timeout_error_type():
|
||||
"""408 and 524 both map to error_type='timeout'."""
|
||||
logger = TickerrLogger()
|
||||
|
||||
for code in (408, 524):
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload, _c=code):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = code
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=5000)
|
||||
kwargs = {"model": "claude-haiku-4-5", "exception": exc}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["error_type"] == "timeout", f"Expected timeout for {code}"
|
||||
assert captured["error_code"] == code
|
||||
|
||||
|
||||
def test_report_auth_error_type():
|
||||
"""401 and 403 both map to error_type='auth'."""
|
||||
logger = TickerrLogger()
|
||||
|
||||
for code in (401, 403):
|
||||
captured = {}
|
||||
|
||||
def fake_fire(payload):
|
||||
captured.update(payload)
|
||||
|
||||
exc = MagicMock()
|
||||
exc.status_code = code
|
||||
start = datetime(2024, 1, 1)
|
||||
end = start + timedelta(milliseconds=200)
|
||||
kwargs = {"model": "gpt-4o", "exception": exc}
|
||||
|
||||
with patch("litellm.integrations.tickerr._fire_and_forget", side_effect=fake_fire):
|
||||
logger._report(kwargs, start, end)
|
||||
|
||||
assert captured["error_type"] == "auth", f"Expected auth for {code}"
|
||||
|
||||
|
||||
# ── _fire_and_forget: normal success path ─────────────────────────────────────
|
||||
|
||||
|
||||
def test_fire_and_forget_success_path_releases_semaphore():
|
||||
"""On a successful HTTP call, the semaphore slot must be released."""
|
||||
mock_response = MagicMock()
|
||||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||||
mock_response.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
with patch("threading.Thread") as mock_thread_cls:
|
||||
mock_thread = MagicMock()
|
||||
mock_thread_cls.return_value = mock_thread
|
||||
|
||||
def run_target_inline(*args, **kwargs):
|
||||
# Extract and call the real target so _send runs synchronously
|
||||
target = mock_thread_cls.call_args[1].get("target") or mock_thread_cls.call_args[0][0]
|
||||
target()
|
||||
|
||||
mock_thread.start.side_effect = run_target_inline
|
||||
|
||||
with patch("urllib.request.urlopen", return_value=mock_response):
|
||||
_fire_and_forget({"provider": "openai", "model": "gpt-4o-mini"})
|
||||
|
||||
# Semaphore must be acquirable — proves the finally block released the slot
|
||||
acquired = _inflight.acquire(blocking=False)
|
||||
assert acquired, "semaphore slot was not released after successful send"
|
||||
_inflight.release()
|
||||
|
||||
|
||||
# ── _latency_ms edge cases ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_latency_ms_zero():
|
||||
assert _latency_ms(1000.0, 1000.0) == 0
|
||||
|
||||
|
||||
def test_latency_ms_large_value():
|
||||
start = datetime(2024, 1, 1, 0, 0, 0)
|
||||
end = start + timedelta(seconds=30)
|
||||
assert _latency_ms(start, end) == 30_000
|
||||
|
||||
|
||||
# ── _extract_status_code edge cases ──────────────────────────────────────────
|
||||
|
||||
|
||||
def test_extract_status_code_non_digit_string():
|
||||
"""A non-numeric string status_code must return None."""
|
||||
exc = MagicMock()
|
||||
exc.status_code = "N/A"
|
||||
assert _extract_status_code(exc) is None
|
||||
|
||||
|
||||
def test_extract_status_code_529():
|
||||
exc = MagicMock()
|
||||
exc.status_code = 529
|
||||
assert _extract_status_code(exc) == 529
|
||||
with patch("litellm.integrations.tickerr.urllib.request.urlopen", side_effect=OSError("refused")):
|
||||
with patch("litellm.integrations.tickerr.threading.Thread") as mock_thread:
|
||||
mock_thread.return_value.start = lambda: mock_thread.call_args[1]["target"]()
|
||||
logger._report({"model": "gpt-4o", "exception": None}, start, end)
|
||||
# No exception raised — test passes
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue