tracer-cloud--opensre/tests/integrations/test_betterstack.py

"""Unit tests for the Better Stack integration module.

Mirrors the test_rabbitmq.py pattern: config layer + validation against
mocked ``httpx.MockTransport`` responses, no real Better Stack calls.
"""

from __future__ import annotations

from collections.abc import Callable

import httpx
import pytest
from pydantic import ValidationError

from integrations import betterstack as bs_module
from integrations.betterstack import (
    DEFAULT_BETTERSTACK_MAX_ROWS,
    DEFAULT_BETTERSTACK_TIMEOUT_S,
    BetterStackConfig,
    BetterStackValidationResult,
    betterstack_config_from_env,
    betterstack_extract_params,
    betterstack_is_available,
    build_betterstack_config,
    query_logs,
    validate_betterstack_config,
)

# ---------------------------------------------------------------------------
# Mock transport helper
# ---------------------------------------------------------------------------


Handler = Callable[[httpx.Request], httpx.Response]


@pytest.fixture
def patched_sql_client(monkeypatch: pytest.MonkeyPatch):
    """Monkeypatch ``_sql_client`` to route through an ``httpx.MockTransport``.

    Returns a callable ``install(handler)`` that accepts a per-test request
    handler and wires it into the module under test.
    """

    def install(handler: Handler) -> None:
        def _fake_client(config: BetterStackConfig) -> httpx.Client:
            return httpx.Client(
                auth=(config.username, config.password),
                timeout=float(config.timeout_seconds),
                transport=httpx.MockTransport(handler),
            )

        monkeypatch.setattr(bs_module, "_sql_client", _fake_client)

    return install


def _configured() -> BetterStackConfig:
    return BetterStackConfig(
        query_endpoint="https://eu-nbg-2-connect.betterstackdata.com",
        username="u",
        password="p",
    )


class TestBetterStackConfig:
    def test_defaults(self) -> None:
        c = BetterStackConfig()
        assert c.query_endpoint == ""
        assert c.username == ""
        assert c.password == ""
        assert c.sources == []
        assert c.timeout_seconds == DEFAULT_BETTERSTACK_TIMEOUT_S
        assert c.max_rows == DEFAULT_BETTERSTACK_MAX_ROWS
        assert c.is_configured is False

    def test_is_configured_requires_endpoint_and_username(self) -> None:
        assert BetterStackConfig(query_endpoint="https://x", username="u").is_configured is True
        assert BetterStackConfig(query_endpoint="https://x").is_configured is False
        assert BetterStackConfig(username="u").is_configured is False

    def test_normalize_endpoint_strips_trailing_slash_and_whitespace(self) -> None:
        c = BetterStackConfig(query_endpoint="  https://eu-nbg-2-connect.betterstackdata.com/  ")
        assert c.query_endpoint == "https://eu-nbg-2-connect.betterstackdata.com"

    def test_normalize_username_strips_whitespace(self) -> None:
        assert BetterStackConfig(username="  u  ").username == "u"

    def test_normalize_password_strips_via_parent_validator(self) -> None:
        # StrictConfigModel's wildcard string validator strips all fields
        # before the field-specific validator runs, including passwords.
        assert BetterStackConfig(password="  p  ").password == "p"

    def test_normalize_password_coerces_none(self) -> None:
        assert BetterStackConfig(password=None).password == ""  # type: ignore[arg-type]

    def test_sources_from_comma_string(self) -> None:
        c = BetterStackConfig(sources="t1_myapp, t2_gateway")
        assert c.sources == ["t1_myapp", "t2_gateway"]

    def test_sources_from_list(self) -> None:
        c = BetterStackConfig(sources=["t1_myapp", "t2_gateway"])
        assert c.sources == ["t1_myapp", "t2_gateway"]

    def test_sources_empty_string(self) -> None:
        assert BetterStackConfig(sources="").sources == []

    def test_sources_none(self) -> None:
        assert BetterStackConfig(sources=None).sources == []  # type: ignore[arg-type]

    def test_sources_strip_whitespace_and_drop_empty(self) -> None:
        assert BetterStackConfig(sources="a, , b").sources == ["a", "b"]

    def test_timeout_zero_raises(self) -> None:
        with pytest.raises(ValidationError):
            BetterStackConfig(timeout_seconds=0)

    def test_max_rows_exceeds_cap_raises(self) -> None:
        with pytest.raises(ValidationError):
            BetterStackConfig(max_rows=99_999)

    def test_unknown_field_rejected(self) -> None:
        with pytest.raises(ValidationError):
            BetterStackConfig(endpoint_url="https://x")  # type: ignore[call-arg]


class TestBuildBetterStackConfig:
    def test_empty_input(self) -> None:
        c = build_betterstack_config(None)
        assert c.query_endpoint == ""
        assert c.is_configured is False

    def test_dict_input(self) -> None:
        c = build_betterstack_config(
            {
                "query_endpoint": "https://x",
                "username": "u",
                "password": "p",
                "sources": "t1,t2",
            }
        )
        assert c.query_endpoint == "https://x"
        assert c.username == "u"
        assert c.sources == ["t1", "t2"]


class TestBetterStackConfigFromEnv:
    def test_returns_none_without_endpoint(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.delenv("BETTERSTACK_QUERY_ENDPOINT", raising=False)
        monkeypatch.setenv("BETTERSTACK_USERNAME", "u")
        assert betterstack_config_from_env() is None

    def test_returns_none_without_username(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("BETTERSTACK_QUERY_ENDPOINT", "https://x")
        monkeypatch.delenv("BETTERSTACK_USERNAME", raising=False)
        assert betterstack_config_from_env() is None

    def test_loads_from_env_full(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv(
            "BETTERSTACK_QUERY_ENDPOINT",
            "https://eu-nbg-2-connect.betterstackdata.com",
        )
        monkeypatch.setenv("BETTERSTACK_USERNAME", "u")
        monkeypatch.setenv("BETTERSTACK_PASSWORD", "p")
        monkeypatch.setenv("BETTERSTACK_SOURCES", "t1_myapp,t2_gateway")
        c = betterstack_config_from_env()
        assert c is not None
        assert c.query_endpoint == "https://eu-nbg-2-connect.betterstackdata.com"
        assert c.username == "u"
        assert c.password == "p"
        assert c.sources == ["t1_myapp", "t2_gateway"]

    def test_loads_without_optional_sources(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setenv("BETTERSTACK_QUERY_ENDPOINT", "https://x")
        monkeypatch.setenv("BETTERSTACK_USERNAME", "u")
        monkeypatch.delenv("BETTERSTACK_SOURCES", raising=False)
        c = betterstack_config_from_env()
        assert c is not None
        assert c.sources == []


class TestBetterStackHelpers:
    def test_is_available_true_with_configured_sources(self) -> None:
        assert (
            betterstack_is_available(
                {
                    "betterstack": {
                        "query_endpoint": "https://x",
                        "username": "u",
                        "sources": ["t1_myapp"],
                    }
                }
            )
            is True
        )

    def test_is_available_true_with_source_hint(self) -> None:
        # Alert-derived source_hint (from betterstack_source annotation) is
        # sufficient to make the tool callable even without a configured
        # sources list.
        assert (
            betterstack_is_available(
                {
                    "betterstack": {
                        "query_endpoint": "https://x",
                        "username": "u",
                        "sources": [],
                        "source_hint": "t1_alert_inferred",
                    }
                }
            )
            is True
        )

    def test_is_available_false_without_endpoint(self) -> None:
        assert betterstack_is_available({"betterstack": {"username": "u"}}) is False

    def test_is_available_false_without_username(self) -> None:
        assert betterstack_is_available({"betterstack": {"query_endpoint": "https://x"}}) is False

    def test_is_available_false_when_source_missing(self) -> None:
        assert betterstack_is_available({}) is False

    def test_is_available_false_when_no_way_to_derive_source(self) -> None:
        # Credentials present but neither a configured sources list nor an
        # alert-derived source_hint — the executor has no path to propagate
        # a source to the tool, so availability must be False to prevent a
        # deterministic empty-source failure at call time.
        assert (
            betterstack_is_available(
                {
                    "betterstack": {
                        "query_endpoint": "https://x",
                        "username": "u",
                        "sources": [],
                    }
                }
            )
            is False
        )

    def test_extract_params_full(self) -> None:
        params = betterstack_extract_params(
            {
                "betterstack": {
                    "query_endpoint": "https://x",
                    "username": "u",
                    "password": "p",
                    "sources": ["t1"],
                    "source_hint": "t2_alert_inferred",
                }
            }
        )
        assert params == {
            "query_endpoint": "https://x",
            "username": "u",
            "password": "p",
            "sources": ["t1"],
            "source": "t2_alert_inferred",
        }

    def test_extract_params_defaults_when_missing(self) -> None:
        assert betterstack_extract_params({}) == {
            "query_endpoint": "",
            "username": "",
            "password": "",
            "sources": [],
            "source": "",
        }

    def test_extract_params_source_from_hint(self) -> None:
        # The alert-derived source_hint surfaces as the scalar ``source`` kwarg
        # so the executor can propagate alert context into the tool.
        params = betterstack_extract_params({"betterstack": {"source_hint": "t99_svc"}})
        assert params["source"] == "t99_svc"

    def test_extract_params_source_empty_when_no_hint(self) -> None:
        # When no alert hint is present, ``source`` is empty and the tool's
        # own fallback picks the first configured source (if any).
        params = betterstack_extract_params({"betterstack": {"sources": ["t1", "t2"]}})
        assert params["source"] == ""
        assert params["sources"] == ["t1", "t2"]

    def test_extract_params_sources_copy_not_alias(self) -> None:
        original = ["t1"]
        params = betterstack_extract_params({"betterstack": {"sources": original}})
        params["sources"].append("t2")
        assert original == ["t1"]


class TestValidateBetterStackConfig:
    def test_returns_not_configured_when_missing_creds(self) -> None:
        result = validate_betterstack_config(BetterStackConfig())
        assert isinstance(result, BetterStackValidationResult)
        assert result.ok is False
        assert "required" in result.detail.lower()

    def test_ok_on_200_probe(self, patched_sql_client) -> None:
        patched_sql_client(lambda _req: httpx.Response(200, text='{"1":1}\n'))
        result = validate_betterstack_config(_configured())
        assert result.ok is True

    def test_fails_on_empty_body(self, patched_sql_client) -> None:
        patched_sql_client(lambda _req: httpx.Response(200, text=""))
        result = validate_betterstack_config(_configured())
        assert result.ok is False
        assert "empty body" in result.detail.lower()

    def test_fails_on_401_with_auth_hint(self, patched_sql_client) -> None:
        patched_sql_client(lambda _req: httpx.Response(401, text="bad creds"))
        result = validate_betterstack_config(_configured())
        assert result.ok is False
        assert "authentication" in result.detail.lower()
        assert "BETTERSTACK_USERNAME" in result.detail

    def test_fails_on_404_with_endpoint_hint(self, patched_sql_client) -> None:
        patched_sql_client(lambda _req: httpx.Response(404, text="not found"))
        result = validate_betterstack_config(_configured())
        assert result.ok is False
        assert "endpoint" in result.detail.lower()
        assert "BETTERSTACK_QUERY_ENDPOINT" in result.detail

    def test_fails_on_500_includes_status_and_body(self, patched_sql_client) -> None:
        patched_sql_client(lambda _req: httpx.Response(500, text="boom"))
        result = validate_betterstack_config(_configured())
        assert result.ok is False
        assert "500" in result.detail
        assert "boom" in result.detail

    def test_fails_on_request_error(self, patched_sql_client) -> None:
        def _raiser(_req: httpx.Request) -> httpx.Response:
            raise httpx.ConnectError("dns failed")

        patched_sql_client(_raiser)
        result = validate_betterstack_config(_configured())
        assert result.ok is False
        assert "request failed" in result.detail.lower()
        assert "dns failed" in result.detail

    def test_probe_uses_required_wire_contract(self, patched_sql_client) -> None:
        captured: dict[str, httpx.Request] = {}

        def _capturing(req: httpx.Request) -> httpx.Response:
            captured["req"] = req
            return httpx.Response(200, text='{"1":1}\n')

        patched_sql_client(_capturing)
        validate_betterstack_config(_configured())

        req = captured["req"]
        assert req.method == "POST"
        assert req.url.params.get("output_format_pretty_row_numbers") == "0"
        assert req.headers.get("content-type") == "text/plain"
        assert req.content == b"SELECT 1 FORMAT JSONEachRow"
        # Basic auth header is populated by httpx from the client's ``auth`` tuple.
        assert req.headers.get("authorization", "").lower().startswith("basic ")


class TestQueryLogs:
    def test_not_configured(self) -> None:
        result = query_logs(BetterStackConfig(), source="t1_x")
        assert result["available"] is False
        assert "not configured" in result["error"].lower()
        assert result["rows"] == []
        assert result["row_count"] == 0

    def test_invalid_source_name_rejected(self, patched_sql_client) -> None:
        called = {"n": 0}

        def _never(_req: httpx.Request) -> httpx.Response:
            called["n"] += 1
            return httpx.Response(200, text="")

        patched_sql_client(_never)
        result = query_logs(_configured(), source="t1; DROP TABLE users;--")
        assert result["available"] is False
        assert "invalid" in result["error"].lower()
        assert called["n"] == 0  # must not fire a request

    def test_empty_source_rejected(self) -> None:
        result = query_logs(_configured(), source="")
        assert result["available"] is False
        assert "invalid" in result["error"].lower()

    @pytest.mark.parametrize(
        "payload",
        [
            "t1; DROP TABLE users;--",
            "'; DROP TABLE logs--",
            "source OR 1=1",
            "../../../etc/passwd",
            "t1_logs UNION SELECT password FROM users",
            "t1_logs/*comment*/",
            "t1_logs -- trailing comment",
            "t1_logs`",
            "t1_logs\nDROP TABLE secrets",
            "t1_logs'",
        ],
    )
    def test_source_name_rejects_sql_injection_patterns(self, payload: str) -> None:
        assert bs_module._validate_source_name(payload) is None

    def test_invalid_since_rejected(self, patched_sql_client) -> None:
        patched_sql_client(lambda _r: httpx.Response(200, text=""))
        result = query_logs(_configured(), source="t1_myapp", since="not-a-timestamp")
        assert result["available"] is False
        assert "since" in result["error"].lower()

    def test_invalid_until_rejected(self, patched_sql_client) -> None:
        patched_sql_client(lambda _r: httpx.Response(200, text=""))
        result = query_logs(_configured(), source="t1_myapp", until="nope")
        assert result["available"] is False
        assert "until" in result["error"].lower()

    def test_happy_path_parses_jsoneachrow(self, patched_sql_client) -> None:
        body = (
            '{"dt":"2026-04-20T00:00:00Z","raw":"hello"}\n'
            '{"dt":"2026-04-20T00:00:01Z","raw":"world"}\n'
        )
        patched_sql_client(lambda _r: httpx.Response(200, text=body))
        result = query_logs(_configured(), source="t1_myapp")
        assert result["available"] is True
        assert result["betterstack_source"] == "t1_myapp"
        assert result["row_count"] == 2
        assert result["rows"][0]["raw"] == "hello"
        assert result["rows"][1]["raw"] == "world"

    def test_empty_body_is_zero_rows(self, patched_sql_client) -> None:
        patched_sql_client(lambda _r: httpx.Response(200, text=""))
        result = query_logs(_configured(), source="t1_myapp")
        assert result["available"] is True
        assert result["rows"] == []
        assert result["row_count"] == 0

    def test_skips_malformed_rows(self, patched_sql_client) -> None:
        body = '{"ok":1}\nnot-json\n{"ok":2}\n'
        patched_sql_client(lambda _r: httpx.Response(200, text=body))
        result = query_logs(_configured(), source="t1_myapp")
        assert result["row_count"] == 2
        assert result["rows"] == [{"ok": 1}, {"ok": 2}]

    def test_401_reported_as_auth(self, patched_sql_client) -> None:
        patched_sql_client(lambda _r: httpx.Response(401, text="nope"))
        result = query_logs(_configured(), source="t1_myapp")
        assert result["available"] is False
        assert "authentication" in result["error"].lower()

    def test_5xx_reports_status_and_body(self, patched_sql_client) -> None:
        patched_sql_client(lambda _r: httpx.Response(500, text="boom"))
        result = query_logs(_configured(), source="t1_myapp")
        assert result["available"] is False
        assert "500" in result["error"]
        assert "boom" in result["error"]

    def test_request_error_surfaced(self, patched_sql_client) -> None:
        def _raiser(_r: httpx.Request) -> httpx.Response:
            raise httpx.ConnectError("dns failed")

        patched_sql_client(_raiser)
        result = query_logs(_configured(), source="t1_myapp")
        assert result["available"] is False
        assert "request failed" in result["error"].lower()

    def test_limit_clamped_to_config_max_rows(self, patched_sql_client) -> None:
        captured: dict[str, httpx.Request] = {}

        def _capturing(req: httpx.Request) -> httpx.Response:
            captured["req"] = req
            return httpx.Response(200, text="")

        patched_sql_client(_capturing)
        config = BetterStackConfig(
            query_endpoint="https://x",
            username="u",
            password="p",
            max_rows=50,
        )
        result = query_logs(config, source="t1_myapp", limit=9999)
        assert result["limit"] == 50
        body = captured["req"].content.decode()
        assert "LIMIT 50" in body

    def test_negative_limit_clamped_to_one(self, patched_sql_client) -> None:
        # A negative ``limit`` must never reach the SQL as ``LIMIT -1`` — ClickHouse
        # rejects that with a 400 parse error. The integration clamps with max(1, ...).
        captured: dict[str, httpx.Request] = {}

        def _capturing(req: httpx.Request) -> httpx.Response:
            captured["req"] = req
            return httpx.Response(200, text="")

        patched_sql_client(_capturing)
        query_logs(_configured(), source="t1_myapp", limit=-1)
        body = captured["req"].content.decode()
        assert "LIMIT 1" in body
        assert "LIMIT -1" not in body

    def test_sql_unions_recent_and_historical(self, patched_sql_client) -> None:
        captured: dict[str, httpx.Request] = {}

        def _capturing(req: httpx.Request) -> httpx.Response:
            captured["req"] = req
            return httpx.Response(200, text="")

        patched_sql_client(_capturing)
        query_logs(_configured(), source="t1_myapp")
        body = captured["req"].content.decode()
        # Both branches of the UNION must be present, with the expected suffixes.
        assert "remote(t1_myapp_logs)" in body
        assert "s3Cluster(primary, t1_myapp_s3)" in body
        assert "UNION ALL" in body
        # _row_type=1 always filters the s3 branch down to log rows.
        assert "_row_type = 1" in body
        assert body.rstrip().endswith("FORMAT JSONEachRow")

    def test_sql_contains_time_bounds_when_set(self, patched_sql_client) -> None:
        captured: dict[str, httpx.Request] = {}

        def _capturing(req: httpx.Request) -> httpx.Response:
            captured["req"] = req
            return httpx.Response(200, text="")

        patched_sql_client(_capturing)
        query_logs(
            _configured(),
            source="t1_myapp",
            since="2026-04-20T00:00:00Z",
            until="2026-04-20T01:00:00Z",
        )
        body = captured["req"].content.decode()
        assert "dt >= parseDateTime64BestEffort('2026-04-20T00:00:00Z'" in body
        assert "dt <= parseDateTime64BestEffort('2026-04-20T01:00:00Z'" in body

    def test_sql_omits_time_bounds_when_unset(self, patched_sql_client) -> None:
        captured: dict[str, httpx.Request] = {}

        def _capturing(req: httpx.Request) -> httpx.Response:
            captured["req"] = req
            return httpx.Response(200, text="")

        patched_sql_client(_capturing)
        query_logs(_configured(), source="t1_myapp")
        body = captured["req"].content.decode()
        # No explicit time bounds in the WHERE clauses — the s3 branch still
        # carries _row_type = 1 but there's no parseDateTime64BestEffort call.
        assert "parseDateTime64BestEffort" not in body
        assert "ORDER BY dt DESC" in body