from collections import deque
from datetime import (
    date,
    datetime,
    time,
    timedelta,
)
import re
import string

import numpy as np
import pytest

from pandas.errors import Pandas4Warning
import pandas.util._test_decorators as td

import pandas as pd
import pandas._testing as tm
from pandas.arrays import SparseArray


@pytest.fixture(params=[np.add, np.logaddexp])
def ufunc(request):
    # dunder op
    return request.param


@pytest.fixture(
    params=[True, False],
    ids=["sparse", "dense"],
)
def sparse(request):
    return request.param


@pytest.fixture
def arrays_for_binary_ufunc():
    """
    A pair of random, length-100 integer-dtype arrays, that are mostly 0.
    """
    a1 = np.random.default_rng(2).integers(0, 10, 100, dtype="int64")
    a2 = np.random.default_rng(2).integers(0, 10, 100, dtype="int64")
    a1[::3] = 0
    a2[::4] = 0
    return a1, a2


@pytest.mark.parametrize("ufunc", [np.positive, np.floor, np.exp])
def test_unary_ufunc(ufunc, sparse):
    # Test that ufunc(pd.Series) == pd.Series(ufunc)
    arr = np.random.default_rng(2).integers(0, 10, 10, dtype="int64")
    arr[::2] = 0
    if sparse:
        arr = SparseArray(arr, dtype=pd.SparseDtype("int64", 0))

    index = list(string.ascii_letters[:10])
    name = "name"
    series = pd.Series(arr, index=index, name=name)

    result = ufunc(series)
    expected = pd.Series(ufunc(arr), index=index, name=name)
    tm.assert_series_equal(result, expected)


@pytest.mark.parametrize("flip", [True, False], ids=["flipped", "straight"])
def test_binary_ufunc_with_array(flip, sparse, ufunc, arrays_for_binary_ufunc):
    # Test that ufunc(pd.Series(a), array) == pd.Series(ufunc(a, b))
    a1, a2 = arrays_for_binary_ufunc
    if sparse:
        a1 = SparseArray(a1, dtype=pd.SparseDtype("int64", 0))
        a2 = SparseArray(a2, dtype=pd.SparseDtype("int64", 0))

    name = "name"  # op(pd.Series, array) preserves the name.
    series = pd.Series(a1, name=name)
    other = a2

    array_args = (a1, a2)
    series_args = (series, other)  # ufunc(series, array)

    if flip:
        array_args = reversed(array_args)
        series_args = reversed(series_args)  # ufunc(array, series)

    expected = pd.Series(ufunc(*array_args), name=name)
    result = ufunc(*series_args)
    tm.assert_series_equal(result, expected)


@pytest.mark.parametrize("flip", [True, False], ids=["flipped", "straight"])
def test_binary_ufunc_with_index(flip, sparse, ufunc, arrays_for_binary_ufunc):
    # Test that
    #   * func(pd.Series(a), pd.Series(b)) == pd.Series(ufunc(a, b))
    #   * ufunc(Index, pd.Series) dispatches to pd.Series (returns a pd.Series)
    a1, a2 = arrays_for_binary_ufunc
    if sparse:
        a1 = SparseArray(a1, dtype=pd.SparseDtype("int64", 0))
        a2 = SparseArray(a2, dtype=pd.SparseDtype("int64", 0))

    name = "name"  # op(pd.Series, array) preserves the name.
    series = pd.Series(a1, name=name)

    other = pd.Index(a2, name=name).astype("int64")

    array_args = (a1, a2)
    series_args = (series, other)  # ufunc(series, array)

    if flip:
        array_args = reversed(array_args)
        series_args = reversed(series_args)  # ufunc(array, series)

    expected = pd.Series(ufunc(*array_args), name=name)
    result = ufunc(*series_args)
    tm.assert_series_equal(result, expected)


@pytest.mark.parametrize("shuffle", [True, False], ids=["unaligned", "aligned"])
@pytest.mark.parametrize("flip", [True, False], ids=["flipped", "straight"])
def test_binary_ufunc_with_series(
    flip, shuffle, sparse, ufunc, arrays_for_binary_ufunc
):
    # Test that
    #   * func(pd.Series(a), pd.Series(b)) == pd.Series(ufunc(a, b))
    #   with alignment between the indices
    a1, a2 = arrays_for_binary_ufunc
    if sparse:
        a1 = SparseArray(a1, dtype=pd.SparseDtype("int64", 0))
        a2 = SparseArray(a2, dtype=pd.SparseDtype("int64", 0))

    name = "name"  # op(pd.Series, array) preserves the name.
    series = pd.Series(a1, name=name)
    other = pd.Series(a2, name=name)

    idx = np.random.default_rng(2).permutation(len(a1))

    if shuffle:
        other = other.take(idx)
        if flip:
            index = other.align(series)[0].index
        else:
            index = series.align(other)[0].index
    else:
        index = series.index

    array_args = (a1, a2)
    series_args = (series, other)  # ufunc(series, array)

    if flip:
        array_args = tuple(reversed(array_args))
        series_args = tuple(reversed(series_args))  # ufunc(array, series)

    expected = pd.Series(ufunc(*array_args), index=index, name=name)
    result = ufunc(*series_args)
    tm.assert_series_equal(result, expected)


@pytest.mark.parametrize("flip", [True, False])
def test_binary_ufunc_scalar(ufunc, sparse, flip, arrays_for_binary_ufunc):
    # Test that
    #   * ufunc(pd.Series, scalar) == pd.Series(ufunc(array, scalar))
    #   * ufunc(pd.Series, scalar) == ufunc(scalar, pd.Series)
    arr, _ = arrays_for_binary_ufunc
    if sparse:
        arr = SparseArray(arr)
    other = 2
    series = pd.Series(arr, name="name")

    series_args = (series, other)
    array_args = (arr, other)

    if flip:
        series_args = tuple(reversed(series_args))
        array_args = tuple(reversed(array_args))

    expected = pd.Series(ufunc(*array_args), name="name")
    result = ufunc(*series_args)

    tm.assert_series_equal(result, expected)


@pytest.mark.parametrize("ufunc", [np.divmod])  # TODO: np.modf, np.frexp
@pytest.mark.parametrize("shuffle", [True, False])
def test_multiple_output_binary_ufuncs(ufunc, sparse, shuffle, arrays_for_binary_ufunc):
    # Test that
    #  the same conditions from binary_ufunc_scalar apply to
    #  ufuncs with multiple outputs.

    a1, a2 = arrays_for_binary_ufunc
    # work around https://github.com/pandas-dev/pandas/issues/26987
    a1[a1 == 0] = 1
    a2[a2 == 0] = 1

    if sparse:
        a1 = SparseArray(a1, dtype=pd.SparseDtype("int64", 0))
        a2 = SparseArray(a2, dtype=pd.SparseDtype("int64", 0))

    s1 = pd.Series(a1)
    s2 = pd.Series(a2)

    if shuffle:
        # ensure we align before applying the ufunc
        s2 = s2.sample(frac=1)

    expected = ufunc(a1, a2)
    assert isinstance(expected, tuple)

    result = ufunc(s1, s2)
    assert isinstance(result, tuple)
    tm.assert_series_equal(result[0], pd.Series(expected[0]))
    tm.assert_series_equal(result[1], pd.Series(expected[1]))


def test_multiple_output_ufunc(sparse, arrays_for_binary_ufunc):
    # Test that the same conditions from unary input apply to multi-output
    # ufuncs
    arr, _ = arrays_for_binary_ufunc

    if sparse:
        arr = SparseArray(arr)

    series = pd.Series(arr, name="name")
    result = np.modf(series)
    expected = np.modf(arr)

    assert isinstance(result, tuple)
    assert isinstance(expected, tuple)

    tm.assert_series_equal(result[0], pd.Series(expected[0], name="name"))
    tm.assert_series_equal(result[1], pd.Series(expected[1], name="name"))


def test_binary_ufunc_drops_series_name(ufunc, sparse, arrays_for_binary_ufunc):
    # Drop the names when they differ.
    a1, a2 = arrays_for_binary_ufunc
    s1 = pd.Series(a1, name="a")
    s2 = pd.Series(a2, name="b")

    result = ufunc(s1, s2)
    assert result.name is None


def test_binary_ufunc_aligns_duplicate_labels():
    # GH#54416
    ser1 = pd.Series([1.0, 2.0], index=[0, 0])
    ser2 = pd.Series([3.0, 4.0], index=[0, 1])
    result = np.heaviside(ser1, ser2)
    expected = pd.Series(
        np.heaviside([1.0, 2.0, np.nan], [3.0, 3.0, 4.0]), index=[0, 0, 1]
    )
    tm.assert_series_equal(result, expected)


def test_binary_ufunc_aligns_index_name():
    # GH#54416 align (not union) sets the result's index name, matching
    #  the behavior of the equivalent binary operator (e.g. `+`).
    ser1 = pd.Series([1.0, 2.0], index=pd.Index([1, 2], name="x"))
    ser2 = pd.Series([1.0, 2.0], index=pd.Index([2, 3], name="y"))

    result = np.heaviside(ser1, ser2)
    expected = (ser1 + ser2).index.name
    assert result.index.name == expected == "x"

    result = np.heaviside(ser2, ser1)
    expected = (ser2 + ser1).index.name
    assert result.index.name == expected == "y"


def test_binary_ufunc_out_pandas_object():
    # GH#43190 passing a Series as the ufunc `out` argument used to recurse
    #  infinitely (RecursionError / segfault) instead of writing the result.
    a = pd.Series([1, 2])
    b = pd.Series([3, 4])
    out = pd.Series([0, 0])

    result = np.fmin(a, b, out=out)

    expected = pd.Series([1, 2])
    tm.assert_series_equal(result, expected)
    # the result is also written into `out` in place
    tm.assert_series_equal(out, expected)


def test_object_series_ok():
    class Dummy:
        def __init__(self, value) -> None:
            self.value = value

        def __add__(self, other):
            return self.value + other.value

    arr = np.array([Dummy(0), Dummy(1)])
    ser = pd.Series(arr)
    tm.assert_series_equal(np.add(ser, ser), pd.Series(np.add(ser, arr)))
    tm.assert_series_equal(np.add(ser, Dummy(1)), pd.Series(np.add(ser, Dummy(1))))


@pytest.fixture(
    params=[
        pd.array([1, 3, 2], dtype=np.int64),
        pd.array([1, 3, 2], dtype="Int64"),
        pd.array([1, 3, 2], dtype="Float32"),
        pd.array([1, 10, 2], dtype="Sparse[int]"),
        pd.to_datetime(["2000", "2010", "2001"]),
        pd.to_datetime(["2000", "2010", "2001"]).tz_localize("CET"),
        pd.to_datetime(["2000", "2010", "2001"]).to_period(freq="D"),
        pd.to_timedelta(["1 Day", "3 Days", "2 Days"]),
        pd.IntervalIndex([pd.Interval(0, 1), pd.Interval(2, 3), pd.Interval(1, 2)]),
    ],
    ids=lambda x: str(x.dtype),
)
def values_for_np_reduce(request):
    # min/max tests assume that these are monotonic increasing
    return request.param


class TestNumpyReductions:
    # TODO: cases with NAs, axis kwarg for DataFrame

    def test_multiply(self, values_for_np_reduce, box_with_array):
        box = box_with_array
        values = values_for_np_reduce

        with tm.assert_produces_warning(None):
            obj = box(values)

        if values.dtype.kind in "iuf":
            result = np.multiply.reduce(obj)
            if box is pd.DataFrame:
                expected = obj.prod(numeric_only=False)
                tm.assert_series_equal(result, expected)
            elif box is pd.Index:
                # Index has no 'prod'
                expected = obj._values.prod()
                assert result == expected
            else:
                expected = obj.prod()
                assert result == expected
        else:
            msg = "|".join(
                [
                    "does not support operation",
                    "unsupported operand type",
                    "ufunc 'multiply' cannot use operands",
                ]
            )
            with pytest.raises(TypeError, match=msg):
                np.multiply.reduce(obj)

    def test_add(self, values_for_np_reduce, box_with_array):
        box = box_with_array
        values = values_for_np_reduce

        with tm.assert_produces_warning(None):
            obj = box(values)

        if values.dtype.kind in "miuf":
            result = np.add.reduce(obj)
            if box is pd.DataFrame:
                expected = obj.sum(numeric_only=False)
                tm.assert_series_equal(result, expected)
            elif box is pd.Index:
                # Index has no 'sum'
                expected = obj._values.sum()
                assert result == expected
            else:
                expected = obj.sum()
                assert result == expected
        else:
            msg = "|".join(
                [
                    "does not support operation",
                    "unsupported operand type",
                    "ufunc 'add' cannot use operands",
                ]
            )
            with pytest.raises(TypeError, match=msg):
                np.add.reduce(obj)

    def test_max(self, values_for_np_reduce, box_with_array, using_python_scalars):
        box = box_with_array
        values = values_for_np_reduce

        same_type = True
        if box is pd.Index and values.dtype.kind in "if":
            # ATM Index casts to object, so we get python ints/floats
            same_type = False

        with tm.assert_produces_warning(None):
            obj = box(values)

        result = np.maximum.reduce(obj)
        if box is pd.DataFrame:
            # TODO: cases with axis kwarg
            expected = obj.max(numeric_only=False)
            tm.assert_series_equal(result, expected)
        else:
            expected = values[1]
            if (
                using_python_scalars
                and values.dtype.kind in "if"
                and box is not pd.array
            ):
                # GH#64266
                expected = expected.item()
            assert result == expected
            if same_type:
                # check we have e.g. Timestamp instead of dt64
                assert type(result) == type(expected)

    def test_min(self, values_for_np_reduce, box_with_array, using_python_scalars):
        box = box_with_array
        values = values_for_np_reduce

        same_type = True
        if box is pd.Index and values.dtype.kind in "if":
            # ATM Index casts to object, so we get python ints/floats
            same_type = False

        with tm.assert_produces_warning(None):
            obj = box(values)

        result = np.minimum.reduce(obj)
        if box is pd.DataFrame:
            expected = obj.min(numeric_only=False)
            tm.assert_series_equal(result, expected)
        else:
            expected = values[0]
            if (
                using_python_scalars
                and values.dtype.kind in "if"
                and box is not pd.array
            ):
                # GH#64266
                expected = expected.item()
            assert result == expected
            if same_type:
                # check we have e.g. Timestamp instead of dt64
                assert type(result) == type(expected)


@pytest.mark.parametrize("type_", [list, deque, tuple])
def test_binary_ufunc_other_types(type_):
    a = pd.Series([1, 2, 3], name="name")
    b = type_([3, 4, 5])

    # GH#62423 tuple/deque operands are deprecated in favor of scalar-like
    warn = None if type_ is list else Pandas4Warning
    depr_msg = "In a future version these will be treated as scalar-like"
    with tm.assert_produces_warning(warn, match=depr_msg):
        result = np.add(a, b)
    expected = pd.Series(np.add(a.to_numpy(), b), name="name")
    tm.assert_series_equal(result, expected)


def test_object_dtype_ok():
    class Thing:
        def __init__(self, value) -> None:
            self.value = value

        def __add__(self, other):
            other = getattr(other, "value", other)
            return type(self)(self.value + other)

        def __eq__(self, other) -> bool:
            return type(other) is Thing and self.value == other.value

        def __repr__(self) -> str:
            return f"Thing({self.value})"

    s = pd.Series([Thing(1), Thing(2)])
    result = np.add(s, Thing(1))
    expected = pd.Series([Thing(2), Thing(3)])
    tm.assert_series_equal(result, expected)


def test_outer():
    # https://github.com/pandas-dev/pandas/issues/27186
    ser = pd.Series([1, 2, 3])
    obj = np.array([1, 2, 3])

    with pytest.raises(NotImplementedError, match="^$"):
        np.subtract.outer(ser, obj)


def test_np_matmul():
    # GH26650
    df1 = pd.DataFrame(data=[[-1, 1, 10]])
    df2 = pd.DataFrame(data=[-1, 1, 10])
    expected = pd.DataFrame(data=[102])

    result = np.matmul(df1, df2)
    tm.assert_frame_equal(expected, result)


@pytest.mark.parametrize("box", [pd.Index, pd.Series])
def test_np_matmul_1D(box, using_python_scalars):
    result = np.matmul(box([1, 2]), box([2, 3]))
    assert result == 8
    if using_python_scalars:
        assert type(result) == int, type(result)
    else:
        assert type(result) == np.int64, type(result)


def test_array_ufuncs_for_many_arguments():
    # GH39853
    def add3(x, y, z):
        return x + y + z

    ufunc = np.frompyfunc(add3, 3, 1)
    ser = pd.Series([1, 2])

    result = ufunc(ser, ser, 1)
    expected = pd.Series([3, 5], dtype=object)
    tm.assert_series_equal(result, expected)

    df = pd.DataFrame([[1, 2]])

    msg = (
        "Cannot apply ufunc <ufunc 'add3 (vectorized)'> "
        "to mixed DataFrame and Series inputs."
    )
    with pytest.raises(NotImplementedError, match=re.escape(msg)):
        ufunc(ser, ser, df)


def test_np_trunc():
    # This used to test np.fix, which is not a ufunc but is composed of
    # several ufunc calls under the hood with `out` and `where` keywords. But numpy
    # is deprecating that (or at least discussing deprecating) in favor of np.trunc,
    # which _is_ a ufunc without the out keyword usage.
    ser = pd.Series([-1.5, -0.5, 0.5, 1.5])
    result = np.trunc(ser)
    expected = pd.Series([-1.0, -0.0, 0.0, 1.0])
    tm.assert_series_equal(result, expected)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("box", [pd.Series, pd.Index, pd.array])
@pytest.mark.parametrize(
    "dtype",
    [
        "datetime64[ns]",
        "timedelta64[ns]",
        "datetime64[ns, US/Pacific]",
        "period[D]",
        "Sparse[datetime64[ns]]",
        pytest.param("timestamp[ns][pyarrow]", marks=td.skip_if_no("pyarrow")),
    ],
)
def test_binary_logical_ufunc_datetimelike_raises(func, box, dtype):
    # GH#68524 the ufunc spelling of &, |, ^, which GH#68452 made raise.  The
    #  numpy-dtype operands used to be truthiness-tested on their i8 values; the
    #  ones np.asarray flattens to object leaked the scalar itself into the result
    left = pd.Series([True, False], dtype="bool")
    values = ["1D", "2D"] if "timedelta" in dtype else ["2016-01-01", "2016-01-02"]
    right = box(pd.array(values, dtype=dtype))

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(left, right)
    with pytest.raises(TypeError, match=msg):
        func(right, left)


@pytest.mark.parametrize("box", [pd.Series, pd.Index, pd.array])
@pytest.mark.parametrize(
    "dtype", ["datetime64[ns]", "datetime64[ns, US/Pacific]", "period[D]"]
)
def test_logical_not_ufunc_datetimelike_raises(box, dtype):
    # GH#68524 the unary one.  numpy already raised for datetime64; the dtypes it
    #  flattens to object returned all-False
    obj = box(pd.array(["2016-01-01", "2016-01-02"], dtype=dtype))

    with pytest.raises(TypeError, match="cannot perform the numpy op logical_not"):
        np.logical_not(obj)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize(
    "scalar",
    [
        pd.Timestamp("2016-01-01"),
        pd.Timestamp("2016-01-01", tz="US/Pacific"),
        pd.Timedelta(days=1),
        pd.Period("2016-01-01", "D"),
        pd.NaT,
        np.datetime64("2016-01-01"),
        np.timedelta64(1, "D"),
        pd.offsets.Day(),
        datetime(2016, 1, 1),
        date(2016, 1, 1),
        time(12, 0),
        timedelta(days=1),
    ],
    ids=lambda x: type(x).__name__,
)
@pytest.mark.parametrize("left_dtype", ["bool", object])
def test_binary_logical_ufunc_datetimelike_scalar_raises(func, scalar, left_dtype):
    # GH#68524 -- GH#68452 left scalar operands to the paths that already rejected
    #  them for &/|/^, and the ufuncs have no such path.  The object-dtype left
    #  operand raises here where `&` truth-tests it, because what it used to return
    #  was the scalar broadcast to a datetimelike result, not a logical one
    left = pd.Series([True, False], dtype=left_dtype)

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(left, scalar)
    with pytest.raises(TypeError, match=msg):
        func(scalar, left)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("left_dtype", ["boolean", "Int64", "Sparse[bool]"])
def test_binary_logical_ufunc_datetimelike_scalar_unboxed(func, left_dtype):
    # GH#68524 a datetimelike scalar is not in any _HANDLED_TYPES, so
    #  __array_ufunc__ defers on an unboxed left operand unless the guard runs first
    left = pd.array([True, False], dtype=left_dtype)
    scalar = pd.Timestamp("2016-01-01")

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(left, scalar)
    with pytest.raises(TypeError, match=msg):
        func(scalar, left)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("left_dtype", ["bool", "boolean", "Int64", "Sparse[bool]"])
@pytest.mark.parametrize("box", [pd.Series, pd.Index, pd.array])
def test_binary_logical_ufunc_datetimelike_left_operand_boxes(func, left_dtype, box):
    # GH#68524 an M8 ndarray has no __array_ufunc__ override to raise first, so
    #  it is what makes each box fall back on its own guard
    left = box(pd.array([True, False], dtype=left_dtype))
    right = np.array(["2016-01-01", "2016-01-02"], dtype="M8[ns]")

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(left, right)
    with pytest.raises(TypeError, match=msg):
        func(right, left)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("dtype", ["M8[ns]", "m8[ns]"])
def test_binary_logical_ufunc_datetimelike_numpy_extension_array(func, dtype):
    # GH#68524 NumpyExtensionArray has its own __array_ufunc__ but is in
    #  _HANDLED_TYPES, so pandas computes the result instead of deferring to it
    left = pd.Series([True, False])
    right = pd.arrays.NumpyExtensionArray(np.array([1, 2], dtype=dtype))

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(left, right)
    with pytest.raises(TypeError, match=msg):
        func(right, left)
    with pytest.raises(TypeError, match="cannot perform the numpy op logical_not"):
        np.logical_not(right)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("left_dtype", ["double[pyarrow]", "int64[pyarrow]"])
@td.skip_if_no("pyarrow")
def test_binary_logical_ufunc_datetimelike_arrow_nan_na(func, left_dtype):
    # GH#68524 the numeric fast path added by GH#62506 returns without reaching
    #  ExtensionArray.__array_ufunc__, so it needs a guard of its own
    left = pd.array([1, 2], dtype=left_dtype)

    msg = f"cannot perform the numpy op {func.__name__}"
    with pd.option_context("future.distinguish_nan_and_na", True):
        with pytest.raises(TypeError, match=msg):
            func(left, pd.Timestamp("2016-01-01"))
        with pytest.raises(TypeError, match=msg):
            func(left, np.array(["2016-01-01", "2016-01-02"], dtype="M8[ns]"))


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@td.skip_if_no("pyarrow")
def test_binary_logical_ufunc_nat_scalar_arrow_backed(func):
    # GH#68524 arrow boxes an NA-like scalar as a null, so `&` takes NaT where the
    #  other backends raise; an NaT datetime64 was truth-tested on its int64
    left = pd.Series([True, False], dtype="bool[pyarrow]")

    msg = f"cannot perform the numpy op {func.__name__}"
    for scalar in [pd.NaT, np.datetime64("NaT", "ns")]:
        with pytest.raises(TypeError, match=msg):
            func(left, scalar)
        with pytest.raises(TypeError, match=msg):
            func(scalar, left)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("method", ["reduce", "accumulate"])
@pytest.mark.parametrize(
    "dtype",
    ["datetime64[ns]", "timedelta64[ns]", "datetime64[ns, US/Pacific]", "period[D]"],
)
def test_logical_ufunc_methods_datetimelike_raise(func, method, dtype):
    # GH#68524 reduce and accumulate take the same fallback as __call__
    values = ["1D", "2D"] if "timedelta" in dtype else ["2016-01-01", "2016-01-02"]
    ser = pd.Series(pd.array(values, dtype=dtype))

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        getattr(func, method)(ser)


class _ThirdPartyArray:
    # stands in for e.g. a polars Series; the default dtype has no "kind"
    def __init__(self, dtype="bool") -> None:
        self.dtype = dtype

    def __array_ufunc__(self, ufunc, method, *inputs, **kwargs):
        return np.array([True, False])


class _ThirdPartyArrayNoUfunc:
    # without __array_ufunc__ nothing defers, so this is the shape the guard
    #  actually reads an unrecognized dtype off
    def __init__(self, dtype) -> None:
        self.dtype = dtype

    def __array__(self, dtype=None, copy=None) -> np.ndarray:
        return np.array([True, False])


class _NonStringKindDtype:
    # a dtype whose "kind" is not one of numpy's single-character codes
    kind = 0


class _SymbolicArray:
    # a symbolic array, as pd.col builds one: every attribute it is asked for is
    #  another expression, and an expression has no truth value
    __hash__ = object.__hash__

    def __getattr__(self, name):
        return self

    def __eq__(self, other):
        return self

    def __bool__(self) -> bool:
        raise TypeError("an expression has no truth value")

    def __array_ufunc__(self, ufunc, method, *inputs, **kwargs):
        return np.array([True, False])


@pytest.mark.parametrize(
    "dtype", ["bool", _NonStringKindDtype()], ids=["no-kind", "non-string-kind"]
)
def test_logical_op_third_party(dtype):
    # GH#68524 the datetimelike guard must not crash on a dtype it does not
    #  recognize, whether that dtype has no "kind" or a non-string one
    left = pd.Series([True, True])
    right = _ThirdPartyArray(dtype)
    expected = np.array([True, False])

    tm.assert_numpy_array_equal(np.logical_and(left, right), expected)
    tm.assert_numpy_array_equal(np.logical_and(right, left), expected)

    tm.assert_series_equal(left & right, pd.Series(expected))


@pytest.mark.parametrize("box", [pd.Series, pd.Index, pd.array])
@pytest.mark.parametrize("dtype", ["M8[ns]", "m8[ns]"])
def test_logical_ufunc_third_party_datetimelike(box, dtype):
    # GH#68524 pandas defers to an operand with its own __array_ufunc__, so the
    #  guard must not raise on its behalf over a dtype it does not own
    left = box(pd.array([True, True]))
    foreign = _ThirdPartyArray(np.dtype(dtype))

    # each order gets its own baseline; a left-hand Index re-wraps, the others do not
    tm.assert_equal(
        np.logical_and(left, foreign), np.logical_and(left, _ThirdPartyArray())
    )
    tm.assert_equal(
        np.logical_and(foreign, left), np.logical_and(_ThirdPartyArray(), left)
    )


@pytest.mark.parametrize("box", [pd.Series, pd.Index, pd.array])
@pytest.mark.parametrize("dtype", ["M8[ns]", "m8[ns]"])
def test_logical_ufunc_third_party_pandas_datetimelike(box, dtype):
    # GH#68524 the deferral is of the whole op, so it holds when the
    #  datetimelike operand is the pandas one too
    values = ["1D", "2D"] if dtype.startswith("m") else ["2016-01-01", "2016-01-02"]
    left = box(pd.array(values, dtype=dtype))
    control = box(pd.array([True, True]))
    foreign = _ThirdPartyArray()

    # baselines as in the test above
    tm.assert_equal(np.logical_and(left, foreign), np.logical_and(control, foreign))
    tm.assert_equal(np.logical_and(foreign, left), np.logical_and(foreign, control))


class _NdarraySubclass(np.ndarray):
    # unwraps itself and re-dispatches, the usual ndarray-wrapper shape
    def __array_ufunc__(self, ufunc, method, *inputs, **kwargs):
        inputs = tuple(
            np.asarray(obj) if isinstance(obj, _NdarraySubclass) else obj
            for obj in inputs
        )
        return getattr(ufunc, method)(*inputs, **kwargs)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("box", [pd.Series, pd.Index, pd.array])
@pytest.mark.parametrize("dtype", ["M8[ns]", "m8[ns]"])
def test_logical_ufunc_ndarray_subclass_datetimelike(func, box, dtype):
    # GH#68524 np.ndarray is in every _HANDLED_TYPES, so pandas computes against a
    #  subclass of one rather than deferring to it, and the guard has to stay on
    left = box(pd.array([True, True]))
    right = np.array([1, 2], dtype=dtype).view(_NdarraySubclass)

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(left, right)
    with pytest.raises(TypeError, match=msg):
        func(right, left)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
@pytest.mark.parametrize("box", [pd.Series, pd.Index, pd.array])
@pytest.mark.parametrize("dtype", ["M8[ns]", "m8[ns]", "period[D]"])
def test_logical_ufunc_na_scalar_datetimelike(func, box, dtype):
    # GH#68524 NA carries an __array_ufunc__ of its own, but Index and the bare
    #  arrays compute against it rather than defer, so it must not exempt the op
    values = ["1D", "2D"] if dtype.startswith("m") else ["2016-01-01", "2016-01-02"]
    obj = box(pd.array(values, dtype=dtype))

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(obj, pd.NA)
    with pytest.raises(TypeError, match=msg):
        func(pd.NA, obj)


@pytest.mark.parametrize("func", [np.logical_and, np.logical_or, np.logical_xor])
def test_logical_ufunc_out_kwarg_datetimelike_raises(func):
    # GH#68524 out= took the same fallback as __call__, and the datetimelike
    #  values were truth-tested into it
    left = pd.Series([True, False])
    right = np.array(["2016-01-01", "2016-01-02"], dtype="M8[ns]")

    msg = f"cannot perform the numpy op {func.__name__}"
    with pytest.raises(TypeError, match=msg):
        func(left, right, out=np.zeros(2, dtype=bool))


@pytest.mark.parametrize(
    "dtype", ["bool", _NonStringKindDtype()], ids=["no-kind", "non-string-kind"]
)
def test_logical_ufunc_third_party_no_array_ufunc(dtype):
    # GH#68524 the ufunc half of test_logical_op_third_party: an operand pandas
    #  does not defer to is the one whose dtype the guard has to read
    result = np.logical_and(pd.Series([True, True]), _ThirdPartyArrayNoUfunc(dtype))
    tm.assert_series_equal(result, pd.Series([True, False]))


@pytest.mark.parametrize("box", [pd.array, pd.arrays.SparseArray])
def test_logical_ufunc_symbolic_operand(box):
    # GH#68524 deciding whether pandas defers must not read an attribute off the
    #  instance: the ABCs probe `_typ`, which an expression answers with itself
    left = box([True, True])
    expected = np.array([True, False])

    tm.assert_numpy_array_equal(np.logical_and(left, _SymbolicArray()), expected)
    tm.assert_numpy_array_equal(np.logical_and(_SymbolicArray(), left), expected)