# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.


import os
import pathlib
import tempfile

import pytest

import nanoarrow as na
from nanoarrow.ipc import InputStream


def test_array_stream_iter():
    stream = na.ArrayStream([1, 2, 3], na.int32())
    assert stream.schema.type == na.Type.INT32
    stream_iter = iter(stream)

    assert list(next(stream_iter).iter_py()) == [1, 2, 3]
    with pytest.raises(StopIteration):
        next(stream_iter)


def test_array_stream_iter_chunks():
    stream = na.ArrayStream([1, 2, 3], na.int32())
    chunks = list(stream.iter_chunks())
    assert len(chunks) == 1
    assert chunks[0].to_pylist() == [1, 2, 3]


def test_array_stream_iter_py():
    stream = na.ArrayStream([1, 2, 3], na.int32())
    assert list(stream.iter_py()) == [1, 2, 3]


def test_array_stream_iter_tuples():
    c_array = na.c_array_from_buffers(
        na.struct({"col1": na.int32(), "col2": na.string()}),
        length=3,
        buffers=[None],
        children=[
            na.c_array([1, 2, 3], na.int32()),
            na.c_array(["a", "b", "c"], na.string()),
        ],
    )

    stream = na.ArrayStream(c_array)
    assert list(stream.iter_tuples()) == [(1, "a"), (2, "b"), (3, "c")]


def test_array_stream_to_pylist():
    stream = na.ArrayStream([1, 2, 3], na.int32())
    assert stream.to_pylist() == [1, 2, 3]


def test_array_stream_to_columns():
    c_array = na.c_array_from_buffers(
        na.struct({"col1": na.int32(), "col2": na.string()}),
        length=3,
        buffers=[None],
        children=[
            na.c_array([1, 2, 3], na.int32()),
            na.c_array(["a", "b", "c"], na.string()),
        ],
    )

    stream = na.ArrayStream(c_array)
    names, columns = stream.to_columns_pysequence()
    assert names == ["col1", "col2"]
    assert list(columns[0]) == [1, 2, 3]
    assert list(columns[1]) == ["a", "b", "c"]


def test_array_stream_read_all():
    stream = na.ArrayStream([1, 2, 3], na.int32())
    array = stream.read_all()
    assert array.schema.type == na.Type.INT32
    assert array.to_pylist() == [1, 2, 3]


def test_array_stream_read_next():
    stream = na.ArrayStream([1, 2, 3], na.int32())
    array = stream.read_next()
    assert array.schema.type == na.Type.INT32
    assert array.to_pylist() == [1, 2, 3]

    with pytest.raises(StopIteration):
        stream.read_next()


def test_array_stream_close():
    stream = na.ArrayStream([], na.int32())
    stream.close()
    with pytest.raises(RuntimeError, match="array stream is released"):
        stream.read_all()


def test_array_stream_context_manager():
    stream = na.ArrayStream([], na.int32())
    with stream:
        pass

    with pytest.raises(RuntimeError, match="array stream is released"):
        stream.read_all()


def test_array_stream_from_readable():
    stream = na.ArrayStream.from_readable(InputStream.example_bytes())
    assert stream.schema.type == na.Type.STRUCT
    assert list(stream.read_all().iter_tuples()) == [(1,), (2,), (3,)]


def test_array_stream_from_path():
    with tempfile.TemporaryDirectory() as td:
        path = os.path.join(td, "test.arrows")
        with open(path, "wb") as f:
            f.write(InputStream.example_bytes())

        stream = na.ArrayStream.from_path(path)
        assert stream.schema.type == na.Type.STRUCT
        assert list(stream.read_all().iter_tuples()) == [(1,), (2,), (3,)]


def test_array_stream_from_url():
    with tempfile.TemporaryDirectory() as td:
        path = os.path.join(td, "test.arrows")
        with open(path, "wb") as f:
            f.write(InputStream.example_bytes())

        uri = pathlib.Path(path).as_uri()
        with na.ArrayStream.from_url(uri) as stream:
            assert stream.schema.type == na.Type.STRUCT
            assert list(stream.read_all().iter_tuples()) == [(1,), (2,), (3,)]