"""Unit tests for src.services.duckdb_delta (DuckDB Delta Lake service)."""

from unittest.mock import MagicMock, patch

import pytest

from src.services import duckdb_delta
from src.services.duckdb_delta import read_delta
from src.utils.errors import DeltaError


class TestEscapeSqlString:
    """_escape_sql_string escapes single quotes for SQL."""

    def test_no_quotes_unchanged(self):
        assert duckdb_delta._escape_sql_string("path/to/table") == "path/to/table"

    def test_single_quote_doubled(self):
        assert duckdb_delta._escape_sql_string("a'b") == "a''b"

    def test_multiple_quotes_doubled(self):
        assert duckdb_delta._escape_sql_string("a'b'c") == "a''b''c"


class TestReadDeltaSafeQuery:
    """read_delta optional query must match safe pattern and contain FROM __t__."""

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_valid_custom_query_uses_delta_scan(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = False
        mock_df.to_dict.return_value = [{"a": 1}]
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn

        read_delta("file:///tmp/delta", "local", query="SELECT a FROM __t__ LIMIT 5")
        calls = mock_conn.execute.call_args_list
        select_calls = [
            c[0][0] for c in calls if "SELECT" in str(c[0][0]) and "delta_scan" in str(c[0][0])
        ]
        assert len(select_calls) == 1
        assert "delta_scan(" in select_calls[0]
        assert "__t__" in select_calls[0]
        assert "LIMIT 5" in select_calls[0]

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_valid_query_with_where_and_order_by(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = True
        mock_df.to_dict.return_value = []
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn

        read_delta(
            "file:///t",
            "local",
            query="SELECT * FROM __t__ WHERE x = 1 ORDER BY y LIMIT 10",
        )
        calls = [c[0][0] for c in mock_conn.execute.call_args_list if "delta_scan" in str(c[0][0])]
        assert len(calls) == 1
        assert "WHERE" in calls[0]
        assert "ORDER BY" in calls[0]

    def test_invalid_query_no_from_t_raises(self):
        with patch("src.services.duckdb_delta.duckdb.connect") as mock_connect:
            mock_conn = MagicMock()
            mock_connect.return_value = mock_conn
            with pytest.raises(DeltaError) as exc_info:
                read_delta("file:///t", "local", query="SELECT * FROM other_table LIMIT 5")
        assert "FROM __t__" in str(exc_info.value)

    def test_invalid_query_pattern_rejected(self):
        with patch("src.services.duckdb_delta.duckdb.connect") as mock_connect:
            mock_conn = MagicMock()
            mock_connect.return_value = mock_conn
            with pytest.raises(DeltaError) as exc_info:
                read_delta("file:///t", "local", query="DELETE FROM __t__")
        assert "SELECT" in str(exc_info.value) or "query" in str(exc_info.value).lower()


class TestReadDeltaLimit:
    """read_delta applies default or given limit in SQL."""

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_default_limit_10000_when_limit_none(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = True
        mock_df.to_dict.return_value = []
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn

        read_delta("file:///t", "local", limit=None)
        select_calls = [
            c[0][0] for c in mock_conn.execute.call_args_list if "LIMIT" in str(c[0][0])
        ]
        assert any("LIMIT 10000" in sql for sql in select_calls)

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_custom_limit_in_sql(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = True
        mock_df.to_dict.return_value = []
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn

        read_delta("file:///t", "local", limit=7)
        select_calls = [
            c[0][0] for c in mock_conn.execute.call_args_list if "LIMIT" in str(c[0][0])
        ]
        assert any("LIMIT 7" in sql for sql in select_calls)


class TestReadDeltaExtensionFailure:
    """When Delta extension fails to load, read_delta raises DeltaError."""

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_install_delta_failure_raises_delta_error(self, mock_connect):
        mock_conn = MagicMock()
        mock_conn.execute.side_effect = RuntimeError("install failed")
        mock_connect.return_value = mock_conn

        with pytest.raises(DeltaError) as exc_info:
            read_delta("file:///t", "local")
        assert "Delta extension" in str(exc_info.value) or "delta" in str(exc_info.value).lower()
        assert "install failed" in str(exc_info.value) or "Failed" in str(exc_info.value)


class TestReadDeltaDuckDBFailure:
    """When DuckDB execute/fetchdf fails, read_delta raises DeltaError."""

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_execute_fetchdf_raises_delta_error(self, mock_connect):
        mock_conn = MagicMock()
        mock_conn.execute.return_value.fetchdf.side_effect = OSError("table not found")
        mock_connect.return_value = mock_conn

        with pytest.raises(DeltaError) as exc_info:
            read_delta("file:///t", "local")
        assert "table not found" in str(exc_info.value) or "failed" in str(exc_info.value).lower()


class TestReadDeltaCloudSecretSetup:
    """Cloud vendors trigger corresponding secret setup."""

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_aws_invokes_create_secret_aws(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = True
        mock_df.to_dict.return_value = []
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn
        read_delta(
            "s3://b/k",
            "aws",
            credentials={
                "access_key_id": "aki",
                "secret_access_key": "sak",
                "region": "us-west-2",
                "session_token": "st",
            },
        )
        create_secret_calls = [
            c[0][0]
            for c in mock_conn.execute.call_args_list
            if "CREATE OR REPLACE SECRET" in str(c[0][0]) and "delta_aws" in str(c[0][0])
        ]
        assert len(create_secret_calls) == 1

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_alicloud_invokes_create_secret_alicloud(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = True
        mock_df.to_dict.return_value = []
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn
        read_delta(
            "s3://b/k",
            "alicloud",
            credentials={
                "access_key_id": "aki",
                "secret_access_key": "sak",
                "region": "oss-cn-hangzhou",
                "session_token": "st",
            },
        )
        create_secret_calls = [
            c[0][0]
            for c in mock_conn.execute.call_args_list
            if "CREATE OR REPLACE SECRET" in str(c[0][0])
            and "delta_alicloud" in str(c[0][0])
        ]
        assert len(create_secret_calls) == 1
        assert "SESSION_TOKEN 'st'" in str(create_secret_calls[0])
        assert "ENDPOINT 'oss-cn-hangzhou.aliyuncs.com'" in str(create_secret_calls[0])

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_alicloud_missing_credentials_raise_error(self, mock_connect):
        mock_conn = MagicMock()
        mock_connect.return_value = mock_conn
        with pytest.raises(DeltaError) as exc_info:
            read_delta(
                "s3://b/k",
                "alicloud",
                credentials={
                    "access_key_id": "aki",
                    "region": "oss-cn-hangzhou",
                },
            )
        assert "Alicloud credentials are required" in str(exc_info.value)

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_local_does_not_create_secret(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = True
        mock_df.to_dict.return_value = []
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn
        read_delta("file:///t", "local")
        secret_calls = [
            c[0][0]
            for c in mock_conn.execute.call_args_list
            if "CREATE OR REPLACE SECRET" in str(c[0][0])
        ]
        assert len(secret_calls) == 0


class TestReadDeltaPathEscaping:
    """Paths with single quotes are escaped in SQL."""

    @patch("src.services.duckdb_delta.duckdb.connect")
    def test_path_with_quote_escaped_in_sql(self, mock_connect):
        mock_conn = MagicMock()
        mock_df = MagicMock()
        mock_df.empty = True
        mock_df.to_dict.return_value = []
        mock_conn.execute.return_value.fetchdf.return_value = mock_df
        mock_connect.return_value = mock_conn

        read_delta("file:///path/with'quote", "local")
        select_calls = [
            c[0][0] for c in mock_conn.execute.call_args_list if "delta_scan" in str(c[0][0])
        ]
        assert len(select_calls) == 1
        assert "''" in select_calls[0]
        assert "with''quote" in select_calls[0]