Claude (Opus 4.6)
f2a277f7b5
Co-authored-by: Claude (Opus 4.6) <claude@hermes.local> Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>
268 lines
9.6 KiB
Python
268 lines
9.6 KiB
Python
"""Tests for Claude Quota Monitor and Metabolic Protocol."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
from infrastructure.claude_quota import (
|
|
MetabolicTier,
|
|
QuotaMonitor,
|
|
QuotaStatus,
|
|
_time_remaining,
|
|
get_quota_monitor,
|
|
)
|
|
|
|
|
|
def _make_status(five_hour: float = 0.0, seven_day: float = 0.0) -> QuotaStatus:
|
|
"""Helper: build a QuotaStatus with given utilization values."""
|
|
return QuotaStatus(
|
|
five_hour_utilization=five_hour,
|
|
five_hour_resets_at=None,
|
|
seven_day_utilization=seven_day,
|
|
seven_day_resets_at=None,
|
|
raw_response={},
|
|
fetched_at=datetime.now(UTC),
|
|
)
|
|
|
|
|
|
class TestMetabolicTierThresholds:
|
|
"""Test the three-tier metabolic protocol thresholds."""
|
|
|
|
def test_burst_when_five_hour_below_50pct(self):
|
|
status = _make_status(five_hour=0.49, seven_day=0.10)
|
|
assert status.recommended_tier == MetabolicTier.BURST
|
|
|
|
def test_burst_at_zero_utilization(self):
|
|
status = _make_status(five_hour=0.0, seven_day=0.0)
|
|
assert status.recommended_tier == MetabolicTier.BURST
|
|
|
|
def test_active_when_five_hour_at_50pct(self):
|
|
status = _make_status(five_hour=0.50, seven_day=0.10)
|
|
assert status.recommended_tier == MetabolicTier.ACTIVE
|
|
|
|
def test_active_when_five_hour_between_50_and_80pct(self):
|
|
status = _make_status(five_hour=0.79, seven_day=0.10)
|
|
assert status.recommended_tier == MetabolicTier.ACTIVE
|
|
|
|
def test_active_when_five_hour_at_80pct(self):
|
|
# five_hour >= 0.80 but seven_day < 0.80 → ACTIVE (not RESTING)
|
|
status = _make_status(five_hour=0.80, seven_day=0.50)
|
|
assert status.recommended_tier == MetabolicTier.ACTIVE
|
|
|
|
def test_resting_when_seven_day_at_80pct(self):
|
|
status = _make_status(five_hour=0.30, seven_day=0.80)
|
|
assert status.recommended_tier == MetabolicTier.RESTING
|
|
|
|
def test_resting_when_seven_day_above_80pct(self):
|
|
status = _make_status(five_hour=0.10, seven_day=0.95)
|
|
assert status.recommended_tier == MetabolicTier.RESTING
|
|
|
|
def test_resting_when_both_critical(self):
|
|
status = _make_status(five_hour=0.90, seven_day=0.90)
|
|
assert status.recommended_tier == MetabolicTier.RESTING
|
|
|
|
def test_seven_day_takes_precedence_over_five_hour(self):
|
|
# Weekly quota critical overrides whatever five-hour says
|
|
status = _make_status(five_hour=0.10, seven_day=0.85)
|
|
assert status.recommended_tier == MetabolicTier.RESTING
|
|
|
|
|
|
class TestQuotaStatusProperties:
|
|
"""Test QuotaStatus computed properties."""
|
|
|
|
def test_five_hour_pct(self):
|
|
status = _make_status(five_hour=0.42)
|
|
assert status.five_hour_pct == 42
|
|
|
|
def test_seven_day_pct(self):
|
|
status = _make_status(seven_day=0.75)
|
|
assert status.seven_day_pct == 75
|
|
|
|
def test_summary_contains_tier(self):
|
|
status = _make_status(five_hour=0.20, seven_day=0.10)
|
|
summary = status.summary()
|
|
assert "burst" in summary
|
|
assert "20%" in summary
|
|
|
|
def test_five_hour_resets_in_unknown_when_none(self):
|
|
status = _make_status()
|
|
assert status.five_hour_resets_in == "unknown"
|
|
|
|
def test_seven_day_resets_in_unknown_when_none(self):
|
|
status = _make_status()
|
|
assert status.seven_day_resets_in == "unknown"
|
|
|
|
|
|
class TestTimeRemaining:
|
|
"""Test _time_remaining helper."""
|
|
|
|
def test_none_returns_unknown(self):
|
|
assert _time_remaining(None) == "unknown"
|
|
|
|
def test_empty_string_returns_unknown(self):
|
|
assert _time_remaining("") == "unknown"
|
|
|
|
def test_past_time_returns_resetting_now(self):
|
|
past = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
|
|
assert _time_remaining(past) == "resetting now"
|
|
|
|
def test_future_time_hours_and_minutes(self):
|
|
future = (datetime.now(UTC) + timedelta(hours=2, minutes=15)).isoformat()
|
|
result = _time_remaining(future)
|
|
assert "2h" in result
|
|
# Minutes may vary ±1 due to test execution time
|
|
assert "m" in result
|
|
|
|
def test_future_time_minutes_only(self):
|
|
future = (datetime.now(UTC) + timedelta(minutes=45)).isoformat()
|
|
result = _time_remaining(future)
|
|
assert "h" not in result
|
|
# Minutes may vary ±1 due to test execution time
|
|
assert "m" in result
|
|
|
|
def test_z_suffix_handled(self):
|
|
future = (datetime.now(UTC) + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
result = _time_remaining(future)
|
|
assert result != "unknown"
|
|
|
|
|
|
class TestQuotaMonitorSelectModel:
|
|
"""Test select_model metabolic routing."""
|
|
|
|
def test_no_quota_high_complexity_returns_14b(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._get_token = lambda: None
|
|
assert monitor.select_model("high") == "qwen3:14b"
|
|
|
|
def test_no_quota_low_complexity_returns_8b(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._get_token = lambda: None
|
|
assert monitor.select_model("low") == "qwen3:8b"
|
|
|
|
def test_burst_tier_high_complexity_returns_cloud(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.10, seven_day=0.10)
|
|
monitor._cache_seconds = 9999
|
|
result = monitor.select_model("high")
|
|
assert result == "claude-sonnet-4-6"
|
|
|
|
def test_burst_tier_medium_complexity_returns_14b(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.10, seven_day=0.10)
|
|
monitor._cache_seconds = 9999
|
|
result = monitor.select_model("medium")
|
|
assert result == "qwen3:14b"
|
|
|
|
def test_active_tier_returns_14b(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.65, seven_day=0.10)
|
|
monitor._cache_seconds = 9999
|
|
result = monitor.select_model("high")
|
|
assert result == "qwen3:14b"
|
|
|
|
def test_resting_tier_returns_8b(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.10, seven_day=0.85)
|
|
monitor._cache_seconds = 9999
|
|
result = monitor.select_model("high")
|
|
assert result == "qwen3:8b"
|
|
|
|
|
|
class TestQuotaMonitorShouldUseCloud:
|
|
"""Test should_use_cloud gate."""
|
|
|
|
def test_no_credentials_always_false(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._get_token = lambda: None
|
|
assert monitor.should_use_cloud("critical") is False
|
|
|
|
def test_critical_task_allowed_when_under_95pct(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.10, seven_day=0.94)
|
|
monitor._cache_seconds = 9999
|
|
assert monitor.should_use_cloud("critical") is True
|
|
|
|
def test_critical_task_blocked_when_over_95pct(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.10, seven_day=0.96)
|
|
monitor._cache_seconds = 9999
|
|
assert monitor.should_use_cloud("critical") is False
|
|
|
|
def test_high_task_allowed_under_60pct(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.59, seven_day=0.10)
|
|
monitor._cache_seconds = 9999
|
|
assert monitor.should_use_cloud("high") is True
|
|
|
|
def test_high_task_blocked_at_60pct(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.60, seven_day=0.10)
|
|
monitor._cache_seconds = 9999
|
|
assert monitor.should_use_cloud("high") is False
|
|
|
|
def test_normal_task_allowed_under_30pct(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.29, seven_day=0.10)
|
|
monitor._cache_seconds = 9999
|
|
assert monitor.should_use_cloud("normal") is True
|
|
|
|
def test_normal_task_blocked_at_30pct(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.30, seven_day=0.10)
|
|
monitor._cache_seconds = 9999
|
|
assert monitor.should_use_cloud("normal") is False
|
|
|
|
def test_routine_task_always_false(self):
|
|
monitor = QuotaMonitor()
|
|
monitor._last_status = _make_status(five_hour=0.0, seven_day=0.0)
|
|
monitor._cache_seconds = 9999
|
|
assert monitor.should_use_cloud("routine") is False
|
|
|
|
|
|
class TestQuotaMonitorCaching:
|
|
"""Test 30-second TTL cache."""
|
|
|
|
def test_cached_result_returned_within_ttl(self):
|
|
monitor = QuotaMonitor()
|
|
fresh_status = _make_status(five_hour=0.10)
|
|
monitor._last_status = fresh_status
|
|
monitor._cache_seconds = 30
|
|
|
|
# Should NOT re-fetch — returns cached
|
|
with patch.object(monitor, "_get_token", return_value="tok") as mock_tok:
|
|
result = monitor.check()
|
|
mock_tok.assert_not_called()
|
|
|
|
assert result is fresh_status
|
|
|
|
def test_stale_cache_triggers_fetch(self):
|
|
monitor = QuotaMonitor()
|
|
old_time = datetime.now(UTC) - timedelta(seconds=60)
|
|
stale_status = QuotaStatus(
|
|
five_hour_utilization=0.10,
|
|
five_hour_resets_at=None,
|
|
seven_day_utilization=0.10,
|
|
seven_day_resets_at=None,
|
|
raw_response={},
|
|
fetched_at=old_time,
|
|
)
|
|
monitor._last_status = stale_status
|
|
|
|
# Token unavailable → returns None (triggers re-fetch path)
|
|
with patch.object(monitor, "_get_token", return_value=None):
|
|
result = monitor.check()
|
|
|
|
assert result is None # No credentials after cache miss
|
|
|
|
|
|
class TestGetQuotaMonitorSingleton:
|
|
"""Test module-level singleton."""
|
|
|
|
def test_returns_same_instance(self):
|
|
m1 = get_quota_monitor()
|
|
m2 = get_quota_monitor()
|
|
assert m1 is m2
|
|
|
|
def test_returns_quota_monitor_instance(self):
|
|
monitor = get_quota_monitor()
|
|
assert isinstance(monitor, QuotaMonitor)
|