from __future__ import annotations

import os
import unittest
from decimal import Decimal
from pathlib import Path
from unittest.mock import MagicMock, patch

import pymysql

from backend.sql_catalog.database import (
    DatabaseRow,
    MySQLConnectionPool,
    MySQLCursor,
    _qmark_to_mysql,
    connect_mysql,
    database_description,
    finding_run_idempotency_key,
    split_sql_script,
)


class DatabaseAdapterTest(unittest.TestCase):
    def test_connection_pool_can_be_warmed_for_parallel_page_requests(self) -> None:
        connections = [MagicMock(name=f"connection-{index}") for index in range(3)]
        create = MagicMock(side_effect=connections)
        pool = MySQLConnectionPool(create, max_size=3)

        pool.warm(3)

        self.assertEqual(3, create.call_count)
        acquired = [pool.acquire() for _ in range(3)]
        self.assertCountEqual(connections, acquired)
        for connection in acquired:
            pool.release(connection, transaction_resolved=True)

    def test_database_row_supports_mapping_and_numeric_access(self) -> None:
        row = DatabaseRow(("id", "name"), (7, "catalog"))

        self.assertEqual(7, row["id"])
        self.assertEqual(7, row[0])
        self.assertEqual("catalog", row[1])
        self.assertEqual({"id": 7, "name": "catalog"}, dict(row))

    def test_mysql_cursor_normalizes_decimal_results_for_sqlite_compatibility(self) -> None:
        cursor = MagicMock()
        cursor.description = (("whole",), ("fraction",))
        cursor.rowcount = 1
        cursor.fetchone.return_value = (Decimal("16"), Decimal("1.25"))

        row = MySQLCursor(cursor).fetchone()

        self.assertEqual(16, row["whole"])
        self.assertIsInstance(row["whole"], int)
        self.assertEqual(1.25, row["fraction"])
        self.assertIsInstance(row["fraction"], float)
        cursor.close.assert_called_once_with()

    def test_qmark_conversion_skips_quoted_text_and_comments(self) -> None:
        sql = """
        SELECT '?', "?", `?`, value
        FROM findings
        WHERE id = ? AND note = 'it''s ?' -- keep ?
          AND source = ? /* keep ? */
        """

        converted = _qmark_to_mysql(sql)

        self.assertIn("WHERE id = %s", converted)
        self.assertIn("AND source = %s", converted)
        self.assertIn("SELECT '?'", converted)
        self.assertIn("-- keep ?", converted)
        self.assertIn("/* keep ? */", converted)

    def test_split_sql_script_preserves_semicolons_in_literals(self) -> None:
        statements = split_sql_script(
            "INSERT INTO records VALUES ('one;two'); SELECT `semi;colon`;"
        )

        self.assertEqual(
            [
                "INSERT INTO records VALUES ('one;two')",
                "SELECT `semi;colon`",
            ],
            statements,
        )

    def test_database_description_never_contains_credentials(self) -> None:
        description = database_description(
            "mysql://user:secret@example.com:4000/sql_catalog?ssl=true",
            Path("unused.sqlite3"),
        )

        self.assertEqual("mysql://example.com:4000/sql_catalog", description)
        self.assertNotIn("user", description)
        self.assertNotIn("secret", description)

    def test_finding_run_idempotency_key_is_stable_and_delimiter_safe(self) -> None:
        first = finding_run_idempotency_key("a", "bc", "d", "ef")
        second = finding_run_idempotency_key("ab", "c", "d", "ef")

        self.assertEqual(32, len(first))
        self.assertEqual(first, finding_run_idempotency_key("a", "bc", "d", "ef"))
        self.assertNotEqual(first, second)

    def test_connect_mysql_retries_transient_gateway_disconnect(self) -> None:
        raw_connection = MagicMock()
        url = "mysql://user@retry-test.invalid:4000/sql_catalog?ssl=false"
        with (
            patch.dict(
                os.environ,
                {
                    "TEST_DB_PASSWORD": "secret",
                    "SQL_CATALOG_DB_CONNECT_ATTEMPTS": "3",
                    "SQL_CATALOG_DB_CONNECT_RETRY_DELAY": "0",
                },
            ),
            patch(
                "pymysql.connect",
                side_effect=[
                    pymysql.err.OperationalError(2013, "gateway waking"),
                    raw_connection,
                ],
            ) as connect,
            patch.dict(
                "backend.sql_catalog.database._MYSQL_POOLS", {}, clear=True
            ),
            patch("backend.sql_catalog.database.time.sleep") as sleep,
        ):
            connection = connect_mysql(url, "TEST_DB_PASSWORD")
            connection.close()

        self.assertEqual(2, connect.call_count)
        sleep.assert_called_once_with(0.0)


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