from __future__ import annotations

from collections.abc import Iterator, Mapping
from typing import Any

from config import (
    OPERATION_MODE_BATCHED_STREAM,
    OPERATION_MODE_STANDARD,
    S3_CSV_READ_MODE_BUFFERED,
    S3_CSV_READ_MODE_STREAMING,
    ConnectorConfig,
)
from errors import ConfigurationError
from storage.object_store import S3ObjectStore
from storage.resolver import StorageURI, redact_storage_uri
from tests.pytest_compat import pytest


class _SdkConfigurationMap(Mapping[str, str]):
    """Mimics the protobuf ScalarMap passed by the Fivetran SDK gRPC path."""

    def __init__(self, values: dict[str, str]) -> None:
        self._values = values

    def __getitem__(self, key: str) -> str:
        return self._values[key]

    def __iter__(self) -> Iterator[str]:
        return iter(self._values)

    def __len__(self) -> int:
        return len(self._values)


def test_storage_uri_file_scheme():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )
    assert config.storage_uri == "file:///tmp/workspace"


def test_accepts_fivetran_sdk_configuration_map():
    config = ConnectorConfig.from_dict(
        _SdkConfigurationMap(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
            }
        )
    )

    assert config.sync_task_id == "task-001"
    assert config.storage_uri == "file:///tmp/workspace"


def test_storage_uri_s3_scheme_with_params():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "s3://my-bucket/prefix?region=us-east-1&endpoint=http://minio:9000",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )
    assert config.storage_uri == "s3://my-bucket/prefix?region=us-east-1&endpoint=http://minio:9000"


def test_control_plane_s3_url_and_nested_blocks_are_supported():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "s3://my-bucket/task1/",
            "s3_region": "us-west-2",
            "s3_ak": "AKIA_TEST",
            "s3_sk": "secret+with&reserved=",
            "snapshot_prefix": "snapshot/",
            "incremental_prefix": "increment/",
            "snapshot": {
                "csv_no_header": False,
                "csv_separator": ",",
                "csv_delimiter": '"',
                "csv_null_value": "NULL",
                "csv_line_terminator": "\r\n",
                "csv_binary_output_dialect": "snowflake",
                "escape_backslash": True,
            },
            "incremental": {
                "csv_delimiter": ",",
                "csv_quote": '"',
                "csv_null_value": "\\N",
                "csv_line_terminator": "\r\n",
                "csv_binary_encoding": "hex",
                "csv_output_old_value": False,
            },
        }
    )

    uri = StorageURI.parse(config.storage_uri)

    assert config.sync_task_id == "task-001"
    assert config.snapshot_prefix == "snapshot"
    assert config.incremental_prefix == "increment"
    assert config.snapshot.csv_null_value == "NULL"
    assert config.snapshot.escape_backslash is True
    assert config.incremental.csv_null_value == "\\N"
    assert config.snapshot.csv_binary_output_dialect == "snowflake"
    assert config.snapshot.binary_encoding == "hex"
    assert config.incremental.csv_binary_encoding == "hex"
    assert uri.bucket == "my-bucket"
    assert uri.prefix == "task1"
    assert uri.region == "us-west-2"
    assert uri.access_key == "AKIA_TEST"
    assert uri.secret_access_key == "secret+with&reserved="


def test_control_plane_csv_blocks_accept_json_strings():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "s3://my-bucket/task1/",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "snapshot": '{"csv_separator":",","csv_delimiter":"\\"","csv_null_value":"NULL","csv_no_header":false,"csv_binary_output_dialect":"hex"}',
            "incremental": '{"csv_delimiter":",","csv_quote":"\\"","csv_null_value":"\\\\N","csv_binary_encoding":"base64","csv_output_old_value":true}',
        }
    )

    assert config.snapshot.csv_null_value == "NULL"
    assert config.incremental.csv_null_value == "\\N"
    assert config.snapshot.binary_encoding == "hex"
    assert config.incremental.csv_binary_encoding == "base64"
    assert config.incremental.output_old_value is True


def test_config_blocks_accept_legacy_boolean_strings():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot": "true",
            "snapshot_prefix": "snapshot",
            "incremental": "true",
            "incremental_prefix": "increment",
        }
    )

    assert config.snapshot.csv_null_value == "\\N"
    assert config.snapshot.csv_binary_output_dialect == "default"
    assert config.snapshot.binary_encoding == "utf8"
    assert config.snapshot.escape_backslash is False
    assert config.snapshot.s3_csv_read_mode == S3_CSV_READ_MODE_STREAMING
    assert config.incremental.csv_binary_encoding == "base64"
    assert config.incremental.output_old_value is False


def test_snapshot_escape_backslash_accepts_only_true_false() -> None:
    true_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "snapshot": {"escape_backslash": "true"},
        }
    )
    false_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "snapshot": {"escape_backslash": "false"},
        }
    )

    assert true_config.snapshot.escape_backslash is True
    assert false_config.snapshot.escape_backslash is False

    with pytest.raises(ConfigurationError, match=r"snapshot\.escape_backslash"):
        ConnectorConfig.from_dict(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
                "snapshot": {"escape_backslash": "yes"},
            }
        )


def test_snapshot_csv_parser_is_not_supported() -> None:
    with pytest.raises(ConfigurationError, match=r"snapshot\.csv_parser"):
        ConnectorConfig.from_dict(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
                "snapshot": {"csv_parser": "csv"},
            }
        )


def test_snapshot_s3_csv_read_mode_accepts_streaming_and_buffered() -> None:
    streaming_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "snapshot": {"s3_csv_read_mode": "streaming"},
        }
    )
    buffered_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "snapshot": '{"s3_csv_read_mode":"buffered"}',
        }
    )

    assert streaming_config.snapshot.s3_csv_read_mode == S3_CSV_READ_MODE_STREAMING
    assert buffered_config.snapshot.s3_csv_read_mode == S3_CSV_READ_MODE_BUFFERED


def test_snapshot_s3_csv_read_mode_rejects_unknown_value() -> None:
    with pytest.raises(ConfigurationError, match=r"snapshot\.s3_csv_read_mode"):
        ConnectorConfig.from_dict(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
                "snapshot": {"s3_csv_read_mode": "memory"},
            }
        )


def test_operation_mode_defaults_to_batched_stream_and_accepts_standard() -> None:
    default_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )
    standard_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "operation_mode": "standard",
        }
    )

    assert default_config.operation_mode == OPERATION_MODE_BATCHED_STREAM
    assert standard_config.operation_mode == OPERATION_MODE_STANDARD


def test_operation_mode_rejects_unknown_value() -> None:
    with pytest.raises(ConfigurationError, match="operation_mode"):
        ConnectorConfig.from_dict(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
                "operation_mode": "sdk",
            }
        )


def test_incremental_legacy_csv_binary_encoding_method_is_supported():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "incremental": {"csv_binary_encoding_method": "hex"},
        }
    )

    assert config.incremental.csv_binary_encoding == "hex"


def test_binary_encoding_configuration_rejects_unknown_values():
    with pytest.raises(ConfigurationError, match=r"snapshot\.csv_binary_output_dialect"):
        ConnectorConfig.from_dict(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
                "snapshot": {"csv_binary_output_dialect": "bigquery"},
            }
        )

    with pytest.raises(ConfigurationError, match=r"incremental\.csv_binary_encoding"):
        ConnectorConfig.from_dict(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
                "incremental": {"csv_binary_encoding": "utf8"},
            }
        )


def test_control_plane_s3_region_is_supported():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "s3://my-bucket/task1",
            "s3_region": "us-west-2",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )

    uri = StorageURI.parse(config.storage_uri)

    assert uri.region == "us-west-2"


def test_snapshot_required_defaults_to_true_and_can_be_disabled():
    default_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )
    incremental_only_config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "snapshot_required": False,
        }
    )

    assert default_config.snapshot_required is True
    assert incremental_only_config.snapshot_required is False


def test_empty_prefixes_are_allowed_when_present_for_incremental_only():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "",
            "incremental_prefix": "",
            "snapshot_required": "false",
        }
    )

    assert config.snapshot_prefix == ""
    assert config.incremental_prefix == ""
    assert config.snapshot_required is False


def test_config_prefix_rejects_path_escape():
    with pytest.raises(ConfigurationError, match="normalized relative"):
        ConnectorConfig.from_dict(
            {
                "sync_task_id": "task-001",
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "../snapshot",
                "incremental_prefix": "increment",
            }
        )


def test_s3_url_merges_control_plane_s3_credentials():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "s3://my-bucket/task1?region=us-east-1",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "s3_ak": "AKIA_TEST",
            "s3_sk": "secret",
        }
    )

    uri = StorageURI.parse(config.storage_uri)

    assert uri.region == "us-east-1"
    assert uri.access_key == "AKIA_TEST"
    assert uri.secret_access_key == "secret"


def test_redact_storage_uri_hides_s3_credentials():
    redacted = redact_storage_uri(
        "s3://my-bucket/task1?region=us-east-1&access-key=AKIA_TEST&secret-access-key=secret&session-token=token"
    )

    assert "AKIA_TEST" not in redacted
    assert "secret-access-key=secret" not in redacted
    assert "session-token=token" not in redacted
    assert "access-key=%2A%2A%2A" in redacted
    assert "region=us-east-1" in redacted


def test_s3_object_store_passes_explicit_credentials(monkeypatch):
    import boto3

    captured: dict[str, Any] = {}

    class FakeSession:
        def __init__(self, **kwargs):
            captured["session_kwargs"] = kwargs

        def client(self, service_name, **kwargs):
            captured["service_name"] = service_name
            captured["client_kwargs"] = kwargs
            return object()

    monkeypatch.setattr(boto3, "Session", FakeSession)

    S3ObjectStore(
        bucket="my-bucket",
        region="us-east-1",
        access_key_id="AKIA_TEST",
        secret_access_key="secret",
        session_token="token",
        force_path_style=True,
    )

    assert captured["service_name"] == "s3"
    assert captured["session_kwargs"] == {
        "aws_access_key_id": "AKIA_TEST",
        "aws_secret_access_key": "secret",
        "aws_session_token": "token",
        "region_name": "us-east-1",
    }
    assert captured["client_kwargs"]["config"].s3 == {"addressing_style": "path"}


def test_storage_uri_is_required():
    with pytest.raises(ConfigurationError, match="s3_url"):
        ConnectorConfig.from_dict({})


def test_sync_task_id_is_required():
    with pytest.raises(ConfigurationError, match="sync_task_id"):
        ConnectorConfig.from_dict(
            {
                "s3_url": "file:///tmp/workspace",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
            }
        )


def test_snapshot_and_incremental_prefix_are_required():
    with pytest.raises(ConfigurationError, match="snapshot_prefix"):
        ConnectorConfig.from_dict({"sync_task_id": "task-001", "s3_url": "file:///tmp/workspace"})

    with pytest.raises(ConfigurationError, match="incremental_prefix"):
        ConnectorConfig.from_dict(
            {"sync_task_id": "task-001", "s3_url": "file:///tmp/workspace", "snapshot_prefix": "snapshot"}
        )


def test_snapshot_prefix_custom_generation_is_cleaned():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "/snapshot/generations/gen_abc/",
            "incremental_prefix": "increment",
        }
    )
    assert config.snapshot_prefix == "snapshot/generations/gen_abc"


def test_enable_diagnostic_columns():
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "enable_diagnostic_columns": True,
        }
    )

    assert config.enable_diagnostic_columns is True