mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-11 19:29:52 -04:00
Carlos Sousa
af655e55d3
An empty sequence passed to any ``execute()`` method now
raised a deprecation warning, since such an executemany
is invalid.
Pull request courtesy of Carlos Sousa.
Fixes: #9647
Closes: #10406
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/10406
Pull-request-sha: 087ba2d88d
Change-Id: I482e91a047477c3156a3ca806e5c1eefb6224b95
301 lines
9.8 KiB
Python
301 lines
9.8 KiB
Python
import datetime
|
|
import re
|
|
from types import MappingProxyType
|
|
|
|
from sqlalchemy import exc
|
|
from sqlalchemy.engine import processors
|
|
from sqlalchemy.testing import assert_raises_message
|
|
from sqlalchemy.testing import combinations
|
|
from sqlalchemy.testing import eq_
|
|
from sqlalchemy.testing import expect_raises_message
|
|
from sqlalchemy.testing import fixtures
|
|
from sqlalchemy.testing import is_none
|
|
from sqlalchemy.testing.assertions import expect_deprecated
|
|
from sqlalchemy.util import immutabledict
|
|
|
|
|
|
class _BooleanProcessorTest(fixtures.TestBase):
|
|
def test_int_to_bool_none(self):
|
|
eq_(self.module.int_to_boolean(None), None)
|
|
|
|
def test_int_to_bool_zero(self):
|
|
eq_(self.module.int_to_boolean(0), False)
|
|
|
|
def test_int_to_bool_one(self):
|
|
eq_(self.module.int_to_boolean(1), True)
|
|
|
|
def test_int_to_bool_positive_int(self):
|
|
eq_(self.module.int_to_boolean(12), True)
|
|
|
|
def test_int_to_bool_negative_int(self):
|
|
eq_(self.module.int_to_boolean(-4), True)
|
|
|
|
|
|
class CyBooleanProcessorTest(_BooleanProcessorTest):
|
|
__requires__ = ("cextensions",)
|
|
|
|
@classmethod
|
|
def setup_test_class(cls):
|
|
from sqlalchemy.engine import _processors_cy
|
|
|
|
cls.module = _processors_cy
|
|
|
|
|
|
class _DateProcessorTest(fixtures.TestBase):
|
|
def test_iso_datetime(self):
|
|
eq_(
|
|
self.module.str_to_datetime("2022-04-03 17:12:34.353"),
|
|
datetime.datetime(2022, 4, 3, 17, 12, 34, 353000),
|
|
)
|
|
|
|
eq_(
|
|
self.module.str_to_datetime("2022-04-03 17:12:34.353123"),
|
|
datetime.datetime(2022, 4, 3, 17, 12, 34, 353123),
|
|
)
|
|
|
|
eq_(
|
|
self.module.str_to_datetime("2022-04-03 17:12:34"),
|
|
datetime.datetime(2022, 4, 3, 17, 12, 34),
|
|
)
|
|
|
|
eq_(
|
|
self.module.str_to_time("17:12:34.353123"),
|
|
datetime.time(17, 12, 34, 353123),
|
|
)
|
|
|
|
eq_(
|
|
self.module.str_to_time("17:12:34.353"),
|
|
datetime.time(17, 12, 34, 353000),
|
|
)
|
|
|
|
eq_(
|
|
self.module.str_to_time("17:12:34"),
|
|
datetime.time(17, 12, 34),
|
|
)
|
|
|
|
eq_(self.module.str_to_date("2022-04-03"), datetime.date(2022, 4, 3))
|
|
|
|
@combinations("str_to_datetime", "str_to_time", "str_to_date")
|
|
def test_no_string(self, meth):
|
|
with expect_raises_message(
|
|
TypeError, "fromisoformat: argument must be str"
|
|
):
|
|
fn = getattr(self.module, meth)
|
|
fn(2012)
|
|
|
|
def test_datetime_no_string_custom_reg(self):
|
|
assert_raises_message(
|
|
ValueError,
|
|
"Couldn't parse datetime string '2012' - value is not a string",
|
|
processors.str_to_datetime_processor_factory(
|
|
re.compile(r"(\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+)(?:\.(\d+))?"),
|
|
datetime.datetime,
|
|
),
|
|
2012,
|
|
)
|
|
|
|
def test_time_no_string_custom_reg(self):
|
|
assert_raises_message(
|
|
ValueError,
|
|
"Couldn't parse time string '2012' - value is not a string",
|
|
processors.str_to_datetime_processor_factory(
|
|
re.compile(r"^(\d+):(\d+):(\d+)(?:\.(\d{6}))?$"), datetime.time
|
|
),
|
|
2012,
|
|
)
|
|
|
|
@combinations("str_to_datetime", "str_to_time", "str_to_date")
|
|
def test_invalid_string(self, meth):
|
|
with expect_raises_message(
|
|
ValueError, "Invalid isoformat string: '5:a'"
|
|
):
|
|
fn = getattr(self.module, meth)
|
|
fn("5:a")
|
|
|
|
@combinations("str_to_datetime", "str_to_time", "str_to_date")
|
|
def test_none(self, meth):
|
|
fn = getattr(self.module, meth)
|
|
is_none(fn(None))
|
|
|
|
|
|
class PyDateProcessorTest(_DateProcessorTest):
|
|
@classmethod
|
|
def setup_test_class(cls):
|
|
from sqlalchemy.engine import _processors_cy
|
|
from sqlalchemy.util.langhelpers import load_uncompiled_module
|
|
|
|
py_mod = load_uncompiled_module(_processors_cy)
|
|
|
|
cls.module = py_mod
|
|
|
|
|
|
class CyDateProcessorTest(_DateProcessorTest):
|
|
__requires__ = ("cextensions",)
|
|
|
|
@classmethod
|
|
def setup_test_class(cls):
|
|
from sqlalchemy.engine import _processors_cy
|
|
|
|
assert _processors_cy._is_compiled()
|
|
cls.module = _processors_cy
|
|
|
|
|
|
class _DistillArgsTest(fixtures.TestBase):
|
|
def test_distill_20_none(self):
|
|
eq_(self.module._distill_params_20(None), ())
|
|
|
|
def test_distill_20_empty_sequence(self):
|
|
with expect_deprecated(
|
|
r"Empty parameter sequence passed to execute\(\). "
|
|
"This use is deprecated and will raise an exception in a "
|
|
"future SQLAlchemy release"
|
|
):
|
|
eq_(self.module._distill_params_20(()), ())
|
|
eq_(self.module._distill_params_20([]), [])
|
|
|
|
def test_distill_20_sequence_sequence(self):
|
|
eq_(self.module._distill_params_20(((1, 2, 3),)), ((1, 2, 3),))
|
|
eq_(self.module._distill_params_20([(1, 2, 3)]), [(1, 2, 3)])
|
|
|
|
eq_(self.module._distill_params_20(((1, 2), (2, 3))), ((1, 2), (2, 3)))
|
|
eq_(self.module._distill_params_20([(1, 2), (2, 3)]), [(1, 2), (2, 3)])
|
|
|
|
def test_distill_20_sequence_dict(self):
|
|
eq_(self.module._distill_params_20(({"a": 1},)), ({"a": 1},))
|
|
eq_(
|
|
self.module._distill_params_20([{"a": 1}, {"a": 2}]),
|
|
[{"a": 1}, {"a": 2}],
|
|
)
|
|
eq_(
|
|
self.module._distill_params_20((MappingProxyType({"a": 1}),)),
|
|
(MappingProxyType({"a": 1}),),
|
|
)
|
|
|
|
def test_distill_20_sequence_error(self):
|
|
with expect_raises_message(
|
|
exc.ArgumentError,
|
|
"List argument must consist only of tuples or dictionaries",
|
|
):
|
|
self.module._distill_params_20((1, 2, 3))
|
|
with expect_raises_message(
|
|
exc.ArgumentError,
|
|
"List argument must consist only of tuples or dictionaries",
|
|
):
|
|
self.module._distill_params_20(([1, 2, 3],))
|
|
with expect_raises_message(
|
|
exc.ArgumentError,
|
|
"List argument must consist only of tuples or dictionaries",
|
|
):
|
|
self.module._distill_params_20([1, 2, 3])
|
|
with expect_raises_message(
|
|
exc.ArgumentError,
|
|
"List argument must consist only of tuples or dictionaries",
|
|
):
|
|
self.module._distill_params_20(["a", "b"])
|
|
|
|
def test_distill_20_dict(self):
|
|
eq_(self.module._distill_params_20({"foo": "bar"}), [{"foo": "bar"}])
|
|
eq_(
|
|
self.module._distill_params_20(immutabledict({"foo": "bar"})),
|
|
[immutabledict({"foo": "bar"})],
|
|
)
|
|
eq_(
|
|
self.module._distill_params_20(MappingProxyType({"foo": "bar"})),
|
|
[MappingProxyType({"foo": "bar"})],
|
|
)
|
|
|
|
def test_distill_20_error(self):
|
|
with expect_raises_message(
|
|
exc.ArgumentError, "mapping or list expected for parameters"
|
|
):
|
|
self.module._distill_params_20("foo")
|
|
with expect_raises_message(
|
|
exc.ArgumentError, "mapping or list expected for parameters"
|
|
):
|
|
self.module._distill_params_20(1)
|
|
|
|
def test_distill_raw_none(self):
|
|
eq_(self.module._distill_raw_params(None), ())
|
|
|
|
def test_distill_raw_empty_list(self):
|
|
eq_(self.module._distill_raw_params([]), [])
|
|
|
|
def test_distill_raw_list_sequence(self):
|
|
eq_(self.module._distill_raw_params([(1, 2, 3)]), [(1, 2, 3)])
|
|
eq_(
|
|
self.module._distill_raw_params([(1, 2), (2, 3)]), [(1, 2), (2, 3)]
|
|
)
|
|
|
|
def test_distill_raw_list_dict(self):
|
|
eq_(
|
|
self.module._distill_raw_params([{"a": 1}, {"a": 2}]),
|
|
[{"a": 1}, {"a": 2}],
|
|
)
|
|
eq_(
|
|
self.module._distill_raw_params([MappingProxyType({"a": 1})]),
|
|
[MappingProxyType({"a": 1})],
|
|
)
|
|
|
|
def test_distill_raw_sequence_error(self):
|
|
with expect_raises_message(
|
|
exc.ArgumentError,
|
|
"List argument must consist only of tuples or dictionaries",
|
|
):
|
|
self.module._distill_raw_params([1, 2, 3])
|
|
with expect_raises_message(
|
|
exc.ArgumentError,
|
|
"List argument must consist only of tuples or dictionaries",
|
|
):
|
|
self.module._distill_raw_params([[1, 2, 3]])
|
|
with expect_raises_message(
|
|
exc.ArgumentError,
|
|
"List argument must consist only of tuples or dictionaries",
|
|
):
|
|
self.module._distill_raw_params(["a", "b"])
|
|
|
|
def test_distill_raw_tuple(self):
|
|
eq_(self.module._distill_raw_params(()), [()])
|
|
eq_(self.module._distill_raw_params((1, 2, 3)), [(1, 2, 3)])
|
|
|
|
def test_distill_raw_dict(self):
|
|
eq_(self.module._distill_raw_params({"foo": "bar"}), [{"foo": "bar"}])
|
|
eq_(
|
|
self.module._distill_raw_params(immutabledict({"foo": "bar"})),
|
|
[immutabledict({"foo": "bar"})],
|
|
)
|
|
eq_(
|
|
self.module._distill_raw_params(MappingProxyType({"foo": "bar"})),
|
|
[MappingProxyType({"foo": "bar"})],
|
|
)
|
|
|
|
def test_distill_raw_error(self):
|
|
with expect_raises_message(
|
|
exc.ArgumentError, "mapping or sequence expected for parameters"
|
|
):
|
|
self.module._distill_raw_params("foo")
|
|
with expect_raises_message(
|
|
exc.ArgumentError, "mapping or sequence expected for parameters"
|
|
):
|
|
self.module._distill_raw_params(1)
|
|
|
|
|
|
class PyDistillArgsTest(_DistillArgsTest):
|
|
@classmethod
|
|
def setup_test_class(cls):
|
|
from sqlalchemy.engine import _util_cy
|
|
from sqlalchemy.util.langhelpers import load_uncompiled_module
|
|
|
|
_py_util = load_uncompiled_module(_util_cy)
|
|
cls.module = _py_util
|
|
|
|
|
|
class CyDistillArgsTest(_DistillArgsTest):
|
|
__requires__ = ("cextensions",)
|
|
|
|
@classmethod
|
|
def setup_test_class(cls):
|
|
from sqlalchemy.engine import _util_cy
|
|
|
|
assert _util_cy._is_compiled()
|
|
cls.module = _util_cy
|