from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from src.services import signoz_client as signoz_module from src.services.openclaw import OpenClawService from src.services.signoz_client import SignOzClient @pytest.mark.asyncio async def test_clickhouse_query_can_fail_closed_on_transport_error( monkeypatch: pytest.MonkeyPatch, ) -> None: client = SignOzClient() http_client = SimpleNamespace( post=AsyncMock(side_effect=RuntimeError("connection refused")) ) monkeypatch.setattr( signoz_module, "get_clickhouse_client", AsyncMock(return_value=http_client), ) assert await client._query_clickhouse("SELECT 1") == [] with pytest.raises(RuntimeError, match="connection refused"): await client._query_clickhouse("SELECT 1", fail_closed=True) @pytest.mark.asyncio async def test_gold_metrics_requires_successful_clickhouse_queries( monkeypatch: pytest.MonkeyPatch, ) -> None: client = SignOzClient() query = AsyncMock(side_effect=RuntimeError("clickhouse unavailable")) monkeypatch.setattr(client, "_query_clickhouse", query) with pytest.raises(RuntimeError, match="clickhouse unavailable"): await client.get_gold_metrics("node-exporter-110") assert query.await_args.kwargs["fail_closed"] is True @pytest.mark.asyncio async def test_openclaw_marks_signoz_unavailable_when_metrics_query_fails( monkeypatch: pytest.MonkeyPatch, ) -> None: service = object.__new__(OpenClawService) service._signoz = SimpleNamespace( get_gold_metrics=AsyncMock(side_effect=RuntimeError("clickhouse unavailable")), generate_trace_url=lambda **_kwargs: "https://unused.example/traces", ) monkeypatch.setattr(signoz_module.settings, "SIGNOZ_URL", "https://signoz.example") metrics, trace_url = await service.get_signoz_context("node-exporter-110") assert metrics is None assert trace_url == "https://signoz.example/traces?service=node-exporter-110"