Files
wehub-resource-sync 4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

252 lines
8.5 KiB
Python

"""Direct unit tests for integrations.grafana.loki.LokiMixin.
Exercise the Loki query path with a fake mixin host so we never hit a real
Grafana datasource. Covers the happy path (stream flattening, metadata
propagation), the not-configured short-circuit, exception handling with and
without a wrapped HTTP response, and the empty-result envelope.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from integrations.grafana.loki import LokiMixin
# ---------------------------------------------------------------------------
# Test host
# ---------------------------------------------------------------------------
class _FakeLokiHost(LokiMixin):
"""Minimal stand-in for GrafanaClientBase that satisfies LokiMixin's needs.
LokiMixin reads ``is_configured``, ``account_id``, and
``loki_datasource_uid`` and calls ``_build_datasource_url`` and
``_make_request``. Everything else on the real base is irrelevant here.
"""
def __init__(
self,
*,
is_configured: bool = True,
account_id: str = "acct-1",
loki_datasource_uid: str = "loki-uid",
instance_url: str = "https://grafana.example.com",
) -> None:
self.is_configured = is_configured # type: ignore[assignment]
self.account_id = account_id
self.loki_datasource_uid = loki_datasource_uid
self.instance_url = instance_url
self.last_url: str | None = None
self.last_params: dict[str, str] | None = None
self.make_request_mock: MagicMock = MagicMock()
def _build_datasource_url(self, datasource_uid: str, path: str) -> str:
return f"{self.instance_url}/api/datasources/proxy/uid/{datasource_uid}{path}"
def _make_request(
self,
url: str,
params: dict[str, str] | None = None,
timeout: int = 10,
) -> dict[str, Any]:
self.last_url = url
self.last_params = params
return self.make_request_mock(url, params=params, timeout=timeout)
# ---------------------------------------------------------------------------
# Fixture data
# ---------------------------------------------------------------------------
_QUERY = '{service_name="lambda-mock-dag"}'
def _two_stream_response() -> dict[str, Any]:
return {
"status": "success",
"data": {
"resultType": "streams",
"result": [
{
"stream": {"service_name": "lambda-mock-dag", "level": "error"},
"values": [
["1700000000000000000", "boom"],
["1700000000000000001", "crash"],
],
},
{
"stream": {"service_name": "lambda-mock-dag", "level": "info"},
"values": [["1700000000000000002", "started"]],
},
],
},
}
# ---------------------------------------------------------------------------
# Happy path
# ---------------------------------------------------------------------------
class TestQueryLokiSuccess:
"""Configured client returns a flattened, well-formed envelope."""
def test_flattens_streams_into_log_rows(self) -> None:
host = _FakeLokiHost()
host.make_request_mock.return_value = _two_stream_response()
result = host.query_loki(_QUERY)
assert result["success"] is True
assert result["query"] == _QUERY
assert result["account_id"] == "acct-1"
assert result["total_streams"] == 2
assert result["total_logs"] == 3
assert len(result["logs"]) == 3
def test_each_log_row_carries_timestamp_message_and_labels(self) -> None:
host = _FakeLokiHost()
host.make_request_mock.return_value = _two_stream_response()
result = host.query_loki(_QUERY)
first = result["logs"][0]
assert first == {
"timestamp": "1700000000000000000",
"message": "boom",
"labels": {"service_name": "lambda-mock-dag", "level": "error"},
}
assert result["logs"][2]["labels"]["level"] == "info"
assert result["logs"][2]["message"] == "started"
def test_uses_loki_datasource_url_and_passes_query_params(self) -> None:
host = _FakeLokiHost(loki_datasource_uid="my-loki")
host.make_request_mock.return_value = _two_stream_response()
fake_now_s = 1_700_000_000.0
with patch("integrations.grafana.loki.time.time", return_value=fake_now_s):
host.query_loki(_QUERY, time_range_minutes=5, limit=42)
assert host.last_url is not None
assert "/api/datasources/proxy/uid/my-loki/loki/api/v1/query_range" in host.last_url
assert host.last_params is not None
assert host.last_params["query"] == _QUERY
assert host.last_params["limit"] == "42"
expected_end = int(fake_now_s * 1e9)
expected_start = expected_end - (5 * 60 * int(1e9))
assert host.last_params["start"] == str(expected_start)
assert host.last_params["end"] == str(expected_end)
# ---------------------------------------------------------------------------
# Not configured short-circuit
# ---------------------------------------------------------------------------
class TestQueryLokiNotConfigured:
"""When is_configured is False we never reach the HTTP layer."""
def test_returns_not_configured_envelope(self) -> None:
host = _FakeLokiHost(is_configured=False, account_id="missing-acct")
result = host.query_loki(_QUERY)
assert result == {
"success": False,
"error": "Grafana client not configured for account 'missing-acct'",
"logs": [],
}
def test_does_not_invoke_make_request(self) -> None:
host = _FakeLokiHost(is_configured=False)
host.query_loki(_QUERY)
host.make_request_mock.assert_not_called()
# ---------------------------------------------------------------------------
# Exception handling
# ---------------------------------------------------------------------------
class TestQueryLokiExceptions:
"""Exceptions surface as a stable failure envelope."""
def test_plain_exception_returns_str_error_and_empty_response(self) -> None:
host = _FakeLokiHost()
host.make_request_mock.side_effect = RuntimeError("connection refused")
result = host.query_loki(_QUERY)
assert result == {
"success": False,
"error": "connection refused",
"response": "",
"logs": [],
}
def test_exception_with_response_includes_status_and_truncated_text(self) -> None:
host = _FakeLokiHost()
long_text = "x" * 500
response = MagicMock(status_code=502, text=long_text)
err = RuntimeError("bad gateway")
err.response = response # type: ignore[attr-defined]
host.make_request_mock.side_effect = err
result = host.query_loki(_QUERY)
assert result["success"] is False
assert result["error"] == "Loki query failed: 502"
assert result["response"] == long_text[:300]
assert len(result["response"]) == 300
assert result["logs"] == []
def test_exception_with_none_response_falls_back_to_str_error(self) -> None:
host = _FakeLokiHost()
err = RuntimeError("transport error")
err.response = None # type: ignore[attr-defined]
host.make_request_mock.side_effect = err
result = host.query_loki(_QUERY)
assert result["success"] is False
assert result["error"] == "transport error"
assert result["response"] == ""
# ---------------------------------------------------------------------------
# Empty-result envelope stability
# ---------------------------------------------------------------------------
class TestQueryLokiEmptyResult:
"""No streams returned should still yield a valid success envelope."""
@pytest.mark.parametrize(
"payload",
[
{"data": {"result": []}},
{"data": {}},
{},
],
)
def test_zero_streams_yields_empty_logs_but_success(self, payload: dict) -> None:
host = _FakeLokiHost()
host.make_request_mock.return_value = payload
result = host.query_loki(_QUERY)
assert result["success"] is True
assert result["logs"] == []
assert result["total_streams"] == 0
assert result["total_logs"] == 0
assert result["query"] == _QUERY
assert result["account_id"] == "acct-1"