tracer-cloud--opensre/tests/tools/test_integration_wave_tools.py

"""Focused tests for integration-wave tool slices."""

from __future__ import annotations

from typing import Any

from integrations.azure.tools.azure_monitor_logs_tool import query_azure_monitor_logs
from integrations.bitbucket.tools.bitbucket_search_code_tool import _resolve_config
from integrations.openobserve.tools.openobserve_logs_tool import query_openobserve_logs
from integrations.opensearch.tools.opensearch_analytics_tool import query_opensearch_analytics
from integrations.snowflake.tools.snowflake_query_history_tool import query_snowflake_history


class _MockResponse:
    def __init__(self, payload: dict[str, Any]) -> None:
        self._payload = payload

    def raise_for_status(self) -> None:
        return None

    def json(self) -> dict[str, Any]:
        return self._payload


def test_bitbucket_resolve_config_accepts_routed_instance_metadata() -> None:
    config = _resolve_config(
        "acme",
        "bb-user",
        "bb-pass",
        "https://api.bitbucket.org/2.0/",
        40,
        "bb-1",
    )

    assert config is not None
    assert config.workspace == "acme"
    assert config.base_url == "https://api.bitbucket.org/2.0"
    assert config.max_results == 40
    assert config.integration_id == "bb-1"


def test_snowflake_tool_enforces_bounded_limit(monkeypatch: Any) -> None:
    captured: dict[str, Any] = {}

    def _fake_post(
        url: str, headers: dict[str, str], json: dict[str, Any], timeout: float
    ) -> _MockResponse:
        captured["url"] = url
        captured["statement"] = json["statement"]
        captured["timeout"] = timeout
        return _MockResponse({"data": [{"id": idx} for idx in range(20)]})

    monkeypatch.setattr(
        "integrations.snowflake.tools.snowflake_query_history_tool.httpx.post", _fake_post
    )

    result = query_snowflake_history(
        account_identifier="xy12345.us-east-1",
        token="sf-token",
        query="SELECT * FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY())",
        limit=500,
        max_results=6,
    )

    assert "LIMIT 6" in captured["statement"].upper()
    assert result["available"] is True
    assert len(result["rows"]) == 6


def test_snowflake_tool_requires_token() -> None:
    result = query_snowflake_history(
        account_identifier="xy12345.us-east-1",
        user="service-user",
        password="secret",
    )

    assert result["available"] is False
    assert result["error"] == "Missing Snowflake token."


def test_azure_tool_enforces_bounded_take_clause(monkeypatch: Any) -> None:
    captured: dict[str, Any] = {}

    def _fake_post(
        url: str, headers: dict[str, str], json: dict[str, Any], timeout: float
    ) -> _MockResponse:
        captured["url"] = url
        captured["query"] = json["query"]
        return _MockResponse(
            {
                "tables": [
                    {
                        "columns": [{"name": "TimeGenerated"}, {"name": "Message"}],
                        "rows": [[f"t{idx}", f"message-{idx}"] for idx in range(10)],
                    }
                ]
            }
        )

    monkeypatch.setattr("integrations.azure.tools.azure_monitor_logs_tool.httpx.post", _fake_post)

    result = query_azure_monitor_logs(
        workspace_id="workspace-1",
        access_token="azure-token",
        query="AppTraces | order by TimeGenerated desc",
        limit=999,
        max_results=3,
    )

    assert "take 3" in captured["query"].lower()
    assert result["available"] is True
    assert len(result["rows"]) == 3


def test_openobserve_tool_caps_size_and_output(monkeypatch: Any) -> None:
    captured: dict[str, Any] = {}

    def _fake_post(
        url: str, headers: dict[str, str], json: dict[str, Any], timeout: float
    ) -> _MockResponse:
        captured["url"] = url
        captured["size"] = json["size"]
        captured["sql"] = json["query"]["sql"]
        return _MockResponse({"hits": [{"message": f"m{idx}"} for idx in range(12)]})

    monkeypatch.setattr(
        "integrations.openobserve.tools.openobserve_logs_tool.httpx.post", _fake_post
    )

    result = query_openobserve_logs(
        base_url="https://openobserve.example.invalid",
        org="acme",
        api_token="oo-token",
        limit=1000,
        max_results=4,
    )

    assert captured["size"] == 4
    assert (
        captured["sql"]
        == "SELECT * FROM \"default\" WHERE level = 'error' ORDER BY _timestamp DESC"
    )
    assert result["available"] is True
    assert len(result["records"]) == 4


def test_opensearch_tool_caps_limit_before_client_query(monkeypatch: Any) -> None:
    captured: dict[str, Any] = {}

    def _fake_search_logs(
        self: Any,
        query: str = "*",
        time_range_minutes: int = 60,
        limit: int = 50,
        index_pattern: str | None = None,
        timestamp_field: str = "@timestamp",
    ) -> dict[str, Any]:
        _ = (query, time_range_minutes, index_pattern, timestamp_field)
        captured["limit"] = limit
        return {"success": True, "logs": [{"message": f"log-{idx}"} for idx in range(12)]}

    monkeypatch.setattr(
        "integrations.opensearch.tools.opensearch_analytics_tool.ElasticsearchClient.search_logs",
        _fake_search_logs,
    )

    result = query_opensearch_analytics(
        url="https://opensearch.example.invalid",
        query="error",
        limit=500,
        max_results=5,
    )

    assert captured["limit"] == 5
    assert result["available"] is True
    assert len(result["logs"]) == 5


def test_opensearch_tool_forwards_basic_auth_to_elasticsearch_config(monkeypatch: Any) -> None:
    """Layer 5 / #1143: username and password must reach ElasticsearchConfig.

    Without this wiring, even though the user configures Basic Auth via the wizard
    or the legacy CLI, the AI agent's OpenSearch tool drops the credentials when
    constructing the runtime client, so the LLM cannot authenticate against the
    cluster during investigations.
    """
    captured: dict[str, Any] = {}

    class _FakeConfig:
        def __init__(
            self,
            url: str,
            api_key: str | None = None,
            username: str | None = None,
            password: str | None = None,
            index_pattern: str = "*",
        ) -> None:
            captured["url"] = url
            captured["api_key"] = api_key
            captured["username"] = username
            captured["password"] = password
            captured["index_pattern"] = index_pattern

    def _fake_search_logs(
        self: Any,
        query: str = "*",
        time_range_minutes: int = 60,
        limit: int = 50,
        index_pattern: str | None = None,
        timestamp_field: str = "@timestamp",
    ) -> dict[str, Any]:
        return {"success": True, "logs": []}

    monkeypatch.setattr(
        "integrations.opensearch.tools.opensearch_analytics_tool.ElasticsearchConfig",
        _FakeConfig,
    )
    monkeypatch.setattr(
        "integrations.opensearch.tools.opensearch_analytics_tool.ElasticsearchClient.search_logs",
        _fake_search_logs,
    )

    result = query_opensearch_analytics(
        url="https://opensearch.example.invalid",
        username="admin",
        password="secret",
        query="*",
    )

    assert captured["username"] == "admin"
    assert captured["password"] == "secret"
    assert result["available"] is True


def test_opensearch_tool_extract_params_reads_basic_auth() -> None:
    """Layer 5 / #1143: _opensearch_extract_params must surface username/password.

    These keys are populated by the catalog classifier (Layer 2) when a user
    configures Basic Auth, and the registered tool's runtime kwargs must
    include them so they reach ElasticsearchConfig.
    """
    from integrations.opensearch.tools.opensearch_analytics_tool import _opensearch_extract_params

    sources = {
        "opensearch": {
            "connection_verified": True,
            "url": "https://opensearch.example.invalid",
            "username": "admin",
            "password": "secret",
        }
    }
    params = _opensearch_extract_params(sources)
    assert params["username"] == "admin"
    assert params["password"] == "secret"