from __future__ import annotations

from typing import Any

from csv_parser import iter_csv_rows
from csv_parser import new as new_csv_parser
from csv_parser.dumpling import DumplingBackslashCSVParser
from errors import SnapshotError
from storage.csv_file import open_csv_text
from storage.object_store import S3ObjectStore
from tests.pytest_compat import pytest


def _iter_dumpling_csv_rows(lines: list[str]):
    parser = DumplingBackslashCSVParser(new_csv_parser(error_type=SnapshotError, context="snapshot", strict=True))
    return parser.iter_rows(lines)


@pytest.mark.parametrize(
    ("line", "expected"),
    [
        ("", [""]),
        (r"1,\N", ["1", r"\N"]),
        (
            '1,"{\\"type\\": \\"signup\\"}","C:\\\\tmp\\\\file","line\\nbreak","carriage\\rreturn","nul\\0byte"',
            ["1", '{"type": "signup"}', r"C:\tmp\file", "line\nbreak", "carriage\rreturn", "nul\0byte"],
        ),
        ('"a,b","c""d","x\\"y"', ["a,b", 'c"d', 'x"y']),
        (r"plain,with\backslash,\N", ["plain", r"with\backslash", r"\N"]),
        ('a"b,c', ['a"b', "c"]),
        ('raw\ue000value,"quoted\ue000value"', ["raw\ue000value", "quoted\ue000value"]),
    ],
)
def test_dumpling_csv_reader_decodes_backslash_escaped_rows(line: str, expected: list[str]) -> None:
    assert next(_iter_dumpling_csv_rows([f"{line}\n"])) == expected


def test_snapshot_csv_reader_uses_standard_csv_by_default() -> None:
    row = next(
        iter_csv_rows(
            ['1,"{""type"": ""signup""}","line\\nbreak",\\N\n'],
            parser=new_csv_parser(error_type=SnapshotError, context="snapshot", strict=True),
            escape_backslash=False,
        )
    )

    assert row == ["1", '{"type": "signup"}', r"line\nbreak", r"\N"]


def test_snapshot_csv_reader_uses_adapter_when_escape_backslash_is_enabled() -> None:
    row = next(
        iter_csv_rows(
            ['1,"line\\nbreak","{\\"type\\": \\"signup\\"}"\n'],
            parser=new_csv_parser(error_type=SnapshotError, context="snapshot", strict=True),
            escape_backslash=True,
        )
    )

    assert row == ["1", "line\nbreak", '{"type": "signup"}']


def test_dumpling_csv_reader_rejects_unclosed_quoted_field() -> None:
    with pytest.raises(SnapshotError, match="unclosed quoted field"):
        list(_iter_dumpling_csv_rows(['"unterminated\n']))


def test_open_csv_text_buffers_s3_body(monkeypatch) -> None:
    import boto3

    captured: dict[str, str] = {}

    class FakeBody:
        closed = False
        read_count = 0

        def read(self) -> bytes:
            self.read_count += 1
            return b"id\n1\n"

        def close(self) -> None:
            self.closed = True

    body = FakeBody()

    class FakeClient:
        def get_object(self, **kwargs: str) -> dict[str, Any]:
            captured.update(kwargs)
            return {"Body": body}

    class FakeSession:
        def __init__(self, **kwargs: str) -> None:
            pass

        def client(self, service_name: str, **kwargs: Any) -> FakeClient:
            return FakeClient()

    monkeypatch.setattr(boto3, "Session", FakeSession)
    store = S3ObjectStore(bucket="my-bucket", prefix="task1")

    with open_csv_text(
        store,
        "snapshot/table.csv",
        buffer_s3=True,
    ) as handle:
        assert handle.read() == "id\n1\n"

    assert body.read_count == 1
    assert body.closed is True
    assert captured == {"Bucket": "my-bucket", "Key": "task1/snapshot/table.csv"}