57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.services import heartbeat_report_service as heartbeat_module
|
|
from src.services.heartbeat_report_service import HeartbeatReportService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gemini_heartbeat_reads_durable_auth_without_external_probe(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
heartbeat_module.settings,
|
|
"GEMINI_API_KEY",
|
|
"configured-test-value",
|
|
)
|
|
limiter = MagicMock()
|
|
limiter.get_provider_authentication_status = AsyncMock(
|
|
return_value={
|
|
"authenticated": None,
|
|
"authentication_verified": False,
|
|
}
|
|
)
|
|
|
|
with patch(
|
|
"src.services.ai_rate_limiter.get_ai_rate_limiter",
|
|
return_value=limiter,
|
|
), patch.object(heartbeat_module.httpx, "AsyncClient") as client:
|
|
unverified = await HeartbeatReportService()._probe_gemini()
|
|
limiter.get_provider_authentication_status.return_value = {
|
|
"authenticated": True,
|
|
"authentication_verified": True,
|
|
}
|
|
verified = await HeartbeatReportService()._probe_gemini()
|
|
|
|
assert unverified.ok is False
|
|
assert "durable receipt" in unverified.status
|
|
assert verified.ok is True
|
|
assert "durable receipt verified" in verified.status
|
|
client.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_cloud_heartbeats_are_shadow_metadata_only() -> None:
|
|
with patch.object(heartbeat_module.httpx, "AsyncClient") as client:
|
|
nemotron = await HeartbeatReportService()._probe_nemotron()
|
|
claude = await HeartbeatReportService()._probe_claude()
|
|
|
|
assert nemotron.ok is False
|
|
assert claude.ok is False
|
|
assert "shadow metadata only" in nemotron.status
|
|
assert "shadow metadata only" in claude.status
|
|
client.assert_not_called()
|