from collections.abc import Iterator
import gzip
from io import (
    BytesIO,
    StringIO,
)
from pathlib import Path

import numpy as np
import pytest

from pandas.errors import Pandas4Warning

import pandas as pd
import pandas._testing as tm

from pandas.io.json._json import JsonReader


@pytest.fixture
def lines_json_df():
    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
    return df.to_json(lines=True, orient="records")


@pytest.fixture(params=["ujson", "pyarrow"])
def engine(request):
    if request.param == "pyarrow":
        pytest.importorskip("pyarrow.json")
    return request.param


def test_read_jsonl():
    # GH9180
    result = pd.read_json(StringIO('{"a": 1, "b": 2}\n{"b":2, "a" :1}\n'), lines=True)
    expected = pd.DataFrame([[1, 2], [1, 2]], columns=["a", "b"])
    tm.assert_frame_equal(result, expected)


def test_read_jsonl_engine_pyarrow(datapath, engine):
    result = pd.read_json(
        datapath("io", "json", "data", "line_delimited.json"),
        lines=True,
        engine=engine,
    )
    expected = pd.DataFrame({"a": [1, 3, 5], "b": [2, 4, 6]})
    tm.assert_frame_equal(result, expected)


def test_read_datetime(request, engine):
    # GH33787
    if engine == "pyarrow":
        # GH 48893
        reason = "Pyarrow only supports a file path as an input and line delimited json"
        request.applymarker(pytest.mark.xfail(reason=reason, raises=ValueError))

    df = pd.DataFrame(
        [([1, 2], ["2020-03-05", "2020-04-08T09:58:49+00:00"], "hector")],
        columns=["accounts", "date", "name"],
    )
    json_line = df.to_json(lines=True, orient="records")

    if engine == "pyarrow":
        result = pd.read_json(StringIO(json_line), engine=engine)
    else:
        result = pd.read_json(StringIO(json_line), engine=engine)
    expected = pd.DataFrame(
        [[1, "2020-03-05", "hector"], [2, "2020-04-08T09:58:49+00:00", "hector"]],
        columns=["accounts", "date", "name"],
    )
    tm.assert_frame_equal(result, expected)


def test_read_jsonl_unicode_chars():
    # GH15132: non-ascii unicode characters
    # \u201d == RIGHT DOUBLE QUOTATION MARK

    # simulate file handle
    json = '{"a": "foo”", "b": "bar"}\n{"a": "foo", "b": "bar"}\n'
    json = StringIO(json)
    result = pd.read_json(json, lines=True)
    expected = pd.DataFrame([["foo\u201d", "bar"], ["foo", "bar"]], columns=["a", "b"])
    tm.assert_frame_equal(result, expected)

    # simulate string
    json = '{"a": "foo”", "b": "bar"}\n{"a": "foo", "b": "bar"}\n'
    result = pd.read_json(StringIO(json), lines=True)
    expected = pd.DataFrame([["foo\u201d", "bar"], ["foo", "bar"]], columns=["a", "b"])
    tm.assert_frame_equal(result, expected)


def test_to_jsonl():
    # GH9180
    df = pd.DataFrame([[1, 2], [1, 2]], columns=["a", "b"])
    result = df.to_json(orient="records", lines=True)
    expected = '{"a":1,"b":2}\n{"a":1,"b":2}\n'
    assert result == expected

    df = pd.DataFrame([["foo}", "bar"], ['foo"', "bar"]], columns=["a", "b"])
    result = df.to_json(orient="records", lines=True)
    expected = '{"a":"foo}","b":"bar"}\n{"a":"foo\\"","b":"bar"}\n'
    assert result == expected
    tm.assert_frame_equal(pd.read_json(StringIO(result), lines=True), df)

    # GH15096: escaped characters in columns and data
    df = pd.DataFrame([["foo\\", "bar"], ['foo"', "bar"]], columns=["a\\", "b"])
    result = df.to_json(orient="records", lines=True)
    expected = '{"a\\\\":"foo\\\\","b":"bar"}\n{"a\\\\":"foo\\"","b":"bar"}\n'
    assert result == expected
    tm.assert_frame_equal(pd.read_json(StringIO(result), lines=True), df)


def test_to_jsonl_count_new_lines():
    # GH36888
    df = pd.DataFrame([[1, 2], [1, 2]], columns=["a", "b"])
    actual_new_lines_count = df.to_json(orient="records", lines=True).count("\n")
    expected_new_lines_count = 2
    assert actual_new_lines_count == expected_new_lines_count


@pytest.mark.parametrize("chunksize", [1, 1.0])
@pytest.mark.parametrize("buffer", [BytesIO, StringIO])
def test_readjson_chunks(request, lines_json_df, chunksize, buffer, engine):
    # Basic test that read_json(chunks=True) gives the same result as
    # read_json(chunks=False)
    # GH17048: memory usage when lines=True
    # GH#28906: read binary json lines in chunks

    if buffer == BytesIO:
        lines_json_df = lines_json_df.encode()

    if engine == "pyarrow":
        # GH 48893
        reason = (
            "Pyarrow only supports a file path as an input and line delimited json"
            "and doesn't support chunksize parameter."
        )
        request.applymarker(pytest.mark.xfail(reason=reason, raises=ValueError))

    unchunked = pd.read_json(buffer(lines_json_df), lines=True)
    with (
        buffer(lines_json_df) as buf,
        pd.read_json(buf, lines=True, chunksize=chunksize, engine=engine) as reader,
    ):
        chunked = pd.concat(reader)

    tm.assert_frame_equal(chunked, unchunked)


def test_readjson_chunksize_requires_lines(lines_json_df, engine):
    msg = "chunksize can only be passed if lines=True"
    with pytest.raises(ValueError, match=msg):
        with pd.read_json(
            StringIO(lines_json_df), lines=False, chunksize=2, engine=engine
        ) as _:
            pass


def test_readjson_chunks_series(request, engine):
    if engine == "pyarrow":
        # GH 48893
        reason = (
            "Pyarrow only supports a file path as an input and line delimited json"
            "and doesn't support chunksize parameter."
        )
        request.applymarker(pytest.mark.xfail(reason=reason))

    # Test reading line-format JSON to Series with chunksize param
    s = pd.Series({"A": 1, "B": 2})

    strio = StringIO(s.to_json(lines=True, orient="records"))
    unchunked = pd.read_json(strio, lines=True, typ="series", engine=engine)

    strio = StringIO(s.to_json(lines=True, orient="records"))
    with pd.read_json(
        strio, lines=True, typ="series", chunksize=1, engine=engine
    ) as reader:
        chunked = pd.concat(reader)

    tm.assert_series_equal(chunked, unchunked)


def test_readjson_each_chunk(request, lines_json_df, engine):
    if engine == "pyarrow":
        # GH 48893
        reason = (
            "Pyarrow only supports a file path as an input and line delimited json"
            "and doesn't support chunksize parameter."
        )
        request.applymarker(pytest.mark.xfail(reason=reason, raises=ValueError))

    # Other tests check that the final result of read_json(chunksize=True)
    # is correct. This checks the intermediate chunks.
    with pd.read_json(
        StringIO(lines_json_df), lines=True, chunksize=2, engine=engine
    ) as reader:
        chunks = list(reader)
    assert chunks[0].shape == (2, 2)
    assert chunks[1].shape == (1, 2)


def test_readjson_chunks_from_file(request, engine, temp_file):
    if engine == "pyarrow":
        # GH 48893
        reason = (
            "Pyarrow only supports a file path as an input and line delimited json"
            "and doesn't support chunksize parameter."
        )
        request.applymarker(pytest.mark.xfail(reason=reason, raises=ValueError))

    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
    df.to_json(temp_file, lines=True, orient="records")
    with pd.read_json(temp_file, lines=True, chunksize=1, engine=engine) as reader:
        chunked = pd.concat(reader)
    unchunked = pd.read_json(temp_file, lines=True, engine=engine)
    tm.assert_frame_equal(unchunked, chunked)


def test_readjson_chunks_from_binary_stream():
    # GH47659 - reading a binary stream (e.g. from S3) in chunks used to raise
    # "TypeError: initial_value must be str or None, not bytes"
    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
    data = df.to_json(lines=True, orient="records").encode("utf-8")
    with pd.read_json(BytesIO(data), lines=True, chunksize=1) as reader:
        chunked = pd.concat(reader)
    tm.assert_frame_equal(chunked, df)


@pytest.mark.parametrize("chunksize", [None, 1])
def test_readjson_chunks_closes(chunksize, temp_file):
    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
    df.to_json(temp_file, lines=True, orient="records")
    reader = JsonReader(
        temp_file,
        orient=None,
        typ="frame",
        dtype=True,
        convert_axes=True,
        convert_dates=True,
        keep_default_dates=True,
        precise_float=False,
        date_unit=None,
        encoding=None,
        lines=True,
        chunksize=chunksize,
        compression=None,
        nrows=None,
    )
    with reader:
        reader.read()
    assert reader.handles.handle.closed, (
        f"didn't close stream with chunksize = {chunksize}"
    )


@pytest.mark.parametrize("chunksize", [0, -1, 2.2, "foo"])
def test_readjson_invalid_chunksize(lines_json_df, chunksize, engine):
    msg = r"'chunksize' must be an integer >=1"

    with pytest.raises(ValueError, match=msg):
        with pd.read_json(
            StringIO(lines_json_df), lines=True, chunksize=chunksize, engine=engine
        ) as _:
            pass


@pytest.mark.parametrize("chunksize", [None, 1, 2])
def test_readjson_chunks_multiple_empty_lines(chunksize):
    j = """

    {"A":1,"B":4}



    {"A":2,"B":5}







    {"A":3,"B":6}
    """
    orig = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})
    test = pd.read_json(StringIO(j), lines=True, chunksize=chunksize)
    if chunksize is not None:
        with test:
            test = pd.concat(test)
    tm.assert_frame_equal(orig, test, obj=f"chunksize: {chunksize}")


def test_readjson_unicode(request, monkeypatch, engine, temp_file):
    if engine == "pyarrow":
        # GH 48893
        reason = (
            "Pyarrow only supports a file path as an input and line delimited json"
            "and doesn't support chunksize parameter."
        )
        request.applymarker(pytest.mark.xfail(reason=reason, raises=ValueError))

    monkeypatch.setattr("locale.getpreferredencoding", lambda do_setlocale: "cp949")
    with open(temp_file, "w", encoding="utf-8") as f:
        f.write('{"£©µÀÆÖÞßéöÿ":["АБВГДабвгд가"]}')

    result = pd.read_json(temp_file, engine=engine)
    expected = pd.DataFrame({"£©µÀÆÖÞßéöÿ": ["АБВГДабвгд가"]})
    tm.assert_frame_equal(result, expected)


@pytest.mark.parametrize("nrows", [1, 2])
def test_readjson_nrows(nrows, engine):
    # GH 33916
    # Test reading line-format JSON to Series with nrows param
    jsonl = """{"a": 1, "b": 2}
        {"a": 3, "b": 4}
        {"a": 5, "b": 6}
        {"a": 7, "b": 8}"""
    result = pd.read_json(StringIO(jsonl), lines=True, nrows=nrows)
    expected = pd.DataFrame({"a": [1, 3, 5, 7], "b": [2, 4, 6, 8]}).iloc[:nrows]
    tm.assert_frame_equal(result, expected)

    jsonl = """{"a": 1, "b": 2}
        {"a": 3, "b": 4}"""
    result = pd.read_json(StringIO(jsonl), lines=True, nrows=0)
    expected = pd.DataFrame()
    tm.assert_frame_equal(result, expected)


def test_readjson_nrows_not_implemented_pyarrow(temp_file):
    # GH 64025
    pytest.importorskip("pyarrow.json")

    jsonl = """{"a": 1, "b": 2}
        {"a": 3, "b": 4}"""
    Path(temp_file).write_text(jsonl, encoding="utf-8")

    msg = "currently pyarrow engine doesn't support nrows parameter"
    with pytest.raises(NotImplementedError, match=msg):
        pd.read_json(temp_file, lines=True, nrows=1, engine="pyarrow")


@pytest.mark.parametrize("nrows,chunksize", [(2, 2), (3, 2), (4, 2)])
def test_readjson_nrows_chunks(request, nrows, chunksize, engine):
    # GH 33916
    # Test reading line-format JSON to Series with nrows and chunksize param
    if engine == "pyarrow":
        # GH 48893
        reason = (
            "Pyarrow only supports a file path as an input and line delimited json"
            "and doesn't support chunksize parameter."
        )
        request.applymarker(pytest.mark.xfail(reason=reason, raises=ValueError))

    jsonl = """{"a": 1, "b": 2}
        {"a": 3, "b": 4}
        {"a": 5, "b": 6}
        {"a": 7, "b": 8}"""

    if engine != "pyarrow":
        with pd.read_json(
            StringIO(jsonl), lines=True, nrows=nrows, chunksize=chunksize, engine=engine
        ) as reader:
            chunked = pd.concat(reader)
    else:
        with pd.read_json(
            jsonl, lines=True, nrows=nrows, chunksize=chunksize, engine=engine
        ) as reader:
            chunked = pd.concat(reader)
    expected = pd.DataFrame({"a": [1, 3, 5, 7], "b": [2, 4, 6, 8]}).iloc[:nrows]
    tm.assert_frame_equal(chunked, expected)


@pytest.mark.parametrize(
    "jsonl,nrows",
    [('{"a":1}\n{"a":2}\n', 0), ("", None)],
)
def test_readjson_read_empty_chunks(jsonl, nrows):
    # GH 64025
    kwargs = {"lines": True, "chunksize": 2}
    if nrows is not None:
        kwargs["nrows"] = nrows

    with pd.read_json(StringIO(jsonl), **kwargs) as reader:
        result = reader.read()

    tm.assert_frame_equal(result, pd.DataFrame())


def test_readjson_nrows_requires_lines(engine):
    # GH 33916
    # Test ValueError raised if nrows is set without setting lines in read_json
    jsonl = """{"a": 1, "b": 2}
        {"a": 3, "b": 4}
        {"a": 5, "b": 6}
        {"a": 7, "b": 8}"""
    msg = "nrows can only be passed if lines=True"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(jsonl, lines=False, nrows=2, engine=engine)


def test_readjson_lines_chunks_fileurl(request, datapath, engine):
    # GH 27135
    # Test reading line-format JSON from file url
    if engine == "pyarrow":
        # GH 48893
        reason = (
            "Pyarrow only supports a file path as an input and line delimited json"
            "and doesn't support chunksize parameter."
        )
        request.applymarker(pytest.mark.xfail(reason=reason, raises=ValueError))

    df_list_expected = [
        pd.DataFrame([[1, 2]], columns=["a", "b"], index=[0]),
        pd.DataFrame([[3, 4]], columns=["a", "b"], index=[1]),
        pd.DataFrame([[5, 6]], columns=["a", "b"], index=[2]),
    ]
    os_path = datapath("io", "json", "data", "line_delimited.json")
    file_url = Path(os_path).as_uri()
    with pd.read_json(file_url, lines=True, chunksize=1, engine=engine) as url_reader:
        for index, chuck in enumerate(url_reader):
            tm.assert_frame_equal(chuck, df_list_expected[index])


def test_chunksize_is_incremental():
    # See https://github.com/pandas-dev/pandas/issues/34548
    jsonl = (
        """{"a": 1, "b": 2}
        {"a": 3, "b": 4}
        {"a": 5, "b": 6}
        {"a": 7, "b": 8}\n"""
        * 1000
    )

    class MyReader:
        def __init__(self, contents) -> None:
            self.read_count = 0
            self.stringio = StringIO(contents)

        def read(self, *args):
            self.read_count += 1
            return self.stringio.read(*args)

        def __iter__(self) -> Iterator:
            self.read_count += 1
            return iter(self.stringio)

    reader = MyReader(jsonl)
    assert len(list(pd.read_json(reader, lines=True, chunksize=100))) > 1
    assert reader.read_count > 10


@pytest.mark.parametrize("orient_", ["split", "index", "table"])
def test_to_json_append_orient(orient_):
    # GH 35849
    # Test ValueError when orient is not 'records'
    df = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
    msg = (
        r"mode='a' \(append\) is only supported when "
        "lines is True and orient is 'records'"
    )
    with pytest.raises(ValueError, match=msg):
        df.to_json(mode="a", orient=orient_)


def test_to_json_append_lines():
    # GH 35849
    # Test ValueError when lines is not True
    df = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
    msg = (
        r"mode='a' \(append\) is only supported when "
        "lines is True and orient is 'records'"
    )
    with pytest.raises(ValueError, match=msg):
        df.to_json(mode="a", lines=False, orient="records")


@pytest.mark.parametrize("mode_", ["r", "x"])
def test_to_json_append_mode(mode_):
    # GH 35849
    # Test ValueError when mode is not supported option
    df = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
    msg = (
        f"mode={mode_} is not a valid option.Only 'w' and 'a' are currently supported."
    )
    with pytest.raises(ValueError, match=msg):
        df.to_json(mode=mode_, lines=False, orient="records")


def test_to_json_append_output_consistent_columns(temp_file):
    # GH 35849
    # Testing that resulting output reads in as expected.
    # Testing same columns, new rows
    df1 = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
    df2 = pd.DataFrame({"col1": [3, 4], "col2": ["c", "d"]})

    expected = pd.DataFrame({"col1": [1, 2, 3, 4], "col2": ["a", "b", "c", "d"]})
    # Save dataframes to the same file
    df1.to_json(temp_file, lines=True, orient="records")
    df2.to_json(temp_file, mode="a", lines=True, orient="records")

    # Read path file
    result = pd.read_json(temp_file, lines=True)
    tm.assert_frame_equal(result, expected)


def test_to_json_append_output_inconsistent_columns(temp_file):
    # GH 35849
    # Testing that resulting output reads in as expected.
    # Testing one new column, one old column, new rows
    df1 = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
    df3 = pd.DataFrame({"col2": ["e", "f"], "col3": ["!", "#"]})

    expected = pd.DataFrame(
        {
            "col1": [1, 2, None, None],
            "col2": ["a", "b", "e", "f"],
            "col3": [np.nan, np.nan, "!", "#"],
        }
    )
    # Save dataframes to the same file
    df1.to_json(temp_file, mode="a", lines=True, orient="records")
    df3.to_json(temp_file, mode="a", lines=True, orient="records")

    # Read path file
    result = pd.read_json(temp_file, lines=True)
    tm.assert_frame_equal(result, expected)


def test_to_json_append_output_different_columns(temp_file):
    # GH 35849
    # Testing that resulting output reads in as expected.
    # Testing same, differing and new columns
    df1 = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
    df2 = pd.DataFrame({"col1": [3, 4], "col2": ["c", "d"]})
    df3 = pd.DataFrame({"col2": ["e", "f"], "col3": ["!", "#"]})
    df4 = pd.DataFrame({"col4": [True, False]})

    expected = pd.DataFrame(
        {
            "col1": [1, 2, 3, 4, None, None, None, None],
            "col2": ["a", "b", "c", "d", "e", "f", np.nan, np.nan],
            "col3": [np.nan, np.nan, np.nan, np.nan, "!", "#", np.nan, np.nan],
            "col4": [None, None, None, None, None, None, True, False],
        }
    ).astype({"col4": "float"})
    # Save dataframes to the same file
    df1.to_json(temp_file, mode="a", lines=True, orient="records")
    df2.to_json(temp_file, mode="a", lines=True, orient="records")
    df3.to_json(temp_file, mode="a", lines=True, orient="records")
    df4.to_json(temp_file, mode="a", lines=True, orient="records")

    # Read path file
    result = pd.read_json(temp_file, lines=True)
    tm.assert_frame_equal(result, expected)


def test_to_json_append_output_different_columns_reordered(temp_file):
    # GH 35849
    # Testing that resulting output reads in as expected.
    # Testing specific result column order.
    df1 = pd.DataFrame({"col1": [1, 2], "col2": ["a", "b"]})
    df2 = pd.DataFrame({"col1": [3, 4], "col2": ["c", "d"]})
    df3 = pd.DataFrame({"col2": ["e", "f"], "col3": ["!", "#"]})
    df4 = pd.DataFrame({"col4": [True, False]})

    # df4, df3, df2, df1 (in that order)
    expected = pd.DataFrame(
        {
            "col4": [True, False, None, None, None, None, None, None],
            "col2": [np.nan, np.nan, "e", "f", "c", "d", "a", "b"],
            "col3": [np.nan, np.nan, "!", "#", np.nan, np.nan, np.nan, np.nan],
            "col1": [None, None, None, None, 3, 4, 1, 2],
        }
    ).astype({"col4": "float"})
    # Save dataframes to the same file
    df4.to_json(temp_file, mode="a", lines=True, orient="records")
    df3.to_json(temp_file, mode="a", lines=True, orient="records")
    df2.to_json(temp_file, mode="a", lines=True, orient="records")
    df1.to_json(temp_file, mode="a", lines=True, orient="records")

    # Read path file
    result = pd.read_json(temp_file, lines=True)
    tm.assert_frame_equal(result, expected)


# ----------------------------------------------------------------------------
# The pyarrow engine used to silently ignore many read_json options, producing
# a wrong result instead of an error. They now raise ValueError instead.


@pytest.fixture
def pyarrow_jsonl(temp_file):
    pytest.importorskip("pyarrow.json")
    Path(temp_file).write_text(
        '{"a": 1, "b": "x"}\n{"a": 2, "b": "y"}\n', encoding="utf-8"
    )
    return temp_file


def test_pyarrow_engine_typ_series(pyarrow_jsonl):
    msg = "The 'typ' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(pyarrow_jsonl, lines=True, engine="pyarrow", typ="series")


@pytest.mark.parametrize("convert_dates", [False, ["a"]])
def test_pyarrow_engine_convert_dates(pyarrow_jsonl, convert_dates):
    msg = "The 'convert_dates' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(
            pyarrow_jsonl, lines=True, engine="pyarrow", convert_dates=convert_dates
        )


def test_pyarrow_engine_keep_default_dates(pyarrow_jsonl):
    msg = "The 'keep_default_dates' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(
            pyarrow_jsonl, lines=True, engine="pyarrow", keep_default_dates=False
        )


def test_pyarrow_engine_date_unit(pyarrow_jsonl):
    msg = "The 'date_unit' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(pyarrow_jsonl, lines=True, engine="pyarrow", date_unit="ms")


def test_pyarrow_engine_precise_float(pyarrow_jsonl):
    msg = "The 'precise_float' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(pyarrow_jsonl, lines=True, engine="pyarrow", precise_float=True)


@pytest.mark.parametrize("convert_axes", [True, False])
def test_pyarrow_engine_convert_axes(pyarrow_jsonl, convert_axes):
    # convert_axes defaults to None (resolved to True); any explicit value is
    # rejected, since the pyarrow engine cannot honor it.
    msg = "The 'convert_axes' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(
            pyarrow_jsonl, lines=True, engine="pyarrow", convert_axes=convert_axes
        )


@pytest.mark.parametrize("dtype", [True, False, "float64"])
def test_pyarrow_engine_scalar_dtype(pyarrow_jsonl, dtype):
    # dtype defaults to None (resolved to True); a scalar dtype applies to
    # every column but is silently ignored by the pyarrow engine.
    msg = "The 'dtype' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(pyarrow_jsonl, lines=True, engine="pyarrow", dtype=dtype)


def test_pyarrow_engine_non_arrow_dtype_mapping(pyarrow_jsonl):
    # GH#59516 only handled ArrowDtype entries; a non-ArrowDtype dtype in the
    # mapping was silently dropped.
    msg = "only supports ArrowDtype values in the 'dtype' mapping"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(
            pyarrow_jsonl,
            lines=True,
            engine="pyarrow",
            dtype={"a": "int64[pyarrow]", "b": "float64"},
        )


def test_pyarrow_engine_encoding(pyarrow_jsonl):
    # the pyarrow engine assumes UTF-8 and ignores the encoding argument.
    msg = "The 'encoding' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(pyarrow_jsonl, lines=True, engine="pyarrow", encoding="latin-1")


def test_pyarrow_engine_encoding_errors(pyarrow_jsonl):
    msg = "The 'encoding_errors' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(
            pyarrow_jsonl, lines=True, engine="pyarrow", encoding_errors="ignore"
        )


def test_pyarrow_engine_storage_options(pyarrow_jsonl):
    msg = "The 'storage_options' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(
            pyarrow_jsonl,
            lines=True,
            engine="pyarrow",
            storage_options={"anon": True},
        )


@pytest.mark.parametrize("compression", ["gzip", {"method": "gzip"}])
def test_pyarrow_engine_compression(pyarrow_jsonl, compression):
    # pyarrow infers compression from the file extension itself; pandas cannot
    # honor an explicitly-passed method (the default "infer" is allowed).
    msg = "The 'compression' option is not supported with the 'pyarrow' engine"
    with pytest.raises(ValueError, match=msg):
        pd.read_json(
            pyarrow_jsonl, lines=True, engine="pyarrow", compression=compression
        )


@pytest.mark.parametrize("encoding", [None, "utf-8", "utf8", "UTF-8"])
def test_pyarrow_engine_allows_default_options(pyarrow_jsonl, encoding):
    # Passing the supported defaults explicitly must not raise (though the
    # deprecated date kwargs warn; GH#59161).
    depr_msg = "keyword in read_json is deprecated"
    # check_stacklevel=False: older pyarrow emits its own Pandas4Warning from
    #  make_block during to_pandas, which the stacklevel check would trip on.
    with tm.assert_produces_warning(
        Pandas4Warning, match=depr_msg, check_stacklevel=False
    ):
        result = pd.read_json(
            pyarrow_jsonl,
            lines=True,
            engine="pyarrow",
            typ="frame",
            convert_dates=True,
            keep_default_dates=True,
            precise_float=False,
            date_unit=None,
            encoding=encoding,
            encoding_errors="strict",
            compression="infer",
        )
    expected = pd.DataFrame({"a": [1, 2], "b": ["x", "y"]})
    tm.assert_frame_equal(result, expected, check_dtype=False)


def test_pyarrow_engine_reads_gzip_with_default_compression(tmp_path):
    # the default compression="infer" is allowed; pyarrow decompresses a
    # ".gz" file by extension on its own.
    pytest.importorskip("pyarrow.json")
    path = tmp_path / "data.json.gz"
    with gzip.open(path, "wb") as fh:
        fh.write(b'{"a": 1}\n{"a": 2}\n')
    result = pd.read_json(str(path), lines=True, engine="pyarrow")
    expected = pd.DataFrame({"a": [1, 2]})
    tm.assert_frame_equal(result, expected, check_dtype=False)


def test_ujson_engine_allows_pyarrow_rejected_options(temp_file):
    # The new validation is gated on engine="pyarrow"; the default ujson
    # engine must keep accepting these options (though the deprecated date
    # kwargs warn; GH#59161).
    Path(temp_file).write_text('{"a": 1}\n{"a": 2}\n', encoding="utf-8")
    depr_msg = "keyword in read_json is deprecated"
    with tm.assert_produces_warning(Pandas4Warning, match=depr_msg):
        result = pd.read_json(
            temp_file,
            lines=True,
            engine="ujson",
            convert_dates=False,
            keep_default_dates=False,
            convert_axes=False,
            precise_float=True,
            date_unit="ms",
            encoding="utf-8",
            encoding_errors="strict",
        )
    expected = pd.DataFrame({"a": [1, 2]})
    tm.assert_frame_equal(result, expected)