text:
from __future__ import annotations

from functools import reduce
from itertools import product
import operator
import random
import warnings

import numpy as np
import pytest

from pandas.errors import (
    NumExprClobberingError,
    PerformanceWarning,
    UndefinedVariableError,
)
import pandas.util._test_decorators as td

from pandas.core.dtypes.common import (
    is_bool,
    is_float,
    is_list_like,
    is_scalar,
)

import pandas as pd
from pandas import (
    DataFrame,
    Series,
    date_range,
)
import pandas._testing as tm
from pandas.core.computation import pytables
from pandas.core.computation.engines import ENGINES
import pandas.core.computation.expr as expr
from pandas.core.computation.expr import (
    BaseExprVisitor,
    PandasExprVisitor,
    PythonExprVisitor,
)
from pandas.core.computation.expressions import (
    NUMEXPR_INSTALLED,
    USE_NUMEXPR,
)
from pandas.core.computation.ops import (
    ARITH_OPS_SYMS,
    SPECIAL_CASE_ARITH_OPS_SYMS,
    _binary_math_ops,
    _binary_ops_dict,
    _unary_math_ops,
)
from pandas.core.computation.scope import DEFAULT_GLOBALS

@pytest.fixture(
    params=(
        pytest.param(
            engine,
            marks=[
                pytest.mark.skipif(
                    engine == "numexpr" and not USE_NUMEXPR,
                    reason=f"numexpr enabled->{USE_NUMEXPR}, "
                    f"installed->{NUMEXPR_INSTALLED}",
                ),
                td.skip_if_no_ne,
            ],
        )
        for engine in ENGINES
    )
)
def engine(request):
    return request.param

@pytest.fixture(params=expr.PARSERS)
def parser(request):
    return request.param

@pytest.fixture(params=list(_unary_math_ops) if NUMEXPR_INSTALLED else [])
def unary_fns_for_ne(request):
    return request.param

def _eval_single_bin(lhs, cmp1, rhs, engine):
    c = _binary_ops_dict[cmp1]
    if ENGINES[engine].has_neg_frac:
        try:
            return c(lhs, rhs)
        except ValueError as e:
            if str(e).startswith(
                "negative number cannot be raised to a fractional power"
            ):
                return np.nan
            raise
    return c(lhs, rhs)

# TODO: using range(5) here is a kludge
@pytest.fixture(
    params=list(range(5)),
    ids=["DataFrame", "Series", "SeriesNaN", "DataFrameNaN", "float"],
)
def lhs(request):

    nan_df1 = DataFrame(np.random.rand(10, 5))
    nan_df1[nan_df1 > 0.5] = np.nan

    opts = (
        DataFrame(np.random.randn(10, 5)),
        Series(np.random.randn(5)),
        Series([1, 2, np.nan, np.nan, 5]),
        nan_df1,
        np.random.randn(),
    )
    return opts[request.param]

rhs = lhs
midhs = lhs

class TestEval:
    @pytest.mark.parametrize(
        "cmp1",
        ["!=", "==", "<=", ">=", "<", ">"],
        ids=["ne", "eq", "le", "ge", "lt", "gt"],
    )
    @pytest.mark.parametrize("cmp2", [">", "<"], ids=["gt", "lt"])
    @pytest.mark.parametrize("binop", expr.BOOL_OPS_SYMS)
    def test_complex_cmp_ops(self, cmp1, cmp2, binop, lhs, rhs, engine, parser):
        if parser == "python" and binop in ["and", "or"]:
            msg = "'BoolOp' nodes are not implemented"
            with pytest.raises(NotImplementedError, match=msg):
                ex = f"(lhs {cmp1} rhs) {binop} (lhs {cmp2} rhs)"
                pd.eval(ex, engine=engine, parser=parser)
            return

        lhs_new = _eval_single_bin(lhs, cmp1, rhs, engine)
        rhs_new = _eval_single_bin(lhs, cmp2, rhs, engine)
        expected = _eval_single_bin(lhs_new, binop, rhs_new, engine)

        ex = f"(lhs {cmp1} rhs) {binop} (lhs {cmp2} rhs)"
        result = pd.eval(ex, engine=engine, parser=parser)
        tm.assert_equal(result, expected)

    @pytest.mark.parametrize("cmp_op", expr.CMP_OPS_SYMS)
    def test_simple_cmp_ops(self, cmp_op, lhs, rhs, engine, parser):
        lhs = lhs < 0
        rhs = rhs < 0

        if parser == "python" and cmp_op in ["in", "not in"]:
            msg = "'(In|NotIn)' nodes are not implemented"

            with pytest.raises(NotImplementedError, match=msg):
                ex = f"lhs {cmp_op} rhs"
                pd.eval(ex, engine=engine, parser=parser)
            return

        ex = f"lhs {cmp_op} rhs"
        msg = "|".join(
            [
                r"only list-like( or dict-like)? objects are allowed to be "
                r"passed to (DataFrame\.)?isin\(\), you passed a "
                r"(\[|')bool(\]|')",
                "argument of type 'bool' is not iterable",
            ]
        )
        if cmp_op in ("in", "not in") and not is_list_like(rhs):
            with pytest.raises(TypeError, match=msg):
                pd.eval(
                    ex,
                    engine=engine,
                    parser=parser,
                    local_dict={"lhs": lhs, "rhs": rhs},
                )
        else:
            expected = _eval_single_bin(lhs, cmp_op, rhs, engine)
            result = pd.eval(ex, engine=engine, parser=parser)
            tm.assert_equal(result, expected)

    @pytest.mark.parametrize("op", expr.CMP_OPS_SYMS)
    def test_compound_invert_op(self, op, lhs, rhs, request, engine, parser):
        if parser == "python" and op in ["in", "not in"]:

            msg = "'(In|NotIn)' nodes are not implemented"
            with pytest.raises(NotImplementedError, match=msg):
                ex = f"~(lhs {op} rhs)"
                pd.eval(ex, engine=engine, parser=parser)
            return

        if (
            is_float(lhs)
            and not is_float(rhs)
            and op in ["in", "not in"]
            and engine == "python"
            and parser == "pandas"
        ):
            mark = pytest.mark.xfail(
                reason="Looks like expected is negative, unclear whether "
                "expected is incorrect or result is incorrect"
            )
            request.node.add_marker(mark)
        skip_these = ["in", "not in"]
        ex = f"~(lhs {op} rhs)"

        msg = "|".join(
            [
                r"only list-like( or dict-like)? objects are allowed to be "
                r"passed to (DataFrame\.)?isin\(\), you passed a "
                r"(\[|')float(\]|')",
                "argument of type 'float' is not iterable",
            ]
        )
        if is_scalar(rhs) and op in skip_these:
            with pytest.raises(TypeError, match=msg):
                pd.eval(
                    ex,
                    engine=engine,
                    parser=parser,
                    local_dict={"lhs": lhs, "rhs": rhs},
                )
        else:
            # compound
            if is_scalar(lhs) and is_scalar(rhs):
                lhs, rhs = map(lambda x: np.array([x]), (lhs, rhs))
            expected = _eval_single_bin(lhs, op, rhs, engine)
            if is_scalar(expected):
                expected = not expected
            else:
                expected = ~expected
            result = pd.eval(ex, engine=engine, parser=parser)
            tm.assert_almost_equal(expected, result)

    @pytest.mark.parametrize("cmp1", ["<", ">"])
    @pytest.mark.parametrize("cmp2", ["<", ">"])
    def test_chained_cmp_op(self, cmp1, cmp2, lhs, midhs, rhs, engine, parser):
        mid = midhs
        if parser == "python":
            ex1 = f"lhs {cmp1} mid {cmp2} rhs"
            msg = "'BoolOp' nodes are not implemented"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(ex1, engine=engine, parser=parser)
            return

        lhs_new = _eval_single_bin(lhs, cmp1, mid, engine)
        rhs_new = _eval_single_bin(mid, cmp2, rhs, engine)

        if lhs_new is not None and rhs_new is not None:
            ex1 = f"lhs {cmp1} mid {cmp2} rhs"
            ex2 = f"lhs {cmp1} mid and mid {cmp2} rhs"
            ex3 = f"(lhs {cmp1} mid) & (mid {cmp2} rhs)"
            expected = _eval_single_bin(lhs_new, "&", rhs_new, engine)

            for ex in (ex1, ex2, ex3):
                result = pd.eval(ex, engine=engine, parser=parser)

                tm.assert_almost_equal(result, expected)

    @pytest.mark.parametrize(
        "arith1", sorted(set(ARITH_OPS_SYMS).difference(SPECIAL_CASE_ARITH_OPS_SYMS))
    )
    def test_binary_arith_ops(self, arith1, lhs, rhs, engine, parser):
        ex = f"lhs {arith1} rhs"
        result = pd.eval(ex, engine=engine, parser=parser)
        expected = _eval_single_bin(lhs, arith1, rhs, engine)

        tm.assert_almost_equal(result, expected)
        ex = f"lhs {arith1} rhs {arith1} rhs"
        result = pd.eval(ex, engine=engine, parser=parser)
        nlhs = _eval_single_bin(lhs, arith1, rhs, engine)
        try:
            nlhs, ghs = nlhs.align(rhs)
        except (ValueError, TypeError, AttributeError):
            # ValueError: series frame or frame series align
            # TypeError, AttributeError: series or frame with scalar align
            return
        else:
            if engine == "numexpr":
                import numexpr as ne

                # direct numpy comparison
                expected = ne.evaluate(f"nlhs {arith1} ghs")
                # Update assert statement due to unreliable numerical
                # precision component (GH37328)
                # TODO: update testing code so that assert_almost_equal statement
                #  can be replaced again by the assert_numpy_array_equal statement
                tm.assert_almost_equal(result.values, expected)
            else:
                expected = eval(f"nlhs {arith1} ghs")
                tm.assert_almost_equal(result, expected)

    # modulus, pow, and floor division require special casing

    def test_modulus(self, lhs, rhs, engine, parser):
        ex = r"lhs % rhs"
        result = pd.eval(ex, engine=engine, parser=parser)
        expected = lhs % rhs
        tm.assert_almost_equal(result, expected)

        if engine == "numexpr":
            import numexpr as ne

            expected = ne.evaluate(r"expected % rhs")
            if isinstance(result, (DataFrame, Series)):
                tm.assert_almost_equal(result.values, expected)
            else:
                tm.assert_almost_equal(result, expected.item())
        else:
            expected = _eval_single_bin(expected, "%", rhs, engine)
            tm.assert_almost_equal(result, expected)

    def test_floor_division(self, lhs, rhs, engine, parser):
        ex = "lhs // rhs"

        if engine == "python":
            res = pd.eval(ex, engine=engine, parser=parser)
            expected = lhs // rhs
            tm.assert_equal(res, expected)
        else:
            msg = (
                r"unsupported operand type\(s\) for //: 'VariableNode' and "
                "'VariableNode'"
            )
            with pytest.raises(TypeError, match=msg):
                pd.eval(
                    ex,
                    local_dict={"lhs": lhs, "rhs": rhs},
                    engine=engine,
                    parser=parser,
                )

    @td.skip_if_windows
    def test_pow(self, lhs, rhs, engine, parser):
        # odd failure on win32 platform, so skip
        ex = "lhs ** rhs"
        expected = _eval_single_bin(lhs, "**", rhs, engine)
        result = pd.eval(ex, engine=engine, parser=parser)

        if (
            is_scalar(lhs)
            and is_scalar(rhs)
            and isinstance(expected, (complex, np.complexfloating))
            and np.isnan(result)
        ):
            msg = "(DataFrame.columns|numpy array) are different"
            with pytest.raises(AssertionError, match=msg):
                tm.assert_numpy_array_equal(result, expected)
        else:
            tm.assert_almost_equal(result, expected)

            ex = "(lhs ** rhs) ** rhs"
            result = pd.eval(ex, engine=engine, parser=parser)

            middle = _eval_single_bin(lhs, "**", rhs, engine)
            expected = _eval_single_bin(middle, "**", rhs, engine)
            tm.assert_almost_equal(result, expected)

    def check_single_invert_op(self, lhs, engine, parser):
        # simple
        try:
            elb = lhs.astype(bool)
        except AttributeError:
            elb = np.array([bool(lhs)])
        expected = ~elb
        result = pd.eval("~elb", engine=engine, parser=parser)
        tm.assert_almost_equal(expected, result)

    def test_frame_invert(self, engine, parser):
        expr = "~lhs"

        # ~ ##
        # frame
        # float always raises
        lhs = DataFrame(np.random.randn(5, 2))
        if engine == "numexpr":
            msg = "couldn't find matching opcode for 'invert_dd'"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)
        else:
            msg = "ufunc 'invert' not supported for the input types"
            with pytest.raises(TypeError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)

        # int raises on numexpr
        lhs = DataFrame(np.random.randint(5, size=(5, 2)))
        if engine == "numexpr":
            msg = "couldn't find matching opcode for 'invert"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)
        else:
            expect = ~lhs
            result = pd.eval(expr, engine=engine, parser=parser)
            tm.assert_frame_equal(expect, result)

        # bool always works
        lhs = DataFrame(np.random.rand(5, 2) > 0.5)
        expect = ~lhs
        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_frame_equal(expect, result)

        # object raises
        lhs = DataFrame({"b": ["a", 1, 2.0], "c": np.random.rand(3) > 0.5})
        if engine == "numexpr":
            with pytest.raises(ValueError, match="unknown type object"):
                pd.eval(expr, engine=engine, parser=parser)
        else:
            msg = "bad operand type for unary ~: 'str'"
            with pytest.raises(TypeError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)

    def test_series_invert(self, engine, parser):
        # ~ ####
        expr = "~lhs"

        # series
        # float raises
        lhs = Series(np.random.randn(5))
        if engine == "numexpr":
            msg = "couldn't find matching opcode for 'invert_dd'"
            with pytest.raises(NotImplementedError, match=msg):
                result = pd.eval(expr, engine=engine, parser=parser)
        else:
            msg = "ufunc 'invert' not supported for the input types"
            with pytest.raises(TypeError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)

        # int raises on numexpr
        lhs = Series(np.random.randint(5, size=5))
        if engine == "numexpr":
            msg = "couldn't find matching opcode for 'invert"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)
        else:
            expect = ~lhs
            result = pd.eval(expr, engine=engine, parser=parser)
            tm.assert_series_equal(expect, result)

        # bool
        lhs = Series(np.random.rand(5) > 0.5)
        expect = ~lhs
        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_series_equal(expect, result)

        # float
        # int
        # bool

        # object
        lhs = Series(["a", 1, 2.0])
        if engine == "numexpr":
            with pytest.raises(ValueError, match="unknown type object"):
                pd.eval(expr, engine=engine, parser=parser)
        else:
            msg = "bad operand type for unary ~: 'str'"
            with pytest.raises(TypeError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)

    def test_frame_negate(self, engine, parser):
        expr = "-lhs"

        # float
        lhs = DataFrame(np.random.randn(5, 2))
        expect = -lhs
        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_frame_equal(expect, result)

        # int
        lhs = DataFrame(np.random.randint(5, size=(5, 2)))
        expect = -lhs
        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_frame_equal(expect, result)

        # bool doesn't work with numexpr but works elsewhere
        lhs = DataFrame(np.random.rand(5, 2) > 0.5)
        if engine == "numexpr":
            msg = "couldn't find matching opcode for 'neg_bb'"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)
        else:
            expect = -lhs
            result = pd.eval(expr, engine=engine, parser=parser)
            tm.assert_frame_equal(expect, result)

    def test_series_negate(self, engine, parser):
        expr = "-lhs"

        # float
        lhs = Series(np.random.randn(5))
        expect = -lhs
        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_series_equal(expect, result)

        # int
        lhs = Series(np.random.randint(5, size=5))
        expect = -lhs
        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_series_equal(expect, result)

        # bool doesn't work with numexpr but works elsewhere
        lhs = Series(np.random.rand(5) > 0.5)
        if engine == "numexpr":
            msg = "couldn't find matching opcode for 'neg_bb'"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(expr, engine=engine, parser=parser)
        else:
            expect = -lhs
            result = pd.eval(expr, engine=engine, parser=parser)
            tm.assert_series_equal(expect, result)

    @pytest.mark.parametrize(
        "lhs",
        [
            # Float
            DataFrame(np.random.randn(5, 2)),
            # Int
            DataFrame(np.random.randint(5, size=(5, 2))),
            # bool doesn't work with numexpr but works elsewhere
            DataFrame(np.random.rand(5, 2) > 0.5),
        ],
    )
    def test_frame_pos(self, lhs, engine, parser):
        expr = "+lhs"
        expect = lhs

        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_frame_equal(expect, result)

    @pytest.mark.parametrize(
        "lhs",
        [
            # Float
            Series(np.random.randn(5)),
            # Int
            Series(np.random.randint(5, size=5)),
            # bool doesn't work with numexpr but works elsewhere
            Series(np.random.rand(5) > 0.5),
        ],
    )
    def test_series_pos(self, lhs, engine, parser):
        expr = "+lhs"
        expect = lhs

        result = pd.eval(expr, engine=engine, parser=parser)
        tm.assert_series_equal(expect, result)

    def test_scalar_unary(self, engine, parser):
        msg = "bad operand type for unary ~: 'float'"
        with pytest.raises(TypeError, match=msg):
            pd.eval("~1.0", engine=engine, parser=parser)

        assert pd.eval("-1.0", parser=parser, engine=engine) == -1.0
        assert pd.eval("+1.0", parser=parser, engine=engine) == +1.0
        assert pd.eval("~1", parser=parser, engine=engine) == ~1
        assert pd.eval("-1", parser=parser, engine=engine) == -1
        assert pd.eval("+1", parser=parser, engine=engine) == +1
        assert pd.eval("~True", parser=parser, engine=engine) == ~True
        assert pd.eval("~False", parser=parser, engine=engine) == ~False
        assert pd.eval("-True", parser=parser, engine=engine) == -True
        assert pd.eval("-False", parser=parser, engine=engine) == -False
        assert pd.eval("+True", parser=parser, engine=engine) == +True
        assert pd.eval("+False", parser=parser, engine=engine) == +False

    def test_unary_in_array(self):
        # GH 11235
        # TODO: 2022-01-29: result return list with numexpr 2.7.3 in CI
        # but cannot reproduce locally
        result = np.array(
            pd.eval(
                "[-True, True, ~True, +True,"
                "-False, False, ~False, +False,"
                "-37, 37, ~37, +37]"
            ),
            dtype=np.object_,
        )
        expected = np.array(
            [
                -True,
                True,
                ~True,
                +True,
                -False,
                False,
                ~False,
                +False,
                -37,
                37,
                ~37,
                +37,
            ],
            dtype=np.object_,
        )
        tm.assert_numpy_array_equal(result, expected)

    @pytest.mark.parametrize("dtype", [np.float32, np.float64])
    @pytest.mark.parametrize("expr", ["x < -0.1", "-5 > x"])
    def test_float_comparison_bin_op(self, dtype, expr):
        # GH 16363
        df = DataFrame({"x": np.array([0], dtype=dtype)})
        res = df.eval(expr)
        assert res.values == np.array([False])

    @pytest.mark.parametrize(
        "ex",
        (
            "1 or 2",
            "1 and 2",
            "a and b",
            "a or b",
            "1 or 2 and (3 + 2) > 3",
            "2 * x > 2 or 1 and 2",
            "2 * df > 3 and 1 or a",
        ),
    )
    def test_disallow_scalar_bool_ops(self, ex, engine, parser):
        x, a, b = np.random.randn(3), 1, 2  # noqa:F841
        df = DataFrame(np.random.randn(3, 2))  # noqa:F841

        msg = "cannot evaluate scalar only bool ops|'BoolOp' nodes are not"
        with pytest.raises(NotImplementedError, match=msg):
            pd.eval(ex, engine=engine, parser=parser)

    def test_identical(self, engine, parser):
        # see gh-10546
        x = 1
        result = pd.eval("x", engine=engine, parser=parser)
        assert result == 1
        assert is_scalar(result)

        x = 1.5
        result = pd.eval("x", engine=engine, parser=parser)
        assert result == 1.5
        assert is_scalar(result)

        x = False
        result = pd.eval("x", engine=engine, parser=parser)
        assert not result
        assert is_bool(result)
        assert is_scalar(result)

        x = np.array([1])
        result = pd.eval("x", engine=engine, parser=parser)
        tm.assert_numpy_array_equal(result, np.array([1]))
        assert result.shape == (1,)

        x = np.array([1.5])
        result = pd.eval("x", engine=engine, parser=parser)
        tm.assert_numpy_array_equal(result, np.array([1.5]))
        assert result.shape == (1,)

        x = np.array([False])  # noqa:F841
        result = pd.eval("x", engine=engine, parser=parser)
        tm.assert_numpy_array_equal(result, np.array([False]))
        assert result.shape == (1,)

    def test_line_continuation(self, engine, parser):
        # GH 11149
        exp = """1 + 2 * \
        5 - 1 + 2 """
        result = pd.eval(exp, engine=engine, parser=parser)
        assert result == 12

    def test_float_truncation(self, engine, parser):
        # GH 14241
        exp = "1000000000.006"
        result = pd.eval(exp, engine=engine, parser=parser)
        expected = np.float64(exp)
        assert result == expected

        df = DataFrame({"A": [1000000000.0009, 1000000000.0011, 1000000000.0015]})
        cutoff = 1000000000.0006
        result = df.query(f"A < {cutoff:.4f}")
        assert result.empty

        cutoff = 1000000000.0010
        result = df.query(f"A > {cutoff:.4f}")
        expected = df.loc[[1, 2], :]
        tm.assert_frame_equal(expected, result)

        exact = 1000000000.0011
        result = df.query(f"A == {exact:.4f}")
        expected = df.loc[[1], :]
        tm.assert_frame_equal(expected, result)

    def test_disallow_python_keywords(self):
        # GH 18221
        df = DataFrame([[0, 0, 0]], columns=["foo", "bar", "class"])
        msg = "Python keyword not valid identifier in numexpr query"
        with pytest.raises(SyntaxError, match=msg):
            df.query("class == 0")

        df = DataFrame()
        df.index.name = "lambda"
        with pytest.raises(SyntaxError, match=msg):
            df.query("lambda == 0")

    def test_true_false_logic(self):
        # GH 25823
        assert pd.eval("not True") == -2
        assert pd.eval("not False") == -1
        assert pd.eval("True and not True") == 0

    def test_and_logic_string_match(self):
        # GH 25823
        event = Series({"a": "hello"})
        assert pd.eval(f"{event.str.match('hello').a}")
        assert pd.eval(f"{event.str.match('hello').a and event.str.match('hello').a}")

f = lambda *args, **kwargs: np.random.randn()

# -------------------------------------
# gh-12388: Typecasting rules consistency with python

class TestTypeCasting:
    @pytest.mark.parametrize("op", ["+", "-", "*", "**", "/"])
    # maybe someday... numexpr has too many upcasting rules now
    # chain(*(np.sctypes[x] for x in ['uint', 'int', 'float']))
    @pytest.mark.parametrize("dt", [np.float32, np.float64])
    @pytest.mark.parametrize("left_right", [("df", "3"), ("3", "df")])
    def test_binop_typecasting(self, engine, parser, op, dt, left_right):
        df = tm.makeCustomDataframe(5, 3, data_gen_f=f, dtype=dt)
        left, right = left_right
        s = f"{left} {op} {right}"
        res = pd.eval(s, engine=engine, parser=parser)
        assert df.values.dtype == dt
        assert res.values.dtype == dt
        tm.assert_frame_equal(res, eval(s))

# -------------------------------------
# Basic and complex alignment

def should_warn(*args):
    not_mono = not any(map(operator.attrgetter("is_monotonic_increasing"), args))
    only_one_dt = reduce(
        operator.xor, map(lambda x: issubclass(x.dtype.type, np.datetime64), args)
    )
    return not_mono and only_one_dt

class TestAlignment:

    index_types = ["i", "s", "dt"]
    lhs_index_types = index_types + ["s"]  # 'p'

    def test_align_nested_unary_op(self, engine, parser):
        s = "df * ~2"
        df = tm.makeCustomDataframe(5, 3, data_gen_f=f)
        res = pd.eval(s, engine=engine, parser=parser)
        tm.assert_frame_equal(res, df * ~2)

    @pytest.mark.parametrize("lr_idx_type", lhs_index_types)
    @pytest.mark.parametrize("rr_idx_type", index_types)
    @pytest.mark.parametrize("c_idx_type", index_types)
    def test_basic_frame_alignment(
        self, engine, parser, lr_idx_type, rr_idx_type, c_idx_type
    ):
        with warnings.catch_warnings(record=True):
            warnings.simplefilter("always", RuntimeWarning)

            df = tm.makeCustomDataframe(
                10, 10, data_gen_f=f, r_idx_type=lr_idx_type, c_idx_type=c_idx_type
            )
            df2 = tm.makeCustomDataframe(
                20, 10, data_gen_f=f, r_idx_type=rr_idx_type, c_idx_type=c_idx_type
            )
            # only warns if not monotonic and not sortable
            if should_warn(df.index, df2.index):
                with tm.assert_produces_warning(RuntimeWarning):
                    res = pd.eval("df + df2", engine=engine, parser=parser)
            else:
                res = pd.eval("df + df2", engine=engine, parser=parser)
            tm.assert_frame_equal(res, df + df2)

    @pytest.mark.parametrize("r_idx_type", lhs_index_types)
    @pytest.mark.parametrize("c_idx_type", lhs_index_types)
    def test_frame_comparison(self, engine, parser, r_idx_type, c_idx_type):
        df = tm.makeCustomDataframe(
            10, 10, data_gen_f=f, r_idx_type=r_idx_type, c_idx_type=c_idx_type
        )
        res = pd.eval("df < 2", engine=engine, parser=parser)
        tm.assert_frame_equal(res, df < 2)

        df3 = DataFrame(np.random.randn(*df.shape), index=df.index, columns=df.columns)
        res = pd.eval("df < df3", engine=engine, parser=parser)
        tm.assert_frame_equal(res, df < df3)

    @pytest.mark.parametrize("r1", lhs_index_types)
    @pytest.mark.parametrize("c1", index_types)
    @pytest.mark.parametrize("r2", index_types)
    @pytest.mark.parametrize("c2", index_types)
    def test_medium_complex_frame_alignment(self, engine, parser, r1, c1, r2, c2):

        with warnings.catch_warnings(record=True):
            warnings.simplefilter("always", RuntimeWarning)

            df = tm.makeCustomDataframe(
                3, 2, data_gen_f=f, r_idx_type=r1, c_idx_type=c1
            )
            df2 = tm.makeCustomDataframe(
                4, 2, data_gen_f=f, r_idx_type=r2, c_idx_type=c2
            )
            df3 = tm.makeCustomDataframe(
                5, 2, data_gen_f=f, r_idx_type=r2, c_idx_type=c2
            )
            if should_warn(df.index, df2.index, df3.index):
                with tm.assert_produces_warning(RuntimeWarning):
                    res = pd.eval("df + df2 + df3", engine=engine, parser=parser)
            else:
                res = pd.eval("df + df2 + df3", engine=engine, parser=parser)
            tm.assert_frame_equal(res, df + df2 + df3)

    @pytest.mark.parametrize("index_name", ["index", "columns"])
    @pytest.mark.parametrize("c_idx_type", index_types)
    @pytest.mark.parametrize("r_idx_type", lhs_index_types)
    def test_basic_frame_series_alignment(
        self, engine, parser, index_name, r_idx_type, c_idx_type
    ):
        with warnings.catch_warnings(record=True):
            warnings.simplefilter("always", RuntimeWarning)
            df = tm.makeCustomDataframe(
                10, 10, data_gen_f=f, r_idx_type=r_idx_type, c_idx_type=c_idx_type
            )
            index = getattr(df, index_name)
            s = Series(np.random.randn(5), index[:5])

            if should_warn(df.index, s.index):
                with tm.assert_produces_warning(RuntimeWarning):
                    res = pd.eval("df + s", engine=engine, parser=parser)
            else:
                res = pd.eval("df + s", engine=engine, parser=parser)

            if r_idx_type == "dt" or c_idx_type == "dt":
                expected = df.add(s) if engine == "numexpr" else df + s
            else:
                expected = df + s
            tm.assert_frame_equal(res, expected)

    @pytest.mark.parametrize("index_name", ["index", "columns"])
    @pytest.mark.parametrize(
        "r_idx_type, c_idx_type",
        list(product(["i", "s"], ["i", "s"])) + [("dt", "dt")],
    )
    @pytest.mark.filterwarnings("ignore::RuntimeWarning")
    def test_basic_series_frame_alignment(
        self, request, engine, parser, index_name, r_idx_type, c_idx_type
    ):
        if (
            engine == "numexpr"
            and parser == "pandas"
            and index_name == "index"
            and r_idx_type == "i"
            and c_idx_type == "s"
        ):
            reason = (
                f"Flaky column ordering when engine={engine}, "
                f"parser={parser}, index_name={index_name}, "
                f"r_idx_type={r_idx_type}, c_idx_type={c_idx_type}"
            )
            request.node.add_marker(pytest.mark.xfail(reason=reason, strict=False))
        df = tm.makeCustomDataframe(
            10, 7, data_gen_f=f, r_idx_type=r_idx_type, c_idx_type=c_idx_type
        )
        index = getattr(df, index_name)
        s = Series(np.random.randn(5), index[:5])
        if should_warn(s.index, df.index):
            with tm.assert_produces_warning(RuntimeWarning):
                res = pd.eval("s + df", engine=engine, parser=parser)
        else:
            res = pd.eval("s + df", engine=engine, parser=parser)

        if r_idx_type == "dt" or c_idx_type == "dt":
            expected = df.add(s) if engine == "numexpr" else s + df
        else:
            expected = s + df
        tm.assert_frame_equal(res, expected)

    @pytest.mark.parametrize("c_idx_type", index_types)
    @pytest.mark.parametrize("r_idx_type", lhs_index_types)
    @pytest.mark.parametrize("index_name", ["index", "columns"])
    @pytest.mark.parametrize("op", ["+", "*"])
    def test_series_frame_commutativity(
        self, engine, parser, index_name, op, r_idx_type, c_idx_type
    ):

        with warnings.catch_warnings(record=True):
            warnings.simplefilter("always", RuntimeWarning)

            df = tm.makeCustomDataframe(
                10, 10, data_gen_f=f, r_idx_type=r_idx_type, c_idx_type=c_idx_type
            )
            index = getattr(df, index_name)
            s = Series(np.random.randn(5), index[:5])

            lhs = f"s {op} df"
            rhs = f"df {op} s"
            if should_warn(df.index, s.index):
                with tm.assert_produces_warning(RuntimeWarning):
                    a = pd.eval(lhs, engine=engine, parser=parser)
                with tm.assert_produces_warning(RuntimeWarning):
                    b = pd.eval(rhs, engine=engine, parser=parser)
            else:
                a = pd.eval(lhs, engine=engine, parser=parser)
                b = pd.eval(rhs, engine=engine, parser=parser)

            if r_idx_type != "dt" and c_idx_type != "dt":
                if engine == "numexpr":
                    tm.assert_frame_equal(a, b)

    @pytest.mark.parametrize("r1", lhs_index_types)
    @pytest.mark.parametrize("c1", index_types)
    @pytest.mark.parametrize("r2", index_types)
    @pytest.mark.parametrize("c2", index_types)
    def test_complex_series_frame_alignment(self, engine, parser, r1, c1, r2, c2):
        n = 3
        m1 = 5
        m2 = 2 * m1

        with warnings.catch_warnings(record=True):
            warnings.simplefilter("always", RuntimeWarning)

            index_name = random.choice(["index", "columns"])
            obj_name = random.choice(["df", "df2"])

            df = tm.makeCustomDataframe(
                m1, n, data_gen_f=f, r_idx_type=r1, c_idx_type=c1
            )
            df2 = tm.makeCustomDataframe(
                m2, n, data_gen_f=f, r_idx_type=r2, c_idx_type=c2
            )
            index = getattr(locals().get(obj_name), index_name)
            ser = Series(np.random.randn(n), index[:n])

            if r2 == "dt" or c2 == "dt":
                if engine == "numexpr":
                    expected2 = df2.add(ser)
                else:
                    expected2 = df2 + ser
            else:
                expected2 = df2 + ser

            if r1 == "dt" or c1 == "dt":
                if engine == "numexpr":
                    expected = expected2.add(df)
                else:
                    expected = expected2 + df
            else:
                expected = expected2 + df

            if should_warn(df2.index, ser.index, df.index):
                with tm.assert_produces_warning(RuntimeWarning):
                    res = pd.eval("df2 + ser + df", engine=engine, parser=parser)
            else:
                res = pd.eval("df2 + ser + df", engine=engine, parser=parser)
            assert res.shape == expected.shape
            tm.assert_frame_equal(res, expected)

    def test_performance_warning_for_poor_alignment(self, engine, parser):
        df = DataFrame(np.random.randn(1000, 10))
        s = Series(np.random.randn(10000))
        if engine == "numexpr":
            seen = PerformanceWarning
        else:
            seen = False

        with tm.assert_produces_warning(seen):
            pd.eval("df + s", engine=engine, parser=parser)

        s = Series(np.random.randn(1000))
        with tm.assert_produces_warning(False):
            pd.eval("df + s", engine=engine, parser=parser)

        df = DataFrame(np.random.randn(10, 10000))
        s = Series(np.random.randn(10000))
        with tm.assert_produces_warning(False):
            pd.eval("df + s", engine=engine, parser=parser)

        df = DataFrame(np.random.randn(10, 10))
        s = Series(np.random.randn(10000))

        is_python_engine = engine == "python"

        if not is_python_engine:
            wrn = PerformanceWarning
        else:
            wrn = False

        with tm.assert_produces_warning(wrn) as w:
            pd.eval("df + s", engine=engine, parser=parser)

            if not is_python_engine:
                assert len(w) == 1
                msg = str(w[0].message)
                logged = np.log10(s.size - df.shape[1])
                expected = (
                    f"Alignment difference on axis 1 is larger "
                    f"than an order of magnitude on term 'df', "
                    f"by more than {logged:.4g}; performance may suffer."
                )
                assert msg == expected

# ------------------------------------
# Slightly more complex ops

class TestOperations:
    def eval(self, *args, **kwargs):
        kwargs["level"] = kwargs.pop("level", 0) + 1
        return pd.eval(*args, **kwargs)

    def test_simple_arith_ops(self, engine, parser):
        exclude_arith = []
        if parser == "python":
            exclude_arith = ["in", "not in"]

        arith_ops = [
            op
            for op in expr.ARITH_OPS_SYMS + expr.CMP_OPS_SYMS
            if op not in exclude_arith
        ]

        ops = (op for op in arith_ops if op != "//")

        for op in ops:
            ex = f"1 {op} 1"
            ex2 = f"x {op} 1"
            ex3 = f"1 {op} (x + 1)"

            if op in ("in", "not in"):
                msg = "argument of type 'int' is not iterable"
                with pytest.raises(TypeError, match=msg):
                    pd.eval(ex, engine=engine, parser=parser)
            else:
                expec = _eval_single_bin(1, op, 1, engine)
                x = self.eval(ex, engine=engine, parser=parser)
                assert x == expec

                expec = _eval_single_bin(x, op, 1, engine)
                y = self.eval(ex2, local_dict={"x": x}, engine=engine, parser=parser)
                assert y == expec

                expec = _eval_single_bin(1, op, x + 1, engine)
                y = self.eval(ex3, local_dict={"x": x}, engine=engine, parser=parser)
                assert y == expec

    @pytest.mark.parametrize("rhs", [True, False])
    @pytest.mark.parametrize("lhs", [True, False])
    @pytest.mark.parametrize("op", expr.BOOL_OPS_SYMS)
    def test_simple_bool_ops(self, rhs, lhs, op):
        ex = f"{lhs} {op} {rhs}"

        if parser == "python" and op in ["and", "or"]:
            msg = "'BoolOp' nodes are not implemented"
            with pytest.raises(NotImplementedError, match=msg):
                self.eval(ex)
            return

        res = self.eval(ex)
        exp = eval(ex)
        assert res == exp

    @pytest.mark.parametrize("rhs", [True, False])
    @pytest.mark.parametrize("lhs", [True, False])
    @pytest.mark.parametrize("op", expr.BOOL_OPS_SYMS)
    def test_bool_ops_with_constants(self, rhs, lhs, op):
        ex = f"{lhs} {op} {rhs}"

        if parser == "python" and op in ["and", "or"]:
            msg = "'BoolOp' nodes are not implemented"
            with pytest.raises(NotImplementedError, match=msg):
                self.eval(ex)
            return

        res = self.eval(ex)
        exp = eval(ex)
        assert res == exp

    def test_4d_ndarray_fails(self):
        x = np.random.randn(3, 4, 5, 6)
        y = Series(np.random.randn(10))
        msg = "N-dimensional objects, where N > 2, are not supported with eval"
        with pytest.raises(NotImplementedError, match=msg):
            self.eval("x + y", local_dict={"x": x, "y": y})

    def test_constant(self):
        x = self.eval("1")
        assert x == 1

    def test_single_variable(self):
        df = DataFrame(np.random.randn(10, 2))
        df2 = self.eval("df", local_dict={"df": df})
        tm.assert_frame_equal(df, df2)

    def test_truediv(self):
        s = np.array([1])  # noqa:F841
        ex = "s / 1"

        # FutureWarning: The `truediv` parameter in pd.eval is deprecated and will be
        # removed in a future version.
        with tm.assert_produces_warning(FutureWarning):
            res = self.eval(ex, truediv=False)
        tm.assert_numpy_array_equal(res, np.array([1.0]))

        with tm.assert_produces_warning(FutureWarning):
            res = self.eval(ex, truediv=True)
        tm.assert_numpy_array_equal(res, np.array([1.0]))

        with tm.assert_produces_warning(FutureWarning):
            res = self.eval("1 / 2", truediv=True)
        expec = 0.5
        assert res == expec

        with tm.assert_produces_warning(FutureWarning):
            res = self.eval("1 / 2", truediv=False)
        expec = 0.5
        assert res == expec

        with tm.assert_produces_warning(FutureWarning):
            res = self.eval("s / 2", truediv=False)
        expec = 0.5
        assert res == expec

        with tm.assert_produces_warning(FutureWarning):
            res = self.eval("s / 2", truediv=True)
        expec = 0.5
        assert res == expec

    def test_failing_subscript_with_name_error(self):
        df = DataFrame(np.random.randn(5, 3))  # noqa:F841
        with pytest.raises(NameError, match="name 'x' is not defined"):
            self.eval("df[x > 2] > 2")

    def test_lhs_expression_subscript(self):
        df = DataFrame(np.random.randn(5, 3))
        result = self.eval("(df + 1)[df > 2]", local_dict={"df": df})
        expected = (df + 1)[df > 2]
        tm.assert_frame_equal(result, expected)

    def test_attr_expression(self):
        df = DataFrame(np.random.randn(5, 3), columns=list("abc"))
        expr1 = "df.a < df.b"
        expec1 = df.a < df.b
        expr2 = "df.a + df.b + df.c"
        expec2 = df.a + df.b + df.c
        expr3 = "df.a + df.b + df.c[df.b < 0]"
        expec3 = df.a + df.b + df.c[df.b < 0]
        exprs = expr1, expr2, expr3
        expecs = expec1, expec2, expec3
        for e, expec in zip(exprs, expecs):
            tm.assert_series_equal(expec, self.eval(e, local_dict={"df": df}))

    def test_assignment_fails(self):
        df = DataFrame(np.random.randn(5, 3), columns=list("abc"))
        df2 = DataFrame(np.random.randn(5, 3))
        expr1 = "df = df2"
        msg = "cannot assign without a target object"
        with pytest.raises(ValueError, match=msg):
            self.eval(expr1, local_dict={"df": df, "df2": df2})

    def test_assignment_column_multiple_raise(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        # multiple assignees
        with pytest.raises(SyntaxError, match="invalid syntax"):
            df.eval("d c = a + b")

    def test_assignment_column_invalid_assign(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        # invalid assignees
        msg = "left hand side of an assignment must be a single name"
        with pytest.raises(SyntaxError, match=msg):
            df.eval("d,c = a + b")

    def test_assignment_column_invalid_assign_function_call(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        msg = "cannot assign to function call"
        with pytest.raises(SyntaxError, match=msg):
            df.eval('Timestamp("20131001") = a + b')

    def test_assignment_single_assign_existing(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        # single assignment - existing variable
        expected = df.copy()
        expected["a"] = expected["a"] + expected["b"]
        df.eval("a = a + b", inplace=True)
        tm.assert_frame_equal(df, expected)

    def test_assignment_single_assign_new(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        # single assignment - new variable
        expected = df.copy()
        expected["c"] = expected["a"] + expected["b"]
        df.eval("c = a + b", inplace=True)
        tm.assert_frame_equal(df, expected)

    def test_assignment_single_assign_local_overlap(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        df = df.copy()
        a = 1  # noqa:F841
        df.eval("a = 1 + b", inplace=True)

        expected = df.copy()
        expected["a"] = 1 + expected["b"]
        tm.assert_frame_equal(df, expected)

    def test_assignment_single_assign_name(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))

        a = 1  # noqa:F841
        old_a = df.a.copy()
        df.eval("a = a + b", inplace=True)
        result = old_a + df.b
        tm.assert_series_equal(result, df.a, check_names=False)
        assert result.name is None

    def test_assignment_multiple_raises(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        # multiple assignment
        df.eval("c = a + b", inplace=True)
        msg = "can only assign a single expression"
        with pytest.raises(SyntaxError, match=msg):
            df.eval("c = a = b")

    def test_assignment_explicit(self):
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))
        # explicit targets
        self.eval("c = df.a + df.b", local_dict={"df": df}, target=df, inplace=True)
        expected = df.copy()
        expected["c"] = expected["a"] + expected["b"]
        tm.assert_frame_equal(df, expected)

    def test_column_in(self):
        # GH 11235
        df = DataFrame({"a": [11], "b": [-32]})
        result = df.eval("a in [11, -32]")
        expected = Series([True])
        # TODO: 2022-01-29: Name check failed with numexpr 2.7.3 in CI
        # but cannot reproduce locally
        tm.assert_series_equal(result, expected, check_names=False)

    @pytest.mark.xfail(reason="Unknown: Omitted test_ in name prior.")
    def test_assignment_not_inplace(self):
        # see gh-9297
        df = DataFrame(np.random.randn(5, 2), columns=list("ab"))

        actual = df.eval("c = a + b", inplace=False)
        assert actual is not None

        expected = df.copy()
        expected["c"] = expected["a"] + expected["b"]
        tm.assert_frame_equal(df, expected)

    def test_multi_line_expression(self):
        # GH 11149
        df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
        expected = df.copy()

        expected["c"] = expected["a"] + expected["b"]
        expected["d"] = expected["c"] + expected["b"]
        answer = df.eval(
            """
        c = a + b
        d = c + b""",
            inplace=True,
        )
        tm.assert_frame_equal(expected, df)
        assert answer is None

        expected["a"] = expected["a"] - 1
        expected["e"] = expected["a"] + 2
        answer = df.eval(
            """
        a = a - 1
        e = a + 2""",
            inplace=True,
        )
        tm.assert_frame_equal(expected, df)
        assert answer is None

        # multi-line not valid if not all assignments
        msg = "Multi-line expressions are only valid if all expressions contain"
        with pytest.raises(ValueError, match=msg):
            df.eval(
                """
            a = b + 2
            b - 2""",
                inplace=False,
            )

    def test_multi_line_expression_not_inplace(self):
        # GH 11149
        df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
        expected = df.copy()

        expected["c"] = expected["a"] + expected["b"]
        expected["d"] = expected["c"] + expected["b"]
        df = df.eval(
            """
        c = a + b
        d = c + b""",
            inplace=False,
        )
        tm.assert_frame_equal(expected, df)

        expected["a"] = expected["a"] - 1
        expected["e"] = expected["a"] + 2
        df = df.eval(
            """
        a = a - 1
        e = a + 2""",
            inplace=False,
        )
        tm.assert_frame_equal(expected, df)

    def test_multi_line_expression_local_variable(self):
        # GH 15342
        df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
        expected = df.copy()

        local_var = 7
        expected["c"] = expected["a"] * local_var
        expected["d"] = expected["c"] + local_var
        answer = df.eval(
            """
        c = a * @local_var
        d = c + @local_var
        """,
            inplace=True,
        )
        tm.assert_frame_equal(expected, df)
        assert answer is None

    def test_multi_line_expression_callable_local_variable(self):
        # 26426
        df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})

        def local_func(a, b):
            return b

        expected = df.copy()
        expected["c"] = expected["a"] * local_func(1, 7)
        expected["d"] = expected["c"] + local_func(1, 7)
        answer = df.eval(
            """
        c = a * @local_func(1, 7)
        d = c + @local_func(1, 7)
        """,
            inplace=True,
        )
        tm.assert_frame_equal(expected, df)
        assert answer is None

    def test_multi_line_expression_callable_local_variable_with_kwargs(self):
        # 26426
        df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})

        def local_func(a, b):
            return b

        expected = df.copy()
        expected["c"] = expected["a"] * local_func(b=7, a=1)
        expected["d"] = expected["c"] + local_func(b=7, a=1)
        answer = df.eval(
            """
        c = a * @local_func(b=7, a=1)
        d = c + @local_func(b=7, a=1)
        """,
            inplace=True,
        )
        tm.assert_frame_equal(expected, df)
        assert answer is None

    def test_assignment_in_query(self):
        # GH 8664
        df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
        df_orig = df.copy()
        msg = "cannot assign without a target object"
        with pytest.raises(ValueError, match=msg):
            df.query("a = 1")
        tm.assert_frame_equal(df, df_orig)

    def test_query_inplace(self):
        # see gh-11149
        df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
        expected = df.copy()
        expected = expected[expected["a"] == 2]
        df.query("a == 2", inplace=True)
        tm.assert_frame_equal(expected, df)

        df = {}
        expected = {"a": 3}

        self.eval("a = 1 + 2", target=df, inplace=True)
        tm.assert_dict_equal(df, expected)

    @pytest.mark.parametrize("invalid_target", [1, "cat", [1, 2], np.array([]), (1, 3)])
    @pytest.mark.filterwarnings("ignore::FutureWarning")
    def test_cannot_item_assign(self, invalid_target):
        msg = "Cannot assign expression output to target"
        expression = "a = 1 + 2"

        with pytest.raises(ValueError, match=msg):
            self.eval(expression, target=invalid_target, inplace=True)

        if hasattr(invalid_target, "copy"):
            with pytest.raises(ValueError, match=msg):
                self.eval(expression, target=invalid_target, inplace=False)

    @pytest.mark.parametrize("invalid_target", [1, "cat", (1, 3)])
    def test_cannot_copy_item(self, invalid_target):
        msg = "Cannot return a copy of the target"
        expression = "a = 1 + 2"

        with pytest.raises(ValueError, match=msg):
            self.eval(expression, target=invalid_target, inplace=False)

    @pytest.mark.parametrize("target", [1, "cat", [1, 2], np.array([]), (1, 3), {1: 2}])
    def test_inplace_no_assignment(self, target):
        expression = "1 + 2"

        assert self.eval(expression, target=target, inplace=False) == 3

        msg = "Cannot operate inplace if there is no assignment"
        with pytest.raises(ValueError, match=msg):
            self.eval(expression, target=target, inplace=True)

    def test_basic_period_index_boolean_expression(self):
        df = tm.makeCustomDataframe(2, 2, data_gen_f=f, c_idx_type="p", r_idx_type="i")

        e = df < 2
        r = self.eval("df < 2", local_dict={"df": df})
        x = df < 2

        tm.assert_frame_equal(r, e)
        tm.assert_frame_equal(x, e)

    def test_basic_period_index_subscript_expression(self):
        df = tm.makeCustomDataframe(2, 2, data_gen_f=f, c_idx_type="p", r_idx_type="i")
        r = self.eval("df[df < 2 + 3]", local_dict={"df": df})
        e = df[df < 2 + 3]
        tm.assert_frame_equal(r, e)

    def test_nested_period_index_subscript_expression(self):
        df = tm.makeCustomDataframe(2, 2, data_gen_f=f, c_idx_type="p", r_idx_type="i")
        r = self.eval("df[df[df < 2] < 2] + df * 2", local_dict={"df": df})
        e = df[df[df < 2] < 2] + df * 2
        tm.assert_frame_equal(r, e)

    def test_date_boolean(self, engine, parser):
        df = DataFrame(np.random.randn(5, 3))
        df["dates1"] = date_range("1/1/2012", periods=5)
        res = self.eval(
            "df.dates1 < 20130101",
            local_dict={"df": df},
            engine=engine,
            parser=parser,
        )
        expec = df.dates1 < "20130101"
        tm.assert_series_equal(res, expec, check_names=False)

    def test_simple_in_ops(self, engine, parser):
        if parser != "python":
            res = pd.eval("1 in [1, 2]", engine=engine, parser=parser)
            assert res

            res = pd.eval("2 in (1, 2)", engine=engine, parser=parser)
            assert res

            res = pd.eval("3 in (1, 2)", engine=engine, parser=parser)
            assert not res

            res = pd.eval("3 not in (1, 2)", engine=engine, parser=parser)
            assert res

            res = pd.eval("[3] not in (1, 2)", engine=engine, parser=parser)
            assert res

            res = pd.eval("[3] in ([3], 2)", engine=engine, parser=parser)
            assert res

            res = pd.eval("[[3]] in [[[3]], 2]", engine=engine, parser=parser)
            assert res

            res = pd.eval("(3,) in [(3,), 2]", engine=engine, parser=parser)
            assert res

            res = pd.eval("(3,) not in [(3,), 2]", engine=engine, parser=parser)
            assert not res

            res = pd.eval("[(3,)] in [[(3,)], 2]", engine=engine, parser=parser)
            assert res
        else:
            msg = "'In' nodes are not implemented"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval("1 in [1, 2]", engine=engine, parser=parser)
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval("2 in (1, 2)", engine=engine, parser=parser)
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval("3 in (1, 2)", engine=engine, parser=parser)
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval("[(3,)] in (1, 2, [(3,)])", engine=engine, parser=parser)
            msg = "'NotIn' nodes are not implemented"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval("3 not in (1, 2)", engine=engine, parser=parser)
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval("[3] not in (1, 2, [[3]])", engine=engine, parser=parser)

    def test_check_many_exprs(self, engine, parser):
        a = 1  # noqa:F841
        expr = " * ".join("a" * 33)
        expected = 1
        res = pd.eval(expr, engine=engine, parser=parser)
        assert res == expected

    @pytest.mark.parametrize(
        "expr",
        [
            "df > 2 and df > 3",
            "df > 2 or df > 3",
            "not df > 2",
        ],
    )
    def test_fails_and_or_not(self, expr, engine, parser):
        df = DataFrame(np.random.randn(5, 3))
        if parser == "python":
            msg = "'BoolOp' nodes are not implemented"
            if "not" in expr:
                msg = "'Not' nodes are not implemented"

            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(
                    expr,
                    local_dict={"df": df},
                    parser=parser,
                    engine=engine,
                )
        else:
            # smoke-test, should not raise
            pd.eval(
                expr,
                local_dict={"df": df},
                parser=parser,
                engine=engine,
            )

    @pytest.mark.parametrize("char", ["|", "&"])
    def test_fails_ampersand_pipe(self, char, engine, parser):
        df = DataFrame(np.random.randn(5, 3))  # noqa:F841
        ex = f"(df + 2)[df > 1] > 0 {char} (df > 0)"
        if parser == "python":
            msg = "cannot evaluate scalar only bool ops"
            with pytest.raises(NotImplementedError, match=msg):
                pd.eval(ex, parser=parser, engine=engine)
        else:
            # smoke-test, should not raise
            pd.eval(ex, parser=parser, engine=engine)

class TestMath:
    def eval(self, *args, **kwargs):
        kwargs["level"] = kwargs.pop("level", 0) + 1
        return pd.eval(*args, **kwargs)

    def test_unary_functions(self, unary_fns_for_ne):
        df = DataFrame({"a": np.random.randn(10)})
        a = df.a

        fn = unary_fns_for_ne

        expr = f"{fn}(a)"
        got = self.eval(expr)
        with np.errstate(all="ignore"):
            expect = getattr(np, fn)(a)
        tm.assert_series_equal(got, expect, check_names=False)

    @pytest.mark.parametrize("fn", _binary_math_ops)
    def test_binary_functions(self, fn):
        df = DataFrame({"a": np.random.randn(10), "b": np.random.randn(10)})
        a = df.a
        b = df.b

        expr = f"{fn}(a, b)"
        got = self.eval(expr)
        with np.errstate(all="ignore"):
            expect = getattr(np, fn)(a, b)
        tm.assert_almost_equal(got, expect, check_names=False)

    def test_df_use_case(self, engine, parser):
        df = DataFrame({"a": np.random.randn(10), "b": np.random.randn(10)})
        df.eval(
            "e = arctan2(sin(a), b)",
            engine=engine,
            parser=parser,
            inplace=True,
        )
        got = df.e
        expect = np.arctan2(np.sin(df.a), df.b)
        tm.assert_series_equal(got, expect, check_names=False)

    def test_df_arithmetic_subexpression(self, engine, parser):
        df = DataFrame({"a": np.random.randn(10), "b": np.random.randn(10)})
        df.eval("e = sin(a + b)", engine=engine, parser=parser, inplace=True)
        got = df.e
        expect = np.sin(df.a + df.b)
        tm.assert_series_equal(got, expect, check_names=False)

    @pytest.mark.parametrize(
        "dtype, expect_dtype",
        [
            (np.int32, np.float64),
            (np.int64, np.float64),
            (np.float32, np.float32),
            (np.float64, np.float64),
            pytest.param(np.complex128, np.complex128, marks=td.skip_if_windows),
        ],
    )
    def test_result_types(self, dtype, expect_dtype, engine, parser):
        # xref https://github.com/pandas-dev/pandas/issues/12293
        #  this fails on Windows, apparently a floating point precision issue

        # Did not test complex64 because DataFrame is converting it to
        # complex128. Due to https://github.com/pandas-dev/pandas/issues/10952
        df = DataFrame({"a": np.random.randn(10).astype(dtype)})
        assert df.a.dtype == dtype
        df.eval("b = sin(a)", engine=engine, parser=parser, inplace=True)
        got = df.b
        expect = np.sin(df.a)
        assert expect.dtype == got.dtype
        assert expect_dtype == got.dtype
        tm.assert_series_equal(got, expect, check_names=False)

    def test_undefined_func(self, engine, parser):
        df = DataFrame({"a": np.random.randn(10)})
        msg = '"mysin" is not a supported function'

        with pytest.raises(ValueError, match=msg):
            df.eval("mysin(a)", engine=engine, parser=parser)

    def test_keyword_arg(self, engine, parser):
        df = DataFrame({"a": np.random.randn(10)})
        msg = 'Function "sin" does not support keyword arguments'

        with pytest.raises(TypeError, match=msg):
            df.eval("sin(x=a)", engine=engine, parser=parser)

_var_s = np.random.randn(10)

class TestScope:
    def test_global_scope(self, engine, parser):
        e = "_var_s * 2"
        tm.assert_numpy_array_equal(
            _var_s * 2, pd.eval(e, engine=engine, parser=parser)
        )

    def test_no_new_locals(self, engine, parser):
        x = 1
        lcls = locals().copy()
        pd.eval("x + 1", local_dict=lcls, engine=engine, parser=parser)
        lcls2 = locals().copy()
        lcls2.pop("lcls")
        assert lcls == lcls2

    def test_no_new_globals(self, engine, parser):
        x = 1  # noqa:F841
        gbls = globals().copy()
        pd.eval("x + 1", engine=engine, parser=parser)
        gbls2 = globals().copy()
        assert gbls == gbls2

    def test_empty_locals(self, engine, parser):
        # GH 47084
        x = 1  # noqa: F841
        msg = "name 'x' is not defined"
        with pytest.raises(UndefinedVariableError, match=msg):
            pd.eval("x + 1", engine=engine, parser=parser, local_dict={})

    def test_empty_globals(self, engine, parser):
        # GH 47084
        msg = "name '_var_s' is not defined"
        e = "_var_s * 2"
        with pytest.raises(UndefinedVariableError, match=msg):
            pd.eval(e, engine=engine, parser=parser, global_dict={})

@td.skip_if_no_ne
def test_invalid_engine():
    msg = "Invalid engine 'asdf' passed"
    with pytest.raises(KeyError, match=msg):
        pd.eval("x + y", local_dict={"x": 1, "y": 2}, engine="asdf")

@td.skip_if_no_ne
@pytest.mark.parametrize(
    ("use_numexpr", "expected"),
    (
        (True, "numexpr"),
        (False, "python"),
    ),
)
def test_numexpr_option_respected(use_numexpr, expected):
    # GH 32556
    from pandas.core.computation.eval import _check_engine

    with pd.option_context("compute.use_numexpr", use_numexpr):
        result = _check_engine(None)
        assert result == expected

@td.skip_if_no_ne
def test_numexpr_option_incompatible_op():
    # GH 32556
    with pd.option_context("compute.use_numexpr", False):
        df = DataFrame(
            {"A": [True, False, True, False, None, None], "B": [1, 2, 3, 4, 5, 6]}
        )
        result = df.query("A.isnull()")
        expected = DataFrame({"A": [None, None], "B": [5, 6]}, index=[4, 5])
        tm.assert_frame_equal(result, expected)

@td.skip_if_no_ne
def test_invalid_parser():
    msg = "Invalid parser 'asdf' passed"
    with pytest.raises(KeyError, match=msg):
        pd.eval("x + y", local_dict={"x": 1, "y": 2}, parser="asdf")

_parsers: dict[str, type[BaseExprVisitor]] = {
    "python": PythonExprVisitor,
    "pytables": pytables.PyTablesExprVisitor,
    "pandas": PandasExprVisitor,
}

@pytest.mark.parametrize("engine", ENGINES)
@pytest.mark.parametrize("parser", _parsers)
def test_disallowed_nodes(engine, parser):
    VisitorClass = _parsers[parser]
    inst = VisitorClass("x + 1", engine, parser)

    for ops in VisitorClass.unsupported_nodes:

        msg = "nodes are not implemented"
        with pytest.raises(NotImplementedError, match=msg):
            getattr(inst, ops)()

def test_syntax_error_exprs(engine, parser):
    e = "s +"
    with pytest.raises(SyntaxError, match="invalid syntax"):
        pd.eval(e, engine=engine, parser=parser)

def test_name_error_exprs(engine, parser):
    e = "s + t"
    msg = "name 's' is not defined"
    with pytest.raises(NameError, match=msg):
        pd.eval(e, engine=engine, parser=parser)

@pytest.mark.parametrize("express", ["a + @b", "@a + b", "@a + @b"])
def test_invalid_local_variable_reference(engine, parser, express):
    a, b = 1, 2  # noqa:F841

    if parser != "pandas":
        with pytest.raises(SyntaxError, match="The '@' prefix is only"):
            pd.eval(express, engine=engine, parser=parser)
    else:
        with pytest.raises(SyntaxError, match="The '@' prefix is not"):
            pd.eval(express, engine=engine, parser=parser)

def test_numexpr_builtin_raises(engine, parser):
    sin, dotted_line = 1, 2
    if engine == "numexpr":
        msg = "Variables in expression .+"
        with pytest.raises(NumExprClobberingError, match=msg):
            pd.eval("sin + dotted_line", engine=engine, parser=parser)
    else:
        res = pd.eval("sin + dotted_line", engine=engine, parser=parser)
        assert res == sin + dotted_line

def test_bad_resolver_raises(engine, parser):
    cannot_resolve = 42, 3.0
    with pytest.raises(TypeError, match="Resolver of type .+"):
        pd.eval("1 + 2", resolvers=cannot_resolve, engine=engine, parser=parser)

def test_empty_string_raises(engine, parser):
    # GH 13139
    with pytest.raises(ValueError, match="expr cannot be an empty string"):
        pd.eval("", engine=engine, parser=parser)

def test_more_than_one_expression_raises(engine, parser):
    with pytest.raises(SyntaxError, match="only a single expression is allowed"):
        pd.eval("1 + 1; 2 + 2", engine=engine, parser=parser)

@pytest.mark.parametrize("cmp", ("and", "or"))
@pytest.mark.parametrize("lhs", (int, float))
@pytest.mark.parametrize("rhs", (int, float))
def test_bool_ops_fails_on_scalars(lhs, cmp, rhs, engine, parser):
    gen = {int: lambda: np.random.randint(10), float: np.random.randn}

    mid = gen[lhs]()  # noqa:F841
    lhs = gen[lhs]()
    rhs = gen[rhs]()

    ex1 = f"lhs {cmp} mid {cmp} rhs"
    ex2 = f"lhs {cmp} mid and mid {cmp} rhs"
    ex3 = f"(lhs {cmp} mid) & (mid {cmp} rhs)"
    for ex in (ex1, ex2, ex3):
        msg = "cannot evaluate scalar only bool ops|'BoolOp' nodes are not"
        with pytest.raises(NotImplementedError, match=msg):
            pd.eval(ex, engine=engine, parser=parser)

@pytest.mark.parametrize(
    "other",
    [
        "'x'",
        "...",
    ],
)
def test_equals_various(other):
    df = DataFrame({"A": ["a", "b", "c"]})
    result = df.eval(f"A == {other}")
    expected = Series([False, False, False], name="A")
    if USE_NUMEXPR:
        # https://github.com/pandas-dev/pandas/issues/10239
        # lose name with numexpr engine. Remove when that's fixed.
        expected.name = None
    tm.assert_series_equal(result, expected)

def test_inf(engine, parser):
    s = "inf + 1"
    expected = np.inf
    result = pd.eval(s, engine=engine, parser=parser)
    assert result == expected

def test_truediv_deprecated(engine, parser):
    # GH#29182
    match = "The `truediv` parameter in pd.eval is deprecated"

    with tm.assert_produces_warning(FutureWarning) as m:
        pd.eval("1+1", engine=engine, parser=parser, truediv=True)

    assert len(m) == 1
    assert match in str(m[0].message)

    with tm.assert_produces_warning(FutureWarning) as m:
        pd.eval("1+1", engine=engine, parser=parser, truediv=False)

    assert len(m) == 1
    assert match in str(m[0].message)

@pytest.mark.parametrize("column", ["Temp(°C)", "Capacitance(μF)"])
def test_query_token(engine, column):
    # See: https://github.com/pandas-dev/pandas/pull/42826
    df = DataFrame(np.random.randn(5, 2), columns=[column, "b"])
    expected = df[df[column] > 5]
    query_string = f"`{column}` > 5"
    result = df.query(query_string, engine=engine)
    tm.assert_frame_equal(result, expected)

def test_negate_lt_eq_le(engine, parser):
    df = DataFrame([[0, 10], [1, 20]], columns=["cat", "count"])
    expected = df[~(df.cat > 0)]

    result = df.query("~(cat > 0)", engine=engine, parser=parser)
    tm.assert_frame_equal(result, expected)

    if parser == "python":
        msg = "'Not' nodes are not implemented"
        with pytest.raises(NotImplementedError, match=msg):
            df.query("not (cat > 0)", engine=engine, parser=parser)
    else:
        result = df.query("not (cat > 0)", engine=engine, parser=parser)
        tm.assert_frame_equal(result, expected)

@pytest.mark.parametrize(
    "column",
    DEFAULT_GLOBALS.keys(),
)
def test_eval_no_support_column_name(request, column):
    # GH 44603
    if column in ["True", "False", "inf", "Inf"]:
        request.node.add_marker(
            pytest.mark.xfail(
                raises=KeyError,
                reason=f"GH 47859 DataFrame eval not supported with {column}",
            )
        )

    df = DataFrame(np.random.randint(0, 100, size=(10, 2)), columns=[column, "col1"])
    expected = df[df[column] > 6]
    result = df.query(f"{column}>6")

    tm.assert_frame_equal(result, expected)

@td.skip_array_manager_not_yet_implemented
def test_set_inplace(using_copy_on_write):
    # https://github.com/pandas-dev/pandas/issues/47449
    # Ensure we don't only update the DataFrame inplace, but also the actual
    # column values, such that references to this column also get updated
    df = DataFrame({"A": [1, 2, 3], "B": [4, 5, 6], "C": [7, 8, 9]})
    result_view = df[:]
    ser = df["A"]
    df.eval("A = B + C", inplace=True)
    expected = DataFrame({"A": [11, 13, 15], "B": [4, 5, 6], "C": [7, 8, 9]})
    tm.assert_frame_equal(df, expected)
    if not using_copy_on_write:
        tm.assert_series_equal(ser, expected["A"])
        tm.assert_series_equal(result_view["A"], expected["A"])
    else:
        expected = Series([1, 2, 3], name="A")
        tm.assert_series_equal(ser, expected)
        tm.assert_series_equal(result_view["A"], expected)

class TestValidate:
    @pytest.mark.parametrize("value", [1, "True", [1, 2, 3], 5.0])
    def test_validate_bool_args(self, value):

        msg = 'For argument "inplace" expected type bool, received type'
        with pytest.raises(ValueError, match=msg):
            pd.eval("2+2", inplace=value)


text:
@interface KBPhotoBrowserDeleteController ()<UICollectionViewDataSource,UICollectionViewDelegate,UICollectionViewDelegateFlowLayout>

{
    UIButton *_navDeleteBtn;
    UICollectionView *_curCollectionView;

    NSInteger _currentPage;
    NSMutableArray<UIImage*> *_curImagesDataSource;
}

@end

@implementation KBPhotoBrowserDeleteController

- (instancetype)init {
    if (self = [super init]) {
        _curImagesDataSource = [NSMutableArray array];
    }
    return self;
}

- (void)viewWillAppear:(BOOL)animated {
    [super viewWillAppear:animated];

    [_curCollectionView setContentOffset:CGPointMake(self.currentIndex*(kWidth+kItemMargin), 0)];
}

- (void)viewDidLoad {
    [super viewDidLoad];
    [self configCurrentViews];
}

- (void)configCurrentViews {
    self.navigationItem.title = [NSString stringWithFormat:@"%ld/%ld", _currentPage, _curImagesDataSource.count];
    self.automaticallyAdjustsScrollViewInsets = NO;

    //init_NavDeleteBtn
    _navDeleteBtn = [UIButton buttonWithType:UIButtonTypeCustom];
    _navDeleteBtn.frame = CGRectMake(0, 0, 25, 25);
    [_navDeleteBtn setBackgroundImage:[UIImage imageNamed:@"express_delete_icon"] forState:UIControlStateNormal];
    [_navDeleteBtn setBackgroundImage:[UIImage imageNamed:@"express_delete_icon"] forState:UIControlStateSelected];
    [_navDeleteBtn addTarget:self action:@selector(navRightBtn_Click:) forControlEvents:UIControlEventTouchUpInside];
    self.navigationItem.rightBarButtonItem = [[UIBarButtonItem alloc] initWithCustomView:_navDeleteBtn];

    //overr backBtn
    UIImage *navBackImg = [UIImage imageNamed:@"navBackBtn"];
    self.navigationItem.leftBarButtonItem = [[UIBarButtonItem alloc] initWithImage:[navBackImg imageWithRenderingMode:UIImageRenderingModeAlwaysOriginal] style:UIBarButtonItemStylePlain target:self action:@selector(btnBack_Click)];

    //init_CollectionView
    UICollectionViewFlowLayout *layOut = [[UICollectionViewFlowLayout alloc]init];
    layOut.scrollDirection = UICollectionViewScrollDirectionHorizontal;
    layOut.minimumLineSpacing = kItemMargin;
    layOut.sectionInset = UIEdgeInsetsMake(0, kItemMargin/2, 0, kItemMargin/2);
    layOut.itemSize = self.view.bounds.size;

    _curCollectionView = [[UICollectionView alloc]initWithFrame:CGRectMake(-kItemMargin/2, 0, kWidth+kItemMargin, kHeight) collectionViewLayout:layOut];
    _curCollectionView.backgroundColor = [UIColor blackColor];
    [_curCollectionView registerClass:[KBPhotoDeleteImgvCollectionCell class] forCellWithReuseIdentifier:@"KBPhotoDeleteImgvCollectionCell"];
    _curCollectionView.dataSource = self;
    _curCollectionView.delegate = self;
    _curCollectionView.pagingEnabled = YES;
    [self.view addSubview:_curCollectionView];
}

    //删除当前页图片-按钮
- (void)navRightBtn_Click:(UIButton *)sender {
    if (_curImagesDataSource.count == 1) {
        [_curImagesDataSource removeAllObjects];
        if (self.compeletionDeleteBlock) {
            self.compeletionDeleteBlock(_curImagesDataSource);
        }
        [self.navigationController popViewControllerAnimated:YES];
        [self.navigationController dismissViewControllerAnimated:YES completion:^{}];
        return;
    }

    UIImage *curImage = _curImagesDataSource[_currentPage - 1];
    __weak typeof(_curImagesDataSource) __weakImageSource = _curImagesDataSource;
    [__weakImageSource enumerateObjectsUsingBlock:^(UIImage * _Nonnull obj,
                                                    NSUInteger idx,
                                                    BOOL * _Nonnull stop) {

        if ([curImage isEqual:obj]) {
            *stop = YES;
            if (*stop == YES) {
                [_curImagesDataSource removeObject:obj];
            }
        }
    }];

    [self scrollViewDidScroll:_curCollectionView];
    [_curCollectionView reloadData];
}

- (void)btnBack_Click {
    if (self.compeletionDeleteBlock) {
        self.compeletionDeleteBlock(_curImagesDataSource);
    }
    [self.navigationController popViewControllerAnimated:YES];
    [self.navigationController dismissViewControllerAnimated:YES completion:^{}];
}

# pragma mark --UICollectionViewDataSource--
- (NSInteger)numberOfSectionsInCollectionView:(UICollectionView *)collectionView {
    return 1;
}

- (NSInteger)collectionView:(UICollectionView *)collectionView numberOfItemsInSection:(NSInteger)section {
    return _curImagesDataSource.count;
}

- (__kindof UICollectionViewCell *)collectionView:(UICollectionView *)collectionView cellForItemAtIndexPath:(NSIndexPath *)indexPath {
    KBPhotoDeleteImgvCollectionCell *cell = [collectionView dequeueReusableCellWithReuseIdentifier:@"KBPhotoDeleteImgvCollectionCell" forIndexPath:indexPath];
    cell.image = _curImagesDataSource[indexPath.row];

    __weak typeof(self) __weakSelf = self;
    cell.singleTapCurImageCallBack = ^{
        if (__weakSelf.navigationController.navigationBar.isHidden) {
            [__weakSelf.navigationController setNavigationBarHidden:NO animated:YES];
            [[UIApplication sharedApplication] setStatusBarHidden:NO withAnimation:UIStatusBarAnimationSlide];
        } else {
            [__weakSelf.navigationController setNavigationBarHidden:YES animated:YES];
            [[UIApplication sharedApplication] setStatusBarHidden:YES withAnimation:UIStatusBarAnimationSlide];
        }
    };
    return cell;
}

# pragma mark --UICollectionViewDelegate--
- (void)scrollViewDidScroll:(UIScrollView *)scrollView {
    if (scrollView == (UIScrollView *)_curCollectionView) {
        //改变导航标题
        CGFloat page = scrollView.contentOffset.x/(kWidth+kItemMargin);
        NSString *str = [NSString stringWithFormat:@"%.0f", page];
        _currentPage = str.integerValue + 1;
        self.navigationItem.title = [NSString stringWithFormat:@"%ld/%ld",_currentPage,_curImagesDataSource.count];
    }
}

- (void)setCurrentSelectedImgeArr:(NSMutableArray<UIImage *> *)currentSelectedImgeArr {
    _curImagesDataSource = currentSelectedImgeArr;
}

- (void)setCurrentIndex:(NSInteger)currentIndex {
    _currentIndex = currentIndex;
    _currentPage = currentIndex + 1;
}

- (void)dealloc {
    NSLog(@"KBPhotoBrowserDeleteController---dealloc");
}

- (void)didReceiveMemoryWarning {
    [super didReceiveMemoryWarning];
}

@end

redpajama_set_name: RedPajamaGithub


text:
layout: page
title: Christensen 15th Anniversary
date: 2016-05-24
author: Madison Campbell
tags: weekly links, java
status: published
summary: Nullam ac luctus quam. Sed arcu arcu, eleifend id.
banner: images/banner/office-01.jpg
booking:
  startDate: 02/21/2017
  endDate: 02/26/2017
  ctyhocn: PTCFLHX
  groupCode: C1A
published: true
---
Cras auctor urna sed finibus faucibus. Ut tincidunt bibendum nibh in euismod. Nulla diam nulla, fringilla a enim non, molestie ultricies enim. Integer porta scelerisque orci sed rutrum. Nullam suscipit erat orci, sit amet semper quam semper in. Vivamus accumsan, leo non fermentum interdum, libero nisl congue purus, et volutpat nisl orci et arcu. Aliquam quis ligula risus. Nullam congue lectus a ipsum pellentesque scelerisque. Proin in placerat massa. Fusce ut mi a nunc dapibus cursus ut sed nibh. Sed sit amet consequat orci. In tincidunt lorem eros, vel varius dolor pharetra id.

* Quisque molestie nunc tincidunt tempus cursus
* Sed sit amet ante rhoncus, consequat libero vel, luctus ante.

Morbi convallis, ipsum et faucibus mattis, tortor arcu interdum erat, at vehicula justo massa rutrum ligula. Maecenas nunc felis, pulvinar vel mi vel, viverra condimentum ex. Aliquam nec condimentum libero, nec malesuada leo. Praesent facilisis quam nec interdum fringilla. Mauris id felis vitae mauris porttitor tincidunt. Integer magna ante, elementum in tristique vitae, ultricies vel tellus. Nulla tristique vel tortor sed laoreet. Nullam urna leo, condimentum ac porttitor sit amet, consequat vitae nunc. Nam id scelerisque metus. Donec semper risus vestibulum, vestibulum enim et, ornare neque. Nulla nulla orci, gravida non ligula nec, convallis fringilla ipsum. Nulla pulvinar et libero eget dignissim. Integer semper eros libero, ac lobortis nulla pellentesque gravida. Suspendisse ut ligula in metus consequat finibus eu at velit. Etiam purus nunc, vehicula ac hendrerit a, feugiat vel leo.


text:
/**
 * Muirhat (NPC 1092007)
 * The Nautilus: Top Floor - Hallway (Map 120000100)
 *
 * Advances A Mysterious Presence On-board The Nautilus - Disciples of the Black
 * Magician (Quest 2175)
 *
 * @author GoldenKevin (content from DelphiMS r418)
 */

if (!player.isQuestStarted(2175)) {
	npc.say("The Black Magician and his followers. Kyrin and the Crew of Nautilus. They'll be chasing one another until one of them doesn't exist, that's for sure.");
} else {
	npc.sayNext("Are you ready? Good, I'll send you to where the disciples of the Black Magician are. Look for the pigs around the area where I'll be sending you. You'll be able to find it by tracking them.");
	npc.sayNext("When they are weakened, they'll change back to their original state. If you find something suspicious, you must fight them until they are weak. I'll be here awaiting your findings.");
	player.changeMap(912000000);
}


text:
package VMOMI::ClusterDrsFaultsFaultsByVm;
use parent 'VMOMI::DynamicData';

use strict;
use warnings;

our @class_ancestors = (
    'DynamicData',
);

our @class_members = (
    ['vm', 'ManagedObjectReference', 0, 1],
    ['fault', 'LocalizedMethodFault', 1, ],
);

sub get_class_ancestors {
    return @class_ancestors;
}

sub get_class_members {
    my $class = shift;
    my @super_members = $class->SUPER::get_class_members();
    return (@super_members, @class_members);
}

1;


text:
package com.daniel.weixin.mp.demo;

import com.daniel.weixin.common.exception.WxErrorException;
import com.daniel.weixin.mp.api.WxMpService;
import com.daniel.weixin.mp.bean.result.WxMpOAuth2AccessToken;
import com.daniel.weixin.mp.bean.result.WxMpUser;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;

public class WxMpOAuth2Servlet extends HttpServlet {

  protected WxMpService wxMpService;

  public WxMpOAuth2Servlet(WxMpService wxMpService) {
    this.wxMpService = wxMpService;
  }

  @Override protected void service(HttpServletRequest request, HttpServletResponse response)
      throws ServletException, IOException {

    response.setContentType("text/html;charset=utf-8");
    response.setStatus(HttpServletResponse.SC_OK);

    String code = request.getParameter("code");
    try {
      response.getWriter().println("
code
");
      response.getWriter().println(code);

      WxMpOAuth2AccessToken wxMpOAuth2AccessToken = wxMpService.oauth2getAccessToken(code);
      response.getWriter().println("
access token
");
      response.getWriter().println(wxMpOAuth2AccessToken.toString());

      WxMpUser wxMpUser = wxMpService.oauth2getUserInfo(wxMpOAuth2AccessToken, null);
      response.getWriter().println("
user info
");
      response.getWriter().println(wxMpUser.toString());

      wxMpOAuth2AccessToken = wxMpService.oauth2refreshAccessToken(wxMpOAuth2AccessToken.getRefreshToken());
      response.getWriter().println("
after refresh
");
      response.getWriter().println(wxMpOAuth2AccessToken.toString());

    } catch (WxErrorException e) {
      e.printStackTrace();
    }

    response.getWriter().flush();
    response.getWriter().close();

  }

}


text:
/* TEMPLATE GENERATED TESTCASE FILE
Filename: CWE680_Integer_Overflow_to_Buffer_Overflow__malloc_listen_socket_66b.c
Label Definition File: CWE680_Integer_Overflow_to_Buffer_Overflow__malloc.label.xml
Template File: sources-sink-66b.tmpl.c
*/
/*
 * @description
 * CWE: 680 Integer Overflow to Buffer Overflow
 * BadSource: listen_socket Read data using a listen socket (server side)
 * GoodSource: Small number greater than zero that will not cause an integer overflow in the sink
 * Sinks:
 *    BadSink : Attempt to allocate array using length value from source
 * Flow Variant: 66 Data flow: data passed in an array from one function to another in different source files
 *
 * */

#include "std_testcase.h"

#ifdef _WIN32
#include <winsock2.h>
#include <windows.h>
#include <direct.h>
#pragma comment(lib, "ws2_32") /* include ws2_32.lib when linking */
#define CLOSE_SOCKET closesocket
#else
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#define INVALID_SOCKET -1
#define SOCKET_ERROR -1
#define CLOSE_SOCKET close
#define SOCKET int
#endif

#define TCP_PORT 27015
#define LISTEN_BACKLOG 5
#define CHAR_ARRAY_SIZE (3 * sizeof(data) + 2)

#ifndef OMITBAD

void CWE680_Integer_Overflow_to_Buffer_Overflow__malloc_listen_socket_66b_badSink(int dataArray[])
{
    /* copy data out of dataArray */
    int data = dataArray[2];
    {
        size_t i;
        int *intPointer;
        /* POTENTIAL FLAW: if data * sizeof(int) > SIZE_MAX, overflows to a small value
         * so that the for loop doing the initialization causes a buffer overflow */
        intPointer = (int*)malloc(data * sizeof(int));
        for (i = 0; i < (size_t)data; i++)
        {
            intPointer[i] = 0; /* Potentially writes beyond the boundary of intPointer */
        }
        printIntLine(intPointer[0]);
        free(intPointer);
    }
}

#endif /* OMITBAD */

#ifndef OMITGOOD

/* goodG2B uses the GoodSource with the BadSink */
void CWE680_Integer_Overflow_to_Buffer_Overflow__malloc_listen_socket_66b_goodG2BSink(int dataArray[])
{
    int data = dataArray[2];
    {
        size_t i;
        int *intPointer;
        /* POTENTIAL FLAW: if data * sizeof(int) > SIZE_MAX, overflows to a small value
         * so that the for loop doing the initialization causes a buffer overflow */
        intPointer = (int*)malloc(data * sizeof(int));
        for (i = 0; i < (size_t)data; i++)
        {
            intPointer[i] = 0; /* Potentially writes beyond the boundary of intPointer */
        }
        printIntLine(intPointer[0]);
        free(intPointer);
    }
}

#endif /* OMITGOOD */


text:
package com.iclockwork.percy.wechat4j.entity;

import com.iclockwork.percy.wechat4j.internal.http.HttpResponse;
import com.iclockwork.percy.wechat4j.internal.json.JSONArray;
import com.iclockwork.percy.wechat4j.internal.json.JSONObject;

import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;

final class DeptsJSONImpl implements Depts, Serializable {
    private static final long serialVersionUID = -4328577975502421677L;

    private Dept[] depts;
    private Response response;

    DeptsJSONImpl(HttpResponse response) throws IOException {
        this(response.asJSONObject());
    }

    DeptsJSONImpl(JSONObject jsonObject) {
        init(jsonObject);
    }

    void init(JSONObject jsonObject) {
        response = new ResponseJSONImpl(jsonObject);
        if (!jsonObject.isNull("department")) {
            JSONArray array = jsonObject.getJSONArray("department");

            int size = array.length();
            depts = new Dept[size];
            for (int i=0; i<size; i++) {
                depts[i] = new DeptJSONImpl(array.getJSONObject(i));
            }
        } else {
            depts = new Dept[]{};
        }
    }

    @Override
    public Dept[] getDepts() {
        return depts;
    }

    @Override
    public Response getResponse() {
        return response;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o==null || getClass()!=o.getClass()) return false;

        DeptsJSONImpl that = (DeptsJSONImpl)o;

        if (!Arrays.equals(depts, that.depts)) return false;
        if (response!=null ? !response.equals(that.response):that.response!=null) return false;
        return true;
    }

    @Override
    public int hashCode() {
        int result = 0;
        result = depts!=null? Arrays.hashCode(depts): 0;
        result = 31*result + (response!=null? response.hashCode(): 0);
        return result;
    }

    @Override
    public String toString() {
        if (response != null) {
            return response.toString();
        } else {
            return "{\"depts\": {}}";
        }
    }
}


text:
"""
Twitter pseudo-API

Created by: Rui Carmo (https://github.com/rcarmo)
"""

import os, sys, logging, time, datetime, calendar
from json import dumps, loads, JSONEncoder
from bottle import route, get, put, post, delete, request, response, abort
from controllers.feeds import FeedController
from decorators import jsonp, cache_results, timed, redis_cache, cache_control
import feedparser

class ExtendedEncoder(JSONEncoder):
    """JSON encoder that knows how to serialize some extra types"""

    def default(self, item):
        if isinstance(item, datetime.datetime):
            return calendar.timegm(item.utctimetuple())
        if isinstance(item, datetime.date):
            return time.mktime(item.timetuple())
        if isinstance(item, time.struct_time):
            return calendar.timegm(item)
        return JSONEncoder.default(self, item)

prefix = '/feeds'
log = logging.getLogger()
c = FeedController()

@get(prefix + '/time')
@timed
@jsonp
def get_server_time():
    """Returns the server clock (we don't trust the client clocks)"""
    return {"server_time":int(time.time() * 1000)}
    return {"server_time":int(calendar.timegm(datetime.datetime.utcnow().utctimetuple()) * 1000)}

@get(prefix + '/twitter')
@timed
@cache_control(60)
@jsonp
def get_twitter_data():
    """Returns the latest twitter data"""

    res = c.twitter_feed
    if not res:
        abort(204, "No content")
        return None

    response.content_type = "application/json"
    return res

@get(prefix + '/rss/<feed_name>')
@timed
@cache_results(60)
@cache_control(60)
@jsonp
def get_rss_feed(feed_name):
    try:
        feed_response = c.get_feed(feed_name)

        response.status = feed_response['status']

        if feed_response['status'] == 200:
            # SAPO Fotos gives us the wrong content-type, so testing for text/xml is kind of pointless
            if 'text/' in feed_response['content-type']:
                encoder = ExtendedEncoder()

                feed_response['data'] = encoder.encode(feedparser.parse(feed_response['data']))

            # return a dict so that Bottle can handle this for us
            return loads(feed_response['data'])
    except Exception as ex:
        abort(404, "Feed not found: %s" % ex)
        log.error(ex)


text:
/* internal.h  -  Foundation library  -  Public Domain  -  2013 Mattias Jansson / Rampant Pixels
 *
 * This library provides a cross-platform foundation library in C11 providing basic support data types and
 * functions to write applications and games in a platform-independent fashion. The latest source code is
 * always available at
 *
 * https://github.com/rampantpixels/foundation_lib
 *
 * This library is put in the public domain; you can redistribute it and/or modify it without any restrictions.
 *
 */

#pragma once

/*! \file internal.h
    Internal types */

#include <foundation/platform.h>
#include <foundation/types.h>

//Internal entry points

FOUNDATION_API int _atomic_initialize( void );
FOUNDATION_API void _atomic_shutdown( void );

FOUNDATION_API void _ringbuffer_stream_initialize( void );
FOUNDATION_API void _buffer_stream_initialize( void );
#if FOUNDATION_PLATFORM_ANDROID
FOUNDATION_API void _asset_stream_initialize( void );
#endif
FOUNDATION_API void _pipe_stream_initialize( void );

FOUNDATION_API int _log_initialize( void );
FOUNDATION_API void _log_shutdown( void );

FOUNDATION_API int _memory_initialize( const memory_system_t memory );
FOUNDATION_API void _memory_preallocate( void );
FOUNDATION_API void _memory_shutdown( void );

FOUNDATION_API int _time_initialize( void );
FOUNDATION_API void _time_shutdown( void );

FOUNDATION_API int _thread_initialize( void );
FOUNDATION_API void _thread_shutdown( void );

FOUNDATION_API int _environment_initialize( const application_t application );
FOUNDATION_API void _environment_shutdown( void );

FOUNDATION_API int _library_initialize( void );
FOUNDATION_API void _library_shutdown( void );

FOUNDATION_API int _system_initialize( void );
FOUNDATION_API void _system_shutdown( void );

FOUNDATION_API int _stream_initialize( void );
FOUNDATION_API void _stream_shutdown( void );

FOUNDATION_API int _fs_initialize( void );
FOUNDATION_API void _fs_shutdown( void );

FOUNDATION_API int _random_initialize( void );
FOUNDATION_API void _random_shutdown( void );

FOUNDATION_API int _config_initialize( void );
FOUNDATION_API void _config_shutdown( void );

FOUNDATION_API void _profile_thread_finalize( void );

FOUNDATION_API int _static_hash_initialize( void );
FOUNDATION_API void _static_hash_shutdown( void );

FOUNDATION_API void foundation_startup( void );

// Internal functions

FOUNDATION_API uint32_t      hashtable32_raw( hashtable32_t* table, uint32_t key );
FOUNDATION_API uint64_t      hashtable64_raw( hashtable64_t* table, uint64_t key );

FOUNDATION_API void          _object_initialize( object_base_t* obj, object_t id );
FOUNDATION_API object_t      _object_ref( object_base_t* obj );
FOUNDATION_API object_t      _object_unref( object_base_t* obj );

FOUNDATION_API void          _environment_main_args( int argc, const char* const* argv );


text:
package com.alachisoft.tayzgrid.web.caching;

public enum QueryChangeType {

    Add,
    Update,
    Remove;

    public static QueryChangeType forValue(int value) {
        return values()[value];
    }
}


text:
<?php

namespace ZendTest\Romans\Filter;

use PHPUnit\Framework\TestCase;
use Zend\Filter\FilterInterface;
use Zend\Romans\Filter\RomanToInt;

/**
 * Roman to Int Test
 */
class RomanToIntTest extends TestCase
{
    /**
     * {@inheritdoc}
     */
    protected function setUp()
    {
        $this->filter = new RomanToInt();
    }

    /**
     * Test Instance Of
     */
    public function testInstanceOf()
    {
        $this->assertInstanceOf(FilterInterface::class, $this->filter);
    }

    /**
     * Test Filter
     */
    public function testFilter()
    {
        $this->assertSame(1, $this->filter->filter('I'));
        $this->assertSame(5, $this->filter->filter('V'));
        $this->assertSame(10, $this->filter->filter('X'));
    }

    /**
     * Test Filter with Invalid Input
     */
    public function testFilterWithInvalidInput()
    {
        $this->assertNull($this->filter->filter('.'));
        $this->assertNull($this->filter->filter('IIX'));
        $this->assertNull($this->filter->filter(1));
        $this->assertNull($this->filter->filter(false));
    }
}


text:
from pygments import lexers, highlight, formatters
from pygments.util import ClassNotFound
from lxml import etree
from lxml.html import fragment_fromstring

def html_highlight(context, language, code, config):
    """
    Highlight the given ``code`` in the given ``language``.  ``context`` is
    the XPath context, in which this function was applied.  ``config`` is
    ignored.

    Return a list of HTML nodes containined the highlighted code.
    """
    # necessary, because lxml somehow fails to correctly pass code (reason
    # is unknown to me)
    if not code:
        code = context.context_node.xpath('.//text()')
    try:
        lang = language[0].lower()
        lexer = lexers.get_lexer_by_name(lang)
    except ClassNotFound:
        lexer = lexers.guess_lexer(code[0])
    html = highlight(code[0], lexer, formatters.HtmlFormatter(nowrap=True))
    highlight_div = fragment_fromstring(html, create_parent='code')
    highlight_div.set('class', lang)
    return [highlight_div]

def main(xslt_file,xml_file,out_file,args):
    xslt_params = dict()
    for p in args:
        eq = p.index('=')
        xslt_params[p[0:eq]] = etree.XSLT.strparam(p[eq+1:])
    # print(xslt_params)

    xhl = etree.FunctionNamespace('http://net.sf.xslthl/ConnectorXalan')
    xhl.prefix = 'xhl'
    xhl['highlight'] = html_highlight

    xml_p = etree.XMLParser(remove_blank_text=True, resolve_entities=True, load_dtd=True)
    transform = etree.XSLT(etree.parse(xslt_file, xml_p))
    document = etree.parse(xml_file, xml_p)
    document.xinclude()
    out_file.write(str(transform(document, **xslt_params)))
    return 0

def print_usage(cmd):
    print("USAGE: python %s <xslt> <in_xml> <out_file>"%(cmd))

if __name__=="__main__":
    import sys
    exit = 1
    if len(sys.argv) > 3:
        outf = sys.stdout
        if sys.argv[3] != '-':
            outf = open(sys.argv[3],'w')
        try:
            exit = main(sys.argv[1],sys.argv[2],outf,sys.argv[4:])
        finally:
            outf.close()
    if exit != 0:
        print_usage(sys.argv[0])
    sys.exit(exit)