# 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 pytest
from nanoarrow.c_array_stream import CArrayStream

import nanoarrow as na


def test_c_array_stream_from_c_array_stream():
    # Wrapping an existing stream is a no-op
    array_stream = CArrayStream.from_c_arrays([], na.c_schema(na.int32()))
    stream_from_stream = na.c_array_stream(array_stream)
    assert stream_from_stream is array_stream

    # With requested_schema should go through capsule
    array_stream = CArrayStream.from_c_arrays([], na.c_schema(na.int32()))
    with pytest.raises(NotImplementedError):
        na.c_array_stream(array_stream, na.int64())


def test_c_array_stream_from_capsule_protocol():
    # Use wrapper object to ensure this is the path taken in the constructor
    class CArrayStreamWrapper:
        def __init__(self, obj):
            self.obj = obj

        def __arrow_c_stream__(self, *args, **kwargs):
            return self.obj.__arrow_c_stream__(*args, **kwargs)

    array_stream = CArrayStream.from_c_arrays([], na.c_schema(na.int32()))
    array_stream_wrapper = CArrayStreamWrapper(array_stream)
    from_protocol = na.c_array_stream(array_stream_wrapper)
    assert array_stream.is_valid() is False
    assert from_protocol.get_schema().format == "i"


def test_c_array_stream_from_old_pyarrow():
    # Simulate a pyarrow RecordBatchReader with no __arrow_c_stream__
    class MockLegacyPyarrowRecordBatchReader:
        def __init__(self, obj):
            self.obj = obj

        def _export_to_c(self, *args):
            return self.obj._export_to_c(*args)

    MockLegacyPyarrowRecordBatchReader.__module__ = "pyarrow.lib"

    pa = pytest.importorskip("pyarrow")
    reader = pa.RecordBatchReader.from_batches(pa.schema([]), [])
    mock_reader = MockLegacyPyarrowRecordBatchReader(reader)

    array_stream = na.c_array_stream(mock_reader)
    assert array_stream.get_schema().format == "+s"


def test_c_array_stream_from_bare_capsule():
    array_stream = CArrayStream.from_c_arrays([], na.c_schema(na.int32()))

    # Check from bare capsule without supplying a schema
    capsule = array_stream.__arrow_c_stream__()
    from_capsule = na.c_array_stream(capsule)
    assert from_capsule.get_schema().format == "i"

    array_stream = CArrayStream.from_c_arrays([], na.c_schema(na.int32()))
    capsule = array_stream.__arrow_c_stream__()

    with pytest.raises(TypeError, match="Can't import c_array_stream"):
        na.c_array_stream(capsule, na.int32())


def test_c_array_stream_from_c_array_fallback():
    # Check that arrays are valid input
    c_array = na.c_array([1, 2, 3], na.int32())
    array_stream = na.c_array_stream(c_array)
    assert array_stream.get_schema().format == "i"
    arrays = list(array_stream)
    assert len(arrays) == 1
    assert arrays[0].buffers == c_array.buffers

    # Check fallback with schema
    array_stream = na.c_array_stream([1, 2, 3], na.int32())
    assert array_stream.get_schema().format == "i"
    arrays = list(array_stream)
    assert len(arrays) == 1


def test_c_array_stream_error():
    msg = "Can't resolve ArrayBuilder"
    with pytest.raises(TypeError, match=msg):
        na.c_array_stream(None)


def test_array_stream_from_arrays_schema():
    schema_in = na.c_schema(na.int32())

    stream = CArrayStream.from_c_arrays([], schema_in)
    assert schema_in.is_valid()
    assert list(stream) == []
    assert stream.get_schema().format == "i"

    # Check move of schema
    CArrayStream.from_c_arrays([], schema_in, move=True)
    assert schema_in.is_valid() is False
    assert stream.get_schema().format == "i"


def test_array_stream_from_arrays():
    schema_in = na.c_schema(na.int32())
    array_in = na.c_array([1, 2, 3], na.int32())
    array_in_buffers = array_in.buffers

    stream = CArrayStream.from_c_arrays([array_in], schema_in)
    assert array_in.is_valid()
    arrays = list(stream)
    assert len(arrays) == 1
    assert arrays[0].buffers == array_in_buffers

    # Check move of array
    stream = CArrayStream.from_c_arrays([array_in], schema_in, move=True)
    assert array_in.is_valid() is False
    arrays = list(stream)
    assert len(arrays) == 1
    assert arrays[0].buffers == array_in_buffers


def test_array_stream_from_arrays_validate():
    schema_in = na.c_schema(na.null())
    array_in = na.c_array([1, 2, 3], na.int32())

    # Check that we can skip validation and proceed without error
    stream = CArrayStream.from_c_arrays([array_in], schema_in, validate=False)
    arrays = list(stream)
    assert len(arrays) == 1
    assert arrays[0].n_buffers == 2

    # ...but that validation does happen by default
    with pytest.raises(ValueError, match="Expected schema"):
        CArrayStream.from_c_arrays([array_in], schema_in)