hy-app-monitor/tests/test_mysql_permissions.py

from __future__ import annotations

import unittest
from unittest.mock import patch

from monitors.yumi.mysql_common import (
    database_exists,
    list_database_summaries,
    list_table_columns,
    list_table_indexes,
    table_exists,
)
from monitors.yumi.mysql_metrics import build_mysql_overview_payload


def normalize_sql(sql: str) -> str:
    return " ".join(str(sql).split()).strip().upper()


class FakeCursor:
    def __init__(self, handler) -> None:
        self._handler = handler
        self._result = None

    def __enter__(self) -> "FakeCursor":
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        return False

    def execute(self, query: str, params=None) -> None:
        response = self._handler(normalize_sql(query), params)
        if isinstance(response, Exception):
            raise response
        self._result = response

    def fetchall(self):
        if self._result is None:
            return []
        if isinstance(self._result, list):
            return self._result
        return [self._result]

    def fetchone(self):
        if isinstance(self._result, list):
            return self._result[0] if self._result else None
        return self._result


class FakeConnection:
    def __init__(self, handler, db: str = "") -> None:
        self._handler = handler
        self.db = db

    def cursor(self) -> FakeCursor:
        return FakeCursor(self._handler)


class DummyOverviewConnection:
    def cursor(self) -> FakeCursor:
        def handler(query: str, params=None):
            if query == "SELECT 1 AS OK":
                return {"ok": 1}
            raise AssertionError(f"unexpected query: {query} {params!r}")

        return FakeCursor(handler)

    def __enter__(self) -> "DummyOverviewConnection":
        return self

    def __exit__(self, exc_type, exc, tb) -> bool:
        return False


class MysqlPermissionFallbackTests(unittest.TestCase):
    def test_list_database_summaries_falls_back_to_show_commands(self) -> None:
        def handler(query: str, params=None):
            if query.startswith("SELECT S.SCHEMA_NAME AS SCHEMANAME"):
                return RuntimeError("Access denied for information_schema")
            if query == "SHOW DATABASES":
                return [{"Database": "biz"}, {"Database": "mysql"}]
            if query == "SHOW TABLE STATUS FROM `BIZ`":
                return [{"Name": "orders", "Rows": "8", "Data_length": "1024", "Index_length": "256"}]
            if query == "SHOW TABLE STATUS FROM `MYSQL`":
                return []
            raise AssertionError(f"unexpected query: {query} {params!r}")

        items = list_database_summaries(FakeConnection(handler))

        self.assertEqual(items[0]["name"], "biz")
        self.assertEqual(items[0]["tableTotal"], 1)
        self.assertEqual(items[0]["rowEstimate"], 8)
        self.assertFalse(items[0]["system"])
        self.assertEqual(items[1]["name"], "mysql")
        self.assertTrue(items[1]["system"])

    def test_table_metadata_falls_back_to_show_commands(self) -> None:
        def handler(query: str, params=None):
            if query.startswith("SELECT COLUMN_NAME AS COLUMNNAME"):
                return RuntimeError("Access denied for columns")
            if query == "SHOW FULL COLUMNS FROM `ORDERS` FROM `BIZ`":
                return [
                    {
                        "Field": "id",
                        "Type": "bigint unsigned",
                        "Null": "NO",
                        "Default": None,
                        "Key": "PRI",
                        "Extra": "auto_increment",
                        "Comment": "primary key",
                    }
                ]
            if query.startswith("SELECT INDEX_NAME AS INDEXNAME"):
                return RuntimeError("Access denied for statistics")
            if query == "SHOW INDEX FROM `ORDERS` FROM `BIZ`":
                return [
                    {
                        "Key_name": "PRIMARY",
                        "Column_name": "id",
                        "Non_unique": 0,
                        "Index_type": "BTREE",
                        "Seq_in_index": 1,
                        "Sub_part": None,
                    }
                ]
            raise AssertionError(f"unexpected query: {query} {params!r}")

        columns = list_table_columns(FakeConnection(handler), "biz", "orders")
        indexes = list_table_indexes(FakeConnection(handler), "biz", "orders")

        self.assertEqual(columns[0]["name"], "id")
        self.assertEqual(columns[0]["dataType"], "bigint")
        self.assertEqual(columns[0]["ordinalPosition"], 1)
        self.assertEqual(indexes[0]["name"], "PRIMARY")
        self.assertTrue(indexes[0]["primary"])
        self.assertTrue(indexes[0]["unique"])

    def test_exists_checks_fall_back_to_show_commands(self) -> None:
        def handler(query: str, params=None):
            if "FROM INFORMATION_SCHEMA.SCHEMATA" in query:
                return RuntimeError("Access denied for schemata")
            if "FROM INFORMATION_SCHEMA.TABLES" in query:
                return RuntimeError("Access denied for tables")
            if query == "SHOW DATABASES":
                return [{"Database": "biz"}]
            if query == "SHOW FULL TABLES FROM `BIZ`":
                return [{"Tables_in_biz": "orders", "Table_type": "BASE TABLE"}]
            raise AssertionError(f"unexpected query: {query} {params!r}")

        connection = FakeConnection(handler)

        self.assertTrue(database_exists(connection, "biz"))
        self.assertTrue(table_exists(connection, "biz", "orders"))
        self.assertFalse(table_exists(connection, "biz", "missing"))

    @patch("monitors.yumi.mysql_metrics.list_database_summaries", return_value=[{"name": "biz", "system": False, "tableTotal": 3, "dataSizeMb": 1.0, "indexSizeMb": 0.5, "totalSizeMb": 1.5}])
    @patch("monitors.yumi.mysql_metrics.collect_replication_payload", side_effect=RuntimeError("need REPLICATION CLIENT"))
    @patch("monitors.yumi.mysql_metrics.collect_global_status", side_effect=RuntimeError("need PROCESS"))
    @patch("monitors.yumi.mysql_metrics.collect_global_variables", side_effect=RuntimeError("need PROCESS"))
    @patch("monitors.yumi.mysql_metrics.build_mysql_connection")
    def test_overview_keeps_ok_when_partial_permissions_missing(
        self,
        build_connection,
        collect_global_variables,
        collect_global_status,
        collect_replication_payload,
        list_database_summaries_mock,
    ) -> None:
        del collect_global_variables, collect_global_status, collect_replication_payload, list_database_summaries_mock
        build_connection.return_value = DummyOverviewConnection()

        payload = build_mysql_overview_payload()

        self.assertTrue(payload["ok"])
        self.assertEqual(payload["summary"]["databaseTotal"], 1)
        self.assertEqual(payload["connections"]["current"], 0)
        self.assertEqual(payload["replication"]["role"], "primary")
        self.assertEqual(len(payload["warnings"]), 3)


if __name__ == "__main__":
    unittest.main()