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

"""Unit tests for MariaDB integration."""

from __future__ import annotations

import os
import sys
import types
from unittest.mock import MagicMock, patch

import pytest

from integrations.catalog import classify_integrations as _classify_integrations
from integrations.mariadb import (
    _QUERY_TRUNCATE_LEN,
    DEFAULT_MARIADB_PORT,
    MariaDBConfig,
    build_mariadb_config,
    get_global_status,
    get_innodb_status,
    get_process_list,
    get_replication_status,
    get_slow_queries,
    mariadb_config_from_env,
    mariadb_extract_params,
    mariadb_is_available,
    validate_mariadb_config,
)
from platform.common.truncation import truncate as _truncate_shared


def _truncate(text: str, max_len: int = _QUERY_TRUNCATE_LEN) -> str:
    return _truncate_shared(text, max_len)


# ---------------------------------------------------------------------------
# Shared mock helpers
# ---------------------------------------------------------------------------


def _make_mock_conn(
    fetchall_rows: list | None = None,
    fetchone_row: tuple | None = None,
    description: list | None = None,
) -> tuple[MagicMock, MagicMock]:
    """Return (mock_conn, mock_cursor) wired for use with `with conn.cursor() as cur:`."""
    mock_conn = MagicMock()
    mock_cursor = MagicMock()
    mock_cursor.fetchall.return_value = fetchall_rows or []
    mock_cursor.fetchone.return_value = fetchone_row
    mock_cursor.description = description
    mock_conn.cursor.return_value.__enter__ = lambda _self: mock_cursor
    mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
    return mock_conn, mock_cursor


class TestMariaDBConfig:
    def test_default_values(self):
        config = MariaDBConfig(host="localhost", database="testdb")
        assert config.port == 3306
        assert config.ssl is True
        assert config.timeout_seconds == 5
        assert config.max_results == 50
        assert config.username == ""
        assert config.password == ""

    def test_normalization(self):
        config = MariaDBConfig(
            host="  db.example.com  ",
            database="  testdb  ",
            username="  admin  ",
        )
        assert config.host == "db.example.com"
        assert config.database == "testdb"
        assert config.username == "admin"

    def test_is_configured(self):
        assert MariaDBConfig(host="host", database="db").is_configured is True
        assert MariaDBConfig(host="", database="db").is_configured is False
        assert MariaDBConfig(host="host", database="").is_configured is False


class TestMariaDBBuild:
    def test_build_mariadb_config(self):
        raw = {
            "host": "db.example.com",
            "port": 3307,
            "database": "foo",
            "username": "admin",
            "password": "secret",
            "ssl": False,
        }
        config = build_mariadb_config(raw)
        assert config.host == "db.example.com"
        assert config.port == 3307
        assert config.database == "foo"
        assert config.username == "admin"
        assert config.ssl is False

    @patch.dict(
        os.environ,
        {
            "MARIADB_HOST": "env-host",
            "MARIADB_PORT": "3307",
            "MARIADB_DATABASE": "env-db",
            "MARIADB_USERNAME": "env-user",
            "MARIADB_PASSWORD": "env-pass",
            "MARIADB_SSL": "false",
        },
    )
    def test_mariadb_config_from_env(self):
        config = mariadb_config_from_env()
        assert config is not None
        assert config.host == "env-host"
        assert config.port == 3307
        assert config.database == "env-db"
        assert config.username == "env-user"
        assert config.ssl is False

    @patch.dict(os.environ, {}, clear=True)
    def test_mariadb_config_from_env_missing(self):
        assert mariadb_config_from_env() is None


class TestMariaDBValidation:
    @patch("integrations.mariadb._get_connection")
    def test_validate_success(self, mock_get_conn):
        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_cursor.fetchone.return_value = ("10.11.6-MariaDB",)
        mock_conn.cursor.return_value.__enter__ = lambda _self: mock_cursor
        mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
        mock_get_conn.return_value = mock_conn

        config = MariaDBConfig(host="host", database="test", username="user")
        result = validate_mariadb_config(config)

        assert result.ok is True
        assert "10.11.6-MariaDB" in result.detail
        assert "test" in result.detail
        mock_conn.close.assert_called_once()

    @patch("integrations.mariadb._get_connection", side_effect=Exception("Conn error"))
    def test_validate_exception(self, _):
        config = MariaDBConfig(host="host", database="test", username="user")
        result = validate_mariadb_config(config)
        assert result.ok is False
        assert "Conn error" in result.detail

    def test_validate_missing_host(self):
        config = MariaDBConfig(host="", database="test")
        result = validate_mariadb_config(config)
        assert result.ok is False
        assert "required" in result.detail

    def test_validate_missing_database(self):
        config = MariaDBConfig(host="host", database="")
        result = validate_mariadb_config(config)
        assert result.ok is False
        assert "required" in result.detail


class TestResolveIntegrations:
    def test_classify_mariadb(self):
        integrations = [
            {
                "id": "123",
                "service": "mariadb",
                "status": "active",
                "credentials": {
                    "host": "db.example.com",
                    "port": 3306,
                    "database": "prod",
                    "username": "admin",
                    "password": "secret",
                },
            }
        ]
        resolved = _classify_integrations(integrations)
        assert "mariadb" in resolved
        assert resolved["mariadb"].host == "db.example.com"
        assert resolved["mariadb"].database == "prod"
        assert resolved["mariadb"].port == 3306

    def test_classify_mariadb_skipped_without_host(self):
        integrations = [
            {
                "id": "789",
                "service": "mariadb",
                "status": "active",
                "credentials": {
                    "host": "",
                    "database": "prod",
                    "username": "admin",
                },
            }
        ]
        resolved = _classify_integrations(integrations)
        assert "mariadb" not in resolved

    def test_classify_mariadb_skipped_without_database(self):
        integrations = [
            {
                "id": "456",
                "service": "mariadb",
                "status": "active",
                "credentials": {
                    "host": "db.example.com",
                    "database": "",
                    "username": "admin",
                },
            }
        ]
        resolved = _classify_integrations(integrations)
        assert "mariadb" not in resolved


# ---------------------------------------------------------------------------
# Truncation helper
# ---------------------------------------------------------------------------


class TestTruncate:
    def test_short_string_unchanged(self) -> None:
        assert _truncate("SELECT 1") == "SELECT 1"

    def test_exact_limit_unchanged(self) -> None:
        text = "x" * 200
        assert _truncate(text) == text

    def test_over_limit_truncated(self) -> None:
        result = _truncate("x" * 201)
        assert result.endswith("...")
        assert len(result) == 200  # capped at limit

    def test_empty_string(self) -> None:
        assert _truncate("") == ""

    def test_custom_max_len(self) -> None:
        assert _truncate("hello world", max_len=5) == "he..."

    def test_very_long_string(self) -> None:
        result = _truncate("A" * 10_000)
        assert result.endswith("...")
        assert len(result) == 200  # capped at limit


# ---------------------------------------------------------------------------
# Helper utilities
# ---------------------------------------------------------------------------


class TestMariaDBIsAvailable:
    def test_true_with_host_and_database(self) -> None:
        assert (
            mariadb_is_available({"mariadb": {"host": "db.example.com", "database": "prod"}})
            is True
        )

    def test_false_when_missing_host(self) -> None:
        assert mariadb_is_available({"mariadb": {"host": "", "database": "prod"}}) is False

    def test_false_when_missing_database(self) -> None:
        assert mariadb_is_available({"mariadb": {"host": "h", "database": ""}}) is False

    def test_false_when_mariadb_key_absent(self) -> None:
        assert mariadb_is_available({}) is False

    def test_false_when_mariadb_is_empty_dict(self) -> None:
        assert mariadb_is_available({"mariadb": {}}) is False

    def test_false_when_host_is_none(self) -> None:
        assert mariadb_is_available({"mariadb": {"host": None, "database": "prod"}}) is False


class TestMariaDBExtractParams:
    def test_all_fields_returned(self) -> None:
        sources = {
            "mariadb": {
                "host": "h",
                "database": "d",
                "username": "u",
                "password": "p",
                "port": 3307,
                "ssl": False,
            }
        }
        params = mariadb_extract_params(sources)
        assert params["host"] == "h"
        assert params["database"] == "d"
        assert params["username"] == "u"
        assert params["password"] == "p"
        assert params["port"] == 3307
        assert params["ssl"] is False

    def test_defaults_when_keys_missing(self) -> None:
        params = mariadb_extract_params({"mariadb": {}})
        assert params["host"] == ""
        assert params["port"] == DEFAULT_MARIADB_PORT
        assert params["ssl"] is True

    def test_empty_sources_dict(self) -> None:
        params = mariadb_extract_params({})
        assert params["host"] == ""
        assert params["port"] == DEFAULT_MARIADB_PORT


# ---------------------------------------------------------------------------
# Query functions
# ---------------------------------------------------------------------------


class TestGetProcessList:
    def _config(self) -> MariaDBConfig:
        return MariaDBConfig(host="host", database="db", username="user")

    @patch("integrations.mariadb._get_connection")
    def test_returns_process_list_on_success(self, mock_get_conn: MagicMock) -> None:
        rows = [(1, "root", "localhost", "mydb", "Query", 5, "executing", "SELECT sleep(5)")]
        mock_conn, _ = _make_mock_conn(fetchall_rows=rows)
        mock_get_conn.return_value = mock_conn

        result = get_process_list(self._config())

        assert result["available"] is True
        assert result["source"] == "mariadb"
        assert result["total_processes"] == 1
        proc = result["processes"][0]
        assert proc["id"] == 1
        assert proc["user"] == "root"
        assert proc["command"] == "Query"
        assert proc["time_secs"] == 5
        assert proc["query"] == "SELECT sleep(5)"
        mock_conn.close.assert_called_once()

    @patch("integrations.mariadb._get_connection")
    def test_query_text_is_truncated(self, mock_get_conn: MagicMock) -> None:
        long_query = "SELECT " + "x" * 300
        rows = [(1, "root", "localhost", "db", "Query", 1, "", long_query)]
        mock_conn, _ = _make_mock_conn(fetchall_rows=rows)
        mock_get_conn.return_value = mock_conn

        result = get_process_list(self._config())

        assert result["processes"][0]["query"].endswith("...")
        assert len(result["processes"][0]["query"]) == 200

    @patch("integrations.mariadb._get_connection")
    def test_empty_process_list(self, mock_get_conn: MagicMock) -> None:
        mock_conn, _ = _make_mock_conn(fetchall_rows=[])
        mock_get_conn.return_value = mock_conn

        result = get_process_list(self._config())

        assert result["available"] is True
        assert result["total_processes"] == 0
        assert result["processes"] == []

    @patch("integrations.mariadb._get_connection")
    def test_max_results_override(self, mock_get_conn: MagicMock) -> None:
        mock_conn, mock_cursor = _make_mock_conn(fetchall_rows=[])
        mock_get_conn.return_value = mock_conn

        get_process_list(self._config(), max_results=10)

        sql_call = mock_cursor.execute.call_args
        assert sql_call[0][1] == (10,)

    @patch("integrations.mariadb._get_connection", side_effect=Exception("timeout"))
    def test_exception_returns_error(self, _: MagicMock) -> None:
        result = get_process_list(self._config())
        assert result["available"] is False
        assert "timeout" in result["error"]

    def test_not_configured_returns_error(self) -> None:
        result = get_process_list(MariaDBConfig())
        assert result["available"] is False
        assert "Not configured" in result["error"]


class TestGetGlobalStatus:
    def _config(self) -> MariaDBConfig:
        return MariaDBConfig(host="host", database="db", username="user")

    @patch("integrations.mariadb._get_connection")
    def test_returns_curated_metrics(self, mock_get_conn: MagicMock) -> None:
        all_status = [
            ("Threads_connected", "5"),
            ("Threads_running", "2"),
            ("Uptime", "86400"),
            ("Slow_queries", "10"),
            ("SomeOtherVar", "ignored"),
        ]
        mock_conn, _ = _make_mock_conn(fetchall_rows=all_status)
        mock_get_conn.return_value = mock_conn

        result = get_global_status(self._config())

        assert result["available"] is True
        assert result["source"] == "mariadb"
        assert result["metrics"]["Threads_connected"] == "5"
        assert "Uptime" in result["metrics"]
        assert "SomeOtherVar" not in result["metrics"]
        mock_conn.close.assert_called_once()

    @patch("integrations.mariadb._get_connection")
    def test_missing_keys_skipped_gracefully(self, mock_get_conn: MagicMock) -> None:
        mock_conn, _ = _make_mock_conn(fetchall_rows=[("Uptime", "3600")])
        mock_get_conn.return_value = mock_conn

        result = get_global_status(self._config())

        assert result["available"] is True
        assert result["metrics"]["Uptime"] == "3600"

    @patch("integrations.mariadb._get_connection", side_effect=Exception("access denied"))
    def test_exception_returns_error(self, _: MagicMock) -> None:
        result = get_global_status(self._config())
        assert result["available"] is False
        assert "access denied" in result["error"]

    def test_not_configured_returns_error(self) -> None:
        result = get_global_status(MariaDBConfig())
        assert result["available"] is False


class TestGetInnoDBStatus:
    def _config(self) -> MariaDBConfig:
        return MariaDBConfig(host="host", database="db", username="user")

    @patch("integrations.mariadb._get_connection")
    def test_returns_innodb_status_text(self, mock_get_conn: MagicMock) -> None:
        status_text = "================\nBUFFER POOL AND MEMORY\n================"
        mock_conn, _ = _make_mock_conn(fetchone_row=("InnoDB", "", status_text))
        mock_get_conn.return_value = mock_conn

        result = get_innodb_status(self._config())

        assert result["available"] is True
        assert result["source"] == "mariadb"
        assert "BUFFER POOL" in result["innodb_status"]
        mock_conn.close.assert_called_once()

    @patch("integrations.mariadb._get_connection")
    def test_long_status_text_truncated(self, mock_get_conn: MagicMock) -> None:
        long_text = "x" * 5000
        mock_conn, _ = _make_mock_conn(fetchone_row=("InnoDB", "", long_text))
        mock_get_conn.return_value = mock_conn

        result = get_innodb_status(self._config())

        assert result["available"] is True
        assert result["innodb_status"].endswith("(truncated)")
        assert len(result["innodb_status"]) < 5000

    @patch("integrations.mariadb._get_connection")
    def test_none_row_returns_empty_string(self, mock_get_conn: MagicMock) -> None:
        mock_conn, _ = _make_mock_conn(fetchone_row=None)
        mock_get_conn.return_value = mock_conn

        result = get_innodb_status(self._config())

        assert result["available"] is True
        assert result["innodb_status"] == ""

    @patch("integrations.mariadb._get_connection", side_effect=Exception("no privilege"))
    def test_exception_returns_error(self, _: MagicMock) -> None:
        result = get_innodb_status(self._config())
        assert result["available"] is False
        assert "no privilege" in result["error"]

    def test_not_configured_returns_error(self) -> None:
        result = get_innodb_status(MariaDBConfig())
        assert result["available"] is False


class TestGetSlowQueries:
    def _config(self) -> MariaDBConfig:
        return MariaDBConfig(host="host", database="mydb", username="user")

    @patch("integrations.mariadb._get_connection")
    def test_returns_slow_queries_when_perf_schema_enabled(self, mock_get_conn: MagicMock) -> None:
        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        query_rows = [("SELECT * FROM users", 100, 50.5, 5050.0, 1000, 100)]
        mock_cursor.fetchone.side_effect = [(1,)]
        mock_cursor.fetchall.return_value = query_rows
        mock_conn.cursor.return_value.__enter__ = lambda _self: mock_cursor
        mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
        mock_get_conn.return_value = mock_conn

        result = get_slow_queries(self._config())

        assert result["available"] is True
        assert result["source"] == "mariadb"
        assert result["total_queries"] == 1
        q = result["queries"][0]
        assert q["digest_text"] == "SELECT * FROM users"
        assert q["count"] == 100
        assert q["avg_time_ms"] == pytest.approx(50.5)
        assert q["rows_examined"] == 1000
        mock_conn.close.assert_called_once()

    @patch("integrations.mariadb._get_connection")
    def test_returns_note_when_perf_schema_disabled(self, mock_get_conn: MagicMock) -> None:
        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_cursor.fetchone.return_value = (0,)
        mock_conn.cursor.return_value.__enter__ = lambda _self: mock_cursor
        mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
        mock_get_conn.return_value = mock_conn

        result = get_slow_queries(self._config())

        assert result["available"] is True
        assert "note" in result
        assert "performance_schema" in result["note"]
        assert result["queries"] == []

    @patch("integrations.mariadb._get_connection", side_effect=Exception("query error"))
    def test_exception_returns_error(self, _: MagicMock) -> None:
        result = get_slow_queries(self._config())
        assert result["available"] is False
        assert "query error" in result["error"]

    def test_not_configured_returns_error(self) -> None:
        result = get_slow_queries(MariaDBConfig())
        assert result["available"] is False


class TestGetReplicationStatus:
    def _config(self) -> MariaDBConfig:
        return MariaDBConfig(host="host", database="db", username="user")

    @patch("integrations.mariadb._get_connection")
    def test_returns_channels_on_success(self, mock_get_conn: MagicMock) -> None:
        columns = [
            "Slave_IO_Running",
            "Slave_SQL_Running",
            "Seconds_Behind_Master",
            "Last_Error",
            "Last_Errno",
            "Master_Host",
            "Master_Port",
            "Master_Log_File",
            "Relay_Log_Space",
            "Exec_Master_Log_Pos",
            "Connection_name",
        ]
        row = ("Yes", "Yes", 0, "", 0, "primary.db", 3306, "binlog.000001", 1024, 512, "")
        mock_conn, _ = _make_mock_conn(
            fetchall_rows=[row],
            description=[(col,) for col in columns],
        )
        mock_get_conn.return_value = mock_conn

        result = get_replication_status(self._config())

        assert result["available"] is True
        assert result["source"] == "mariadb"
        assert len(result["channels"]) == 1
        ch = result["channels"][0]
        assert ch["Slave_IO_Running"] == "Yes"
        assert ch["Master_Host"] == "primary.db"
        mock_conn.close.assert_called_once()

    @patch("integrations.mariadb._get_connection")
    def test_not_replica_returns_note(self, mock_get_conn: MagicMock) -> None:
        mock_conn, _ = _make_mock_conn(fetchall_rows=[], description=None)
        mock_get_conn.return_value = mock_conn

        result = get_replication_status(self._config())

        assert result["available"] is True
        assert "not configured as a replica" in result["note"]
        assert result["channels"] == []

    @patch("integrations.mariadb._get_connection")
    def test_fallback_to_show_slave_status(self, mock_get_conn: MagicMock) -> None:
        # Stub pymysql so this test does not require PyMySQL at import time (matches the
        # dynamic `import pymysql` inside get_replication_status error handling).
        class _ProgrammingError(Exception):
            pass

        _fake_err = types.ModuleType("pymysql.err")
        _fake_err.ProgrammingError = _ProgrammingError
        _fake_pymysql = types.ModuleType("pymysql")
        _fake_pymysql.err = _fake_err
        _prior = sys.modules.get("pymysql")
        sys.modules["pymysql"] = _fake_pymysql
        try:
            mock_conn = MagicMock()
            mock_cursor = MagicMock()
            columns = ["Slave_IO_Running", "Slave_SQL_Running"]
            row = ("Yes", "Yes")
            execute_call_count = 0

            def execute_side_effect(stmt: str, *args: object, **kwargs: object) -> None:
                nonlocal execute_call_count
                execute_call_count += 1
                if "ALL SLAVES" in stmt:
                    raise _ProgrammingError(1064, "syntax error")

            mock_cursor.execute.side_effect = execute_side_effect
            mock_cursor.fetchall.return_value = [row]
            mock_cursor.description = [(col,) for col in columns]
            mock_conn.cursor.return_value.__enter__ = lambda _self: mock_cursor
            mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
            mock_get_conn.return_value = mock_conn

            result = get_replication_status(self._config())

            assert result["available"] is True
            assert len(result["channels"]) == 1
            assert execute_call_count == 2
        finally:
            if _prior is not None:
                sys.modules["pymysql"] = _prior
            else:
                sys.modules.pop("pymysql", None)

    @patch("integrations.mariadb._get_connection")
    def test_partial_columns_handled(self, mock_get_conn: MagicMock) -> None:
        columns = ["Slave_IO_Running", "Slave_SQL_Running"]
        mock_conn, _ = _make_mock_conn(
            fetchall_rows=[("Yes", "Yes")],
            description=[(col,) for col in columns],
        )
        mock_get_conn.return_value = mock_conn

        result = get_replication_status(self._config())

        assert result["available"] is True
        ch = result["channels"][0]
        assert ch["Slave_IO_Running"] == "Yes"
        assert "Master_Host" not in ch

    @patch("integrations.mariadb._get_connection", side_effect=Exception("network error"))
    def test_exception_returns_error(self, _: MagicMock) -> None:
        result = get_replication_status(self._config())
        assert result["available"] is False
        assert "network error" in result["error"]

    def test_not_configured_returns_error(self) -> None:
        result = get_replication_status(MariaDBConfig())
        assert result["available"] is False