mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
zeekay
1b367e2b62
The tree was a debranded snapshot of upstream taken around 2026-02-20, and upstream has moved about five months since. This replaces it with upstream master, mechanically re-derived: bin/debrand does the rename, and the pipeline and squashed migrations are restored from the shipped tree. Not a merge. The previous import replaced the tree wholesale in a single commit, so there is no shared content history to merge against. Reproducing the rename on the new tree is what actually converges. What survives: .hanzo/workflows, the Dockerfiles and bin/docker* entrypoints, compose files, LLM.md, and insights/migrations. The squashed initial migration is load-bearing — the live database records it by name, so upstream's ~918 individual migrations cannot come back. Nine products are dropped as unreferenced. Twenty-one more that we do not run are kept anyway because the core app still imports them; unpicking those is a refactor and does not belong in a rebase. This base does not build yet, and the reasons are not in the rename: upstream now wants Python 3.13.13 against our 3.12.12 and Django 5.2 against our 4.2, and the dependency manifests still name upstream packages that we republish under our own names. Those are decisions, not renames.
235 lines
9.3 KiB
Python
235 lines
9.3 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Iterator
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from fakeredis import aioredis as fakeredis
|
|
|
|
from llm_gateway.circuit_breaker import (
|
|
BUCKET_WIDTH_SECONDS,
|
|
AnthropicCircuitBreaker,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_redis() -> fakeredis.FakeRedis:
|
|
return fakeredis.FakeRedis()
|
|
|
|
|
|
@pytest.fixture
|
|
def frozen_time() -> Iterator[MagicMock]:
|
|
with patch("llm_gateway.circuit_breaker.time.time") as mock_time:
|
|
mock_time.return_value = 1_000_000.0
|
|
yield mock_time
|
|
|
|
|
|
def make_breaker(
|
|
redis: fakeredis.FakeRedis | None,
|
|
*,
|
|
failure_threshold: float = 0.25,
|
|
window_seconds: int = 300,
|
|
bypass_probability: float = 0.9,
|
|
min_requests: int = 5,
|
|
enabled: bool = True,
|
|
) -> AnthropicCircuitBreaker:
|
|
return AnthropicCircuitBreaker(
|
|
redis=redis,
|
|
failure_threshold=failure_threshold,
|
|
window_seconds=window_seconds,
|
|
bypass_probability=bypass_probability,
|
|
min_requests=min_requests,
|
|
enabled=enabled,
|
|
)
|
|
|
|
|
|
class TestAnthropicCircuitBreaker:
|
|
async def test_disabled_breaker_is_inert(self, fake_redis: fakeredis.FakeRedis, frozen_time: MagicMock) -> None:
|
|
breaker = make_breaker(fake_redis, enabled=False)
|
|
for _ in range(10):
|
|
await breaker.record_outcome(success=False)
|
|
decision = await breaker.evaluate()
|
|
assert decision.bypass is False
|
|
assert decision.open is False
|
|
assert decision.failure_rate == 0.0
|
|
assert decision.total_requests == 0
|
|
|
|
async def test_no_redis_is_inert(self, frozen_time: MagicMock) -> None:
|
|
breaker = make_breaker(None)
|
|
await breaker.record_outcome(success=False)
|
|
decision = await breaker.evaluate()
|
|
assert decision.bypass is False
|
|
assert decision.open is False
|
|
|
|
async def test_below_min_requests_does_not_open(
|
|
self, fake_redis: fakeredis.FakeRedis, frozen_time: MagicMock
|
|
) -> None:
|
|
breaker = make_breaker(fake_redis, min_requests=10, failure_threshold=0.25)
|
|
for _ in range(4):
|
|
await breaker.record_outcome(success=False)
|
|
decision = await breaker.evaluate()
|
|
assert decision.open is False
|
|
assert decision.bypass is False
|
|
assert decision.failure_rate == 1.0
|
|
assert decision.total_requests == 4
|
|
|
|
async def test_opens_when_failure_rate_crosses_threshold(
|
|
self, fake_redis: fakeredis.FakeRedis, frozen_time: MagicMock
|
|
) -> None:
|
|
breaker = make_breaker(fake_redis, min_requests=5, failure_threshold=0.25)
|
|
for _ in range(15):
|
|
await breaker.record_outcome(success=True)
|
|
decision = await breaker.evaluate()
|
|
assert decision.open is False
|
|
assert decision.failure_rate == 0.0
|
|
|
|
for _ in range(5):
|
|
await breaker.record_outcome(success=False)
|
|
decision = await breaker.evaluate()
|
|
assert decision.open is True
|
|
assert decision.failure_rate == pytest.approx(0.25)
|
|
assert decision.total_requests == 20
|
|
|
|
async def test_closes_when_failure_rate_drops(
|
|
self, fake_redis: fakeredis.FakeRedis, frozen_time: MagicMock
|
|
) -> None:
|
|
breaker = make_breaker(fake_redis, min_requests=5, failure_threshold=0.25)
|
|
for _ in range(5):
|
|
await breaker.record_outcome(success=False)
|
|
for _ in range(5):
|
|
await breaker.record_outcome(success=True)
|
|
decision = await breaker.evaluate()
|
|
assert decision.open is True
|
|
assert decision.failure_rate == 0.5
|
|
|
|
for _ in range(50):
|
|
await breaker.record_outcome(success=True)
|
|
decision = await breaker.evaluate()
|
|
assert decision.open is False
|
|
assert decision.failure_rate < 0.25
|
|
|
|
async def test_bypass_uses_probability_when_open(
|
|
self, fake_redis: fakeredis.FakeRedis, frozen_time: MagicMock
|
|
) -> None:
|
|
breaker = make_breaker(fake_redis, min_requests=5, failure_threshold=0.25, bypass_probability=0.9)
|
|
for _ in range(10):
|
|
await breaker.record_outcome(success=False)
|
|
|
|
with patch("llm_gateway.circuit_breaker.random.random", return_value=0.5):
|
|
decision = await breaker.evaluate()
|
|
assert decision.bypass is True
|
|
assert decision.open is True
|
|
|
|
with patch("llm_gateway.circuit_breaker.random.random", return_value=0.95):
|
|
decision = await breaker.evaluate()
|
|
assert decision.bypass is False
|
|
assert decision.open is True
|
|
|
|
async def test_bypass_never_when_closed(self, fake_redis: fakeredis.FakeRedis, frozen_time: MagicMock) -> None:
|
|
breaker = make_breaker(fake_redis, min_requests=5, failure_threshold=0.25, bypass_probability=0.9)
|
|
for _ in range(20):
|
|
await breaker.record_outcome(success=True)
|
|
|
|
with patch("llm_gateway.circuit_breaker.random.random", return_value=0.0):
|
|
decision = await breaker.evaluate()
|
|
assert decision.bypass is False
|
|
assert decision.open is False
|
|
|
|
async def test_old_buckets_outside_window_excluded(self, fake_redis: fakeredis.FakeRedis) -> None:
|
|
# window=300s with 30s buckets → 10 buckets retained.
|
|
# Failures recorded 11 buckets ago should not count.
|
|
breaker = make_breaker(fake_redis, min_requests=5, failure_threshold=0.25, window_seconds=300)
|
|
with patch("llm_gateway.circuit_breaker.time.time", return_value=1_000_000.0):
|
|
for _ in range(20):
|
|
await breaker.record_outcome(success=False)
|
|
|
|
far_future = 1_000_000.0 + (BUCKET_WIDTH_SECONDS * 11)
|
|
with patch("llm_gateway.circuit_breaker.time.time", return_value=far_future):
|
|
for _ in range(5):
|
|
await breaker.record_outcome(success=True)
|
|
decision = await breaker.evaluate()
|
|
assert decision.total_requests == 5
|
|
assert decision.failure_rate == 0.0
|
|
assert decision.open is False
|
|
|
|
async def test_redis_failure_is_swallowed(self, frozen_time: MagicMock) -> None:
|
|
broken_redis = MagicMock()
|
|
broken_redis.pipeline = MagicMock(side_effect=RuntimeError("redis down"))
|
|
breaker = make_breaker(broken_redis)
|
|
await breaker.record_outcome(success=False)
|
|
decision = await breaker.evaluate()
|
|
assert decision.bypass is False
|
|
assert decision.open is False
|
|
assert decision.failure_rate == 0.0
|
|
assert decision.total_requests == 0
|
|
|
|
async def test_redis_failure_increments_error_counter(self, frozen_time: MagicMock) -> None:
|
|
from llm_gateway.metrics.prometheus import ANTHROPIC_CIRCUIT_BREAKER_REDIS_ERRORS
|
|
|
|
broken_redis = MagicMock()
|
|
broken_redis.pipeline = MagicMock(side_effect=RuntimeError("redis down"))
|
|
breaker = make_breaker(broken_redis)
|
|
|
|
before_record = ANTHROPIC_CIRCUIT_BREAKER_REDIS_ERRORS.labels(op="record")._value.get()
|
|
before_read = ANTHROPIC_CIRCUIT_BREAKER_REDIS_ERRORS.labels(op="read")._value.get()
|
|
await breaker.record_outcome(success=False)
|
|
await breaker._get_stats()
|
|
assert ANTHROPIC_CIRCUIT_BREAKER_REDIS_ERRORS.labels(op="record")._value.get() == before_record + 1
|
|
assert ANTHROPIC_CIRCUIT_BREAKER_REDIS_ERRORS.labels(op="read")._value.get() == before_read + 1
|
|
|
|
async def test_redis_timeout_does_not_block(self) -> None:
|
|
import asyncio as _asyncio
|
|
|
|
slow_pipe = MagicMock()
|
|
slow_pipe.hincrby = MagicMock(return_value=slow_pipe)
|
|
slow_pipe.expire = MagicMock(return_value=slow_pipe)
|
|
slow_pipe.hmget = MagicMock(return_value=slow_pipe)
|
|
|
|
async def hang() -> object:
|
|
await _asyncio.sleep(10.0)
|
|
return []
|
|
|
|
slow_pipe.execute = MagicMock(side_effect=hang)
|
|
slow_redis = MagicMock()
|
|
slow_redis.pipeline = MagicMock(return_value=slow_pipe)
|
|
|
|
breaker = make_breaker(slow_redis)
|
|
with patch("llm_gateway.circuit_breaker.REDIS_OP_TIMEOUT_SECONDS", 0.01):
|
|
start = time.monotonic()
|
|
await breaker.record_outcome(success=True)
|
|
decision = await breaker.evaluate()
|
|
elapsed = time.monotonic() - start
|
|
assert elapsed < 1.0
|
|
assert decision.total_requests == 0
|
|
|
|
|
|
class TestPublishGaugesLoop:
|
|
async def test_publishes_gauges_then_cancels_cleanly(self, fake_redis: fakeredis.FakeRedis) -> None:
|
|
import asyncio as _asyncio
|
|
|
|
from llm_gateway.circuit_breaker import publish_anthropic_breaker_gauges_loop
|
|
from llm_gateway.metrics.prometheus import (
|
|
ANTHROPIC_CIRCUIT_BREAKER_FAILURE_RATE,
|
|
ANTHROPIC_CIRCUIT_BREAKER_OPEN,
|
|
ANTHROPIC_CIRCUIT_BREAKER_WINDOW_REQUESTS,
|
|
)
|
|
|
|
breaker = make_breaker(fake_redis, min_requests=5, failure_threshold=0.25)
|
|
for _ in range(15):
|
|
await breaker.record_outcome(success=True)
|
|
for _ in range(5):
|
|
await breaker.record_outcome(success=False)
|
|
|
|
task = _asyncio.create_task(publish_anthropic_breaker_gauges_loop(breaker, interval_seconds=10))
|
|
await _asyncio.sleep(0.05)
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except _asyncio.CancelledError:
|
|
pass
|
|
|
|
assert ANTHROPIC_CIRCUIT_BREAKER_OPEN._value.get() == 1
|
|
assert ANTHROPIC_CIRCUIT_BREAKER_FAILURE_RATE._value.get() == pytest.approx(0.25)
|
|
assert ANTHROPIC_CIRCUIT_BREAKER_WINDOW_REQUESTS._value.get() == 20
|