mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-18 06:32:08 -04:00
Mike Bayer
f1e96cb087
To allow the "connection" pytest fixture and others work correctly in conjunction with setup/teardown that expects to be external to the transaction, remove and prevent any usage of "xdist" style names that are hardcoded by pytest to run inside of fixtures, even function level ones. Instead use pytest autouse fixtures to implement our own r"setup|teardown_test(?:_class)?" methods so that we can ensure function-scoped fixtures are run within them. A new more explicit flow is set up within plugin_base and pytestplugin such that the order of setup/teardown steps, which there are now many, is fully documented and controllable. New granularity has been added to the test teardown phase to distinguish between "end of the test" when lock-holding structures on connections should be released to allow for table drops, vs. "end of the test plus its teardown steps" when we can perform final cleanup on connections and run assertions that everything is closed out. From there we can remove most of the defensive "tear down everything" logic inside of engines which for many years would frequently dispose of pools over and over again, creating for a broken and expensive connection flow. A quick test shows that running test/sql/ against a single Postgresql engine with the new approach uses 75% fewer new connections, creating 42 new connections total, vs. 164 new connections total with the previous system. As part of this, the new fixtures metadata/connection/future_connection have been integrated such that they can be combined together effectively. The fixture_session(), provide_metadata() fixtures have been improved, including that fixture_session() now strongly references sessions which are explicitly torn down before table drops occur afer a test. Major changes have been made to the ConnectionKiller such that it now features different "scopes" for testing engines and will limit its cleanup to those testing engines corresponding to end of test, end of test class, or end of test session. The system by which it tracks DBAPI connections has been reworked, is ultimately somewhat similar to how it worked before but is organized more clearly along with the proxy-tracking logic. A "testing_engine" fixture is also added that works as a pytest fixture rather than a standalone function. The connection cleanup logic should now be very robust, as we now can use the same global connection pools for the whole suite without ever disposing them, while also running a query for PostgreSQL locks remaining after every test and assert there are no open transactions leaking between tests at all. Additional steps are added that also accommodate for asyncio connections not explicitly closed, as is the case for legacy sync-style tests as well as the async tests themselves. As always, hundreds of tests are further refined to use the new fixtures where problems with loose connections were identified, largely as a result of the new PostgreSQL assertions, many more tests have moved from legacy patterns into the newest. An unfortunate discovery during the creation of this system is that autouse fixtures (as well as if they are set up by @pytest.mark.usefixtures) are not usable at our current scale with pytest 4.6.11 running under Python 2. It's unclear if this is due to the older version of pytest or how it implements itself for Python 2, as well as if the issue is CPU slowness or just large memory use, but collecting the full span of tests takes over a minute for a single process when any autouse fixtures are in place and on CI the jobs just time out after ten minutes. So at the moment this patch also reinvents a small version of "autouse" fixtures when py2k is running, which skips generating the real fixture and instead uses two global pytest fixtures (which don't seem to impact performance) to invoke the "autouse" fixtures ourselves outside of pytest. This will limit our ability to do more with fixtures until we can remove py2k support. py.test is still observed to be much slower in collection in the 4.6.11 version compared to modern 6.2 versions, so add support for new TOX_POSTGRESQL_PY2K and TOX_MYSQL_PY2K environment variables that will run the suite for fewer backends under Python 2. For Python 3 pin pytest to modern 6.2 versions where performance for collection has been improved greatly. Includes the following improvements: Fixed bug in asyncio connection pool where ``asyncio.TimeoutError`` would be raised rather than :class:`.exc.TimeoutError`. Also repaired the :paramref:`_sa.create_engine.pool_timeout` parameter set to zero when using the async engine, which previously would ignore the timeout and block rather than timing out immediately as is the behavior with regular :class:`.QueuePool`. For asyncio the connection pool will now also not interact at all with an asyncio connection whose ConnectionFairy is being garbage collected; a warning that the connection was not properly closed is emitted and the connection is discarded. Within the test suite the ConnectionKiller is now maintaining strong references to all DBAPI connections and ensuring they are released when tests end, including those whose ConnectionFairy proxies are GCed. Identified cx_Oracle.stmtcachesize as a major factor in Oracle test scalability issues, this can be reset on a per-test basis rather than setting it to zero across the board. the addition of this flag has resolved the long-standing oracle "two task" error problem. For SQL Server, changed the temp table style used by the "suite" tests to be the double-pound-sign, i.e. global, variety, which is much easier to test generically. There are already reflection tests that are more finely tuned to both styles of temp table within the mssql test suite. Additionally, added an extra step to the "dropfirst" mechanism for SQL Server that will remove all foreign key constraints first as some issues were observed when using this flag when multiple schemas had not been torn down. Identified and fixed two subtle failure modes in the engine, when commit/rollback fails in a begin() context manager, the connection is explicitly closed, and when "initialize()" fails on the first new connection of a dialect, the transactional state on that connection is still rolled back. Fixes: #5826 Fixes: #5827 Change-Id: Ib1d05cb8c7cf84f9a4bfd23df397dc23c9329bfe
1151 lines
38 KiB
Python
1151 lines
38 KiB
Python
from copy import deepcopy
|
|
import datetime
|
|
import decimal
|
|
|
|
from sqlalchemy import ARRAY
|
|
from sqlalchemy import bindparam
|
|
from sqlalchemy import Boolean
|
|
from sqlalchemy import Column
|
|
from sqlalchemy import Date
|
|
from sqlalchemy import DateTime
|
|
from sqlalchemy import extract
|
|
from sqlalchemy import func
|
|
from sqlalchemy import Integer
|
|
from sqlalchemy import literal
|
|
from sqlalchemy import literal_column
|
|
from sqlalchemy import Numeric
|
|
from sqlalchemy import select
|
|
from sqlalchemy import Sequence
|
|
from sqlalchemy import sql
|
|
from sqlalchemy import String
|
|
from sqlalchemy import Table
|
|
from sqlalchemy import testing
|
|
from sqlalchemy import types as sqltypes
|
|
from sqlalchemy import util
|
|
from sqlalchemy.dialects import mysql
|
|
from sqlalchemy.dialects import oracle
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy.dialects import sqlite
|
|
from sqlalchemy.sql import column
|
|
from sqlalchemy.sql import functions
|
|
from sqlalchemy.sql import quoted_name
|
|
from sqlalchemy.sql import table
|
|
from sqlalchemy.sql.compiler import BIND_TEMPLATES
|
|
from sqlalchemy.sql.functions import FunctionElement
|
|
from sqlalchemy.sql.functions import GenericFunction
|
|
from sqlalchemy.testing import assert_raises
|
|
from sqlalchemy.testing import assert_raises_message
|
|
from sqlalchemy.testing import AssertsCompiledSQL
|
|
from sqlalchemy.testing import eq_
|
|
from sqlalchemy.testing import fixtures
|
|
from sqlalchemy.testing import is_
|
|
from sqlalchemy.testing.assertions import expect_warnings
|
|
from sqlalchemy.testing.engines import all_dialects
|
|
|
|
|
|
table1 = table(
|
|
"mytable",
|
|
column("myid", Integer),
|
|
column("name", String),
|
|
column("description", String),
|
|
)
|
|
|
|
|
|
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
|
|
__dialect__ = "default"
|
|
|
|
def setup_test(self):
|
|
self._registry = deepcopy(functions._registry)
|
|
|
|
def teardown_test(self):
|
|
functions._registry = self._registry
|
|
|
|
def test_compile(self):
|
|
for dialect in all_dialects(exclude=("sybase",)):
|
|
bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
|
|
self.assert_compile(
|
|
func.current_timestamp(), "CURRENT_TIMESTAMP", dialect=dialect
|
|
)
|
|
self.assert_compile(func.localtime(), "LOCALTIME", dialect=dialect)
|
|
if dialect.name in ("firebird",):
|
|
self.assert_compile(
|
|
func.nosuchfunction(), "nosuchfunction", dialect=dialect
|
|
)
|
|
else:
|
|
self.assert_compile(
|
|
func.nosuchfunction(), "nosuchfunction()", dialect=dialect
|
|
)
|
|
|
|
# test generic function compile
|
|
class fake_func(GenericFunction):
|
|
__return_type__ = sqltypes.Integer
|
|
|
|
def __init__(self, arg, **kwargs):
|
|
GenericFunction.__init__(self, arg, **kwargs)
|
|
|
|
self.assert_compile(
|
|
fake_func("foo"),
|
|
"fake_func(%s)"
|
|
% bindtemplate
|
|
% {"name": "fake_func_1", "position": 1},
|
|
dialect=dialect,
|
|
)
|
|
|
|
functions._registry["_default"].pop("fake_func")
|
|
|
|
def test_use_labels(self):
|
|
self.assert_compile(
|
|
select(func.foo()).apply_labels(), "SELECT foo() AS foo_1"
|
|
)
|
|
|
|
def test_use_labels_function_element(self):
|
|
from sqlalchemy.ext.compiler import compiles
|
|
|
|
class max_(FunctionElement):
|
|
name = "max"
|
|
|
|
@compiles(max_)
|
|
def visit_max(element, compiler, **kw):
|
|
return "max(%s)" % compiler.process(element.clauses, **kw)
|
|
|
|
self.assert_compile(
|
|
select(max_(5, 6)).apply_labels(),
|
|
"SELECT max(:max_2, :max_3) AS max_1",
|
|
)
|
|
|
|
def test_underscores(self):
|
|
self.assert_compile(func.if_(), "if()")
|
|
|
|
def test_underscores_packages(self):
|
|
self.assert_compile(func.foo_.bar_.if_(), "foo.bar.if()")
|
|
|
|
def test_uppercase(self):
|
|
# for now, we need to keep case insensitivity
|
|
self.assert_compile(func.UNREGISTERED_FN(), "UNREGISTERED_FN()")
|
|
|
|
def test_uppercase_packages(self):
|
|
# for now, we need to keep case insensitivity
|
|
self.assert_compile(func.FOO.BAR.NOW(), "FOO.BAR.NOW()")
|
|
|
|
def test_mixed_case(self):
|
|
# for now, we need to keep case insensitivity
|
|
self.assert_compile(func.SomeFunction(), "SomeFunction()")
|
|
|
|
def test_mixed_case_packages(self):
|
|
# for now, we need to keep case insensitivity
|
|
self.assert_compile(
|
|
func.Foo.Bar.SomeFunction(), "Foo.Bar.SomeFunction()"
|
|
)
|
|
|
|
def test_quote_special_chars(self):
|
|
# however we need to be quoting any other identifiers
|
|
self.assert_compile(
|
|
getattr(func, "im a function")(), '"im a function"()'
|
|
)
|
|
|
|
def test_quote_special_chars_packages(self):
|
|
# however we need to be quoting any other identifiers
|
|
self.assert_compile(
|
|
getattr(
|
|
getattr(getattr(func, "im foo package"), "im bar package"),
|
|
"im a function",
|
|
)(),
|
|
'"im foo package"."im bar package"."im a function"()',
|
|
)
|
|
|
|
def test_generic_now(self):
|
|
assert isinstance(func.now().type, sqltypes.DateTime)
|
|
|
|
for ret, dialect in [
|
|
("CURRENT_TIMESTAMP", sqlite.dialect()),
|
|
("now()", postgresql.dialect()),
|
|
("now()", mysql.dialect()),
|
|
("CURRENT_TIMESTAMP", oracle.dialect()),
|
|
]:
|
|
self.assert_compile(func.now(), ret, dialect=dialect)
|
|
|
|
def test_generic_random(self):
|
|
assert func.random().type == sqltypes.NULLTYPE
|
|
assert isinstance(func.random(type_=Integer).type, Integer)
|
|
|
|
for ret, dialect in [
|
|
("random()", sqlite.dialect()),
|
|
("random()", postgresql.dialect()),
|
|
("rand()", mysql.dialect()),
|
|
("random()", oracle.dialect()),
|
|
]:
|
|
self.assert_compile(func.random(), ret, dialect=dialect)
|
|
|
|
def test_cube_operators(self):
|
|
|
|
t = table(
|
|
"t",
|
|
column("value"),
|
|
column("x"),
|
|
column("y"),
|
|
column("z"),
|
|
column("q"),
|
|
)
|
|
|
|
stmt = select(func.sum(t.c.value))
|
|
|
|
self.assert_compile(
|
|
stmt.group_by(func.cube(t.c.x, t.c.y)),
|
|
"SELECT sum(t.value) AS sum_1 FROM t GROUP BY CUBE(t.x, t.y)",
|
|
)
|
|
|
|
self.assert_compile(
|
|
stmt.group_by(func.rollup(t.c.x, t.c.y)),
|
|
"SELECT sum(t.value) AS sum_1 FROM t GROUP BY ROLLUP(t.x, t.y)",
|
|
)
|
|
|
|
self.assert_compile(
|
|
stmt.group_by(func.grouping_sets(t.c.x, t.c.y)),
|
|
"SELECT sum(t.value) AS sum_1 FROM t "
|
|
"GROUP BY GROUPING SETS(t.x, t.y)",
|
|
)
|
|
|
|
self.assert_compile(
|
|
stmt.group_by(
|
|
func.grouping_sets(
|
|
sql.tuple_(t.c.x, t.c.y), sql.tuple_(t.c.z, t.c.q)
|
|
)
|
|
),
|
|
"SELECT sum(t.value) AS sum_1 FROM t GROUP BY "
|
|
"GROUPING SETS((t.x, t.y), (t.z, t.q))",
|
|
)
|
|
|
|
def test_generic_annotation(self):
|
|
fn = func.coalesce("x", "y")._annotate({"foo": "bar"})
|
|
self.assert_compile(fn, "coalesce(:coalesce_1, :coalesce_2)")
|
|
|
|
def test_custom_default_namespace(self):
|
|
class myfunc(GenericFunction):
|
|
pass
|
|
|
|
assert isinstance(func.myfunc(), myfunc)
|
|
self.assert_compile(func.myfunc(), "myfunc()")
|
|
|
|
def test_custom_type(self):
|
|
class myfunc(GenericFunction):
|
|
type = DateTime
|
|
|
|
assert isinstance(func.myfunc().type, DateTime)
|
|
self.assert_compile(func.myfunc(), "myfunc()")
|
|
|
|
def test_custom_legacy_type(self):
|
|
# in case someone was using this system
|
|
class myfunc(GenericFunction):
|
|
__return_type__ = DateTime
|
|
|
|
assert isinstance(func.myfunc().type, DateTime)
|
|
|
|
def test_case_sensitive(self):
|
|
class MYFUNC(GenericFunction):
|
|
type = DateTime
|
|
|
|
assert isinstance(func.MYFUNC().type, DateTime)
|
|
assert isinstance(func.MyFunc().type, DateTime)
|
|
assert isinstance(func.mYfUnC().type, DateTime)
|
|
assert isinstance(func.myfunc().type, DateTime)
|
|
|
|
def test_replace_function(self):
|
|
class replaceable_func(GenericFunction):
|
|
type = Integer
|
|
identifier = "replaceable_func"
|
|
|
|
assert isinstance(func.Replaceable_Func().type, Integer)
|
|
assert isinstance(func.RePlAcEaBlE_fUnC().type, Integer)
|
|
assert isinstance(func.replaceable_func().type, Integer)
|
|
|
|
with expect_warnings(
|
|
"The GenericFunction 'replaceable_func' is already registered and "
|
|
"is going to be overridden.",
|
|
regex=False,
|
|
):
|
|
|
|
class replaceable_func_override(GenericFunction):
|
|
type = DateTime
|
|
identifier = "replaceable_func"
|
|
|
|
assert isinstance(func.Replaceable_Func().type, DateTime)
|
|
assert isinstance(func.RePlAcEaBlE_fUnC().type, DateTime)
|
|
assert isinstance(func.replaceable_func().type, DateTime)
|
|
|
|
def test_replace_function_case_insensitive(self):
|
|
class replaceable_func(GenericFunction):
|
|
type = Integer
|
|
identifier = "replaceable_func"
|
|
|
|
assert isinstance(func.Replaceable_Func().type, Integer)
|
|
assert isinstance(func.RePlAcEaBlE_fUnC().type, Integer)
|
|
assert isinstance(func.replaceable_func().type, Integer)
|
|
|
|
with expect_warnings(
|
|
"The GenericFunction 'replaceable_func' is already registered and "
|
|
"is going to be overridden.",
|
|
regex=False,
|
|
):
|
|
|
|
class replaceable_func_override(GenericFunction):
|
|
type = DateTime
|
|
identifier = "REPLACEABLE_Func"
|
|
|
|
assert isinstance(func.Replaceable_Func().type, DateTime)
|
|
assert isinstance(func.RePlAcEaBlE_fUnC().type, DateTime)
|
|
assert isinstance(func.replaceable_func().type, DateTime)
|
|
|
|
def test_custom_w_custom_name(self):
|
|
class myfunc(GenericFunction):
|
|
name = "notmyfunc"
|
|
|
|
assert isinstance(func.notmyfunc(), myfunc)
|
|
assert not isinstance(func.myfunc(), myfunc)
|
|
|
|
def test_custom_w_quoted_name(self):
|
|
class myfunc(GenericFunction):
|
|
name = quoted_name("NotMyFunc", quote=True)
|
|
identifier = "myfunc"
|
|
|
|
self.assert_compile(func.myfunc(), '"NotMyFunc"()')
|
|
|
|
def test_custom_w_quoted_name_no_identifier(self):
|
|
class myfunc(GenericFunction):
|
|
name = quoted_name("NotMyFunc", quote=True)
|
|
|
|
# note this requires that the quoted name be lower cased for
|
|
# correct lookup
|
|
self.assert_compile(func.notmyfunc(), '"NotMyFunc"()')
|
|
|
|
def test_custom_package_namespace(self):
|
|
def cls1(pk_name):
|
|
class myfunc(GenericFunction):
|
|
package = pk_name
|
|
|
|
return myfunc
|
|
|
|
f1 = cls1("mypackage")
|
|
f2 = cls1("myotherpackage")
|
|
|
|
assert isinstance(func.mypackage.myfunc(), f1)
|
|
assert isinstance(func.myotherpackage.myfunc(), f2)
|
|
|
|
def test_custom_name(self):
|
|
class MyFunction(GenericFunction):
|
|
name = "my_func"
|
|
|
|
def __init__(self, *args):
|
|
args = args + (3,)
|
|
super(MyFunction, self).__init__(*args)
|
|
|
|
self.assert_compile(
|
|
func.my_func(1, 2), "my_func(:my_func_1, :my_func_2, :my_func_3)"
|
|
)
|
|
|
|
def test_custom_registered_identifier(self):
|
|
class GeoBuffer(GenericFunction):
|
|
type = Integer
|
|
package = "geo"
|
|
name = "BufferOne"
|
|
identifier = "buf1"
|
|
|
|
class GeoBuffer2(GenericFunction):
|
|
type = Integer
|
|
name = "BufferTwo"
|
|
identifier = "buf2"
|
|
|
|
class BufferThree(GenericFunction):
|
|
type = Integer
|
|
identifier = "buf3"
|
|
|
|
class GeoBufferFour(GenericFunction):
|
|
type = Integer
|
|
name = "BufferFour"
|
|
identifier = "Buf4"
|
|
|
|
self.assert_compile(func.geo.buf1(), "BufferOne()")
|
|
self.assert_compile(func.buf2(), "BufferTwo()")
|
|
self.assert_compile(func.buf3(), "BufferThree()")
|
|
self.assert_compile(func.Buf4(), "BufferFour()")
|
|
self.assert_compile(func.BuF4(), "BufferFour()")
|
|
self.assert_compile(func.bUf4(), "BufferFour()")
|
|
self.assert_compile(func.bUf4_(), "BufferFour()")
|
|
self.assert_compile(func.buf4(), "BufferFour()")
|
|
|
|
def test_custom_args(self):
|
|
class myfunc(GenericFunction):
|
|
pass
|
|
|
|
self.assert_compile(
|
|
myfunc(1, 2, 3), "myfunc(:myfunc_1, :myfunc_2, :myfunc_3)"
|
|
)
|
|
|
|
def test_namespacing_conflicts(self):
|
|
self.assert_compile(func.text("foo"), "text(:text_1)")
|
|
|
|
def test_generic_count(self):
|
|
assert isinstance(func.count().type, sqltypes.Integer)
|
|
|
|
self.assert_compile(func.count(), "count(*)")
|
|
self.assert_compile(func.count(1), "count(:count_1)")
|
|
c = column("abc")
|
|
self.assert_compile(func.count(c), "count(abc)")
|
|
|
|
def test_ansi_functions_with_args(self):
|
|
ct = func.current_timestamp("somearg")
|
|
self.assert_compile(ct, "CURRENT_TIMESTAMP(:current_timestamp_1)")
|
|
|
|
def test_char_length_fixed_args(self):
|
|
assert_raises(TypeError, func.char_length, "a", "b")
|
|
assert_raises(TypeError, func.char_length)
|
|
|
|
def test_return_type_detection(self):
|
|
|
|
for fn in [func.coalesce, func.max, func.min, func.sum]:
|
|
for args, type_ in [
|
|
(
|
|
(datetime.date(2007, 10, 5), datetime.date(2005, 10, 15)),
|
|
sqltypes.Date,
|
|
),
|
|
((3, 5), sqltypes.Integer),
|
|
((decimal.Decimal(3), decimal.Decimal(5)), sqltypes.Numeric),
|
|
(("foo", "bar"), sqltypes.String),
|
|
(
|
|
(
|
|
datetime.datetime(2007, 10, 5, 8, 3, 34),
|
|
datetime.datetime(2005, 10, 15, 14, 45, 33),
|
|
),
|
|
sqltypes.DateTime,
|
|
),
|
|
]:
|
|
assert isinstance(fn(*args).type, type_), "%s / %r != %s" % (
|
|
fn(),
|
|
fn(*args).type,
|
|
type_,
|
|
)
|
|
|
|
assert isinstance(func.concat("foo", "bar").type, sqltypes.String)
|
|
|
|
def test_assorted(self):
|
|
table1 = table("mytable", column("myid", Integer))
|
|
|
|
table2 = table("myothertable", column("otherid", Integer))
|
|
|
|
# test an expression with a function
|
|
self.assert_compile(
|
|
func.lala(3, 4, literal("five"), table1.c.myid) * table2.c.otherid,
|
|
"lala(:lala_1, :lala_2, :param_1, mytable.myid) * "
|
|
"myothertable.otherid",
|
|
)
|
|
|
|
# test it in a SELECT
|
|
self.assert_compile(
|
|
select(func.count(table1.c.myid)),
|
|
"SELECT count(mytable.myid) AS count_1 FROM mytable",
|
|
)
|
|
|
|
# test a "dotted" function name
|
|
self.assert_compile(
|
|
select(func.foo.bar.lala(table1.c.myid)),
|
|
"SELECT foo.bar.lala(mytable.myid) AS lala_1 FROM mytable",
|
|
)
|
|
|
|
# test the bind parameter name with a "dotted" function name is
|
|
# only the name (limits the length of the bind param name)
|
|
self.assert_compile(
|
|
select(func.foo.bar.lala(12)),
|
|
"SELECT foo.bar.lala(:lala_2) AS lala_1",
|
|
)
|
|
|
|
# test a dotted func off the engine itself
|
|
self.assert_compile(func.lala.hoho(7), "lala.hoho(:hoho_1)")
|
|
|
|
# test None becomes NULL
|
|
self.assert_compile(
|
|
func.my_func(1, 2, None, 3),
|
|
"my_func(:my_func_1, :my_func_2, NULL, :my_func_3)",
|
|
)
|
|
|
|
f1 = func.my_func(1, 2, None, 3)
|
|
f1._generate_cache_key()
|
|
|
|
# test pickling
|
|
self.assert_compile(
|
|
util.pickle.loads(util.pickle.dumps(f1)),
|
|
"my_func(:my_func_1, :my_func_2, NULL, :my_func_3)",
|
|
)
|
|
|
|
# assert func raises AttributeError for __bases__ attribute, since
|
|
# its not a class fixes pydoc
|
|
try:
|
|
func.__bases__
|
|
assert False
|
|
except AttributeError:
|
|
assert True
|
|
|
|
def test_pickle_over(self):
|
|
# TODO: the test/sql package lacks a comprehensive pickling
|
|
# test suite even though there are __reduce__ methods in several
|
|
# places in sql/elements.py. likely as part of
|
|
# test/sql/test_compare.py might be a place this can happen but
|
|
# this still relies upon a strategy for table metadata as we have
|
|
# in serializer.
|
|
|
|
f1 = func.row_number().over()
|
|
|
|
self.assert_compile(
|
|
util.pickle.loads(util.pickle.dumps(f1)),
|
|
"row_number() OVER ()",
|
|
)
|
|
|
|
def test_functions_with_cols(self):
|
|
users = table(
|
|
"users", column("id"), column("name"), column("fullname")
|
|
)
|
|
calculate = (
|
|
select(column("q"), column("z"), column("r"))
|
|
.select_from(
|
|
func.calculate(bindparam("x", None), bindparam("y", None))
|
|
)
|
|
.subquery()
|
|
)
|
|
|
|
self.assert_compile(
|
|
select(users).where(users.c.id > calculate.c.z),
|
|
"SELECT users.id, users.name, users.fullname "
|
|
"FROM users, (SELECT q, z, r "
|
|
"FROM calculate(:x, :y)) AS anon_1 "
|
|
"WHERE users.id > anon_1.z",
|
|
)
|
|
|
|
s = select(users).where(
|
|
users.c.id.between(
|
|
calculate.alias("c1").unique_params(x=17, y=45).c.z,
|
|
calculate.alias("c2").unique_params(x=5, y=12).c.z,
|
|
),
|
|
)
|
|
|
|
self.assert_compile(
|
|
s,
|
|
"SELECT users.id, users.name, users.fullname "
|
|
"FROM users, (SELECT q, z, r "
|
|
"FROM calculate(:x_1, :y_1)) AS c1, (SELECT q, z, r "
|
|
"FROM calculate(:x_2, :y_2)) AS c2 "
|
|
"WHERE users.id BETWEEN c1.z AND c2.z",
|
|
checkparams={"y_1": 45, "x_1": 17, "y_2": 12, "x_2": 5},
|
|
)
|
|
|
|
def test_non_functions(self):
|
|
expr = func.cast("foo", Integer)
|
|
self.assert_compile(expr, "CAST(:param_1 AS INTEGER)")
|
|
|
|
expr = func.extract("year", datetime.date(2010, 12, 5))
|
|
self.assert_compile(expr, "EXTRACT(year FROM :param_1)")
|
|
|
|
def test_select_method_one(self):
|
|
expr = func.rows("foo")
|
|
self.assert_compile(expr.select(), "SELECT rows(:rows_2) AS rows_1")
|
|
|
|
def test_alias_method_one(self):
|
|
expr = func.rows("foo")
|
|
self.assert_compile(expr.alias(), "rows(:rows_1)")
|
|
|
|
def test_select_method_two(self):
|
|
expr = func.rows("foo")
|
|
self.assert_compile(
|
|
select("*").select_from(expr.select().subquery()),
|
|
"SELECT * FROM (SELECT rows(:rows_2) AS rows_1) AS anon_1",
|
|
)
|
|
|
|
def test_select_method_three(self):
|
|
expr = func.rows("foo")
|
|
self.assert_compile(
|
|
select(column("foo")).select_from(expr),
|
|
"SELECT foo FROM rows(:rows_1)",
|
|
)
|
|
|
|
def test_alias_method_two(self):
|
|
expr = func.rows("foo")
|
|
self.assert_compile(
|
|
select("*").select_from(expr.alias("bar")),
|
|
"SELECT * FROM rows(:rows_1) AS bar",
|
|
)
|
|
|
|
def test_alias_method_columns(self):
|
|
expr = func.rows("foo").alias("bar")
|
|
|
|
# this isn't very useful but is the old behavior
|
|
# prior to #2974.
|
|
# testing here that the expression exports its column
|
|
# list in a way that at least doesn't break.
|
|
self.assert_compile(
|
|
select(expr), "SELECT bar.rows_1 FROM rows(:rows_2) AS bar"
|
|
)
|
|
|
|
def test_alias_method_columns_two(self):
|
|
expr = func.rows("foo").alias("bar")
|
|
assert len(expr.c)
|
|
|
|
def test_funcfilter_empty(self):
|
|
self.assert_compile(func.count(1).filter(), "count(:count_1)")
|
|
|
|
def test_funcfilter_criterion(self):
|
|
self.assert_compile(
|
|
func.count(1).filter(table1.c.name != None), # noqa
|
|
"count(:count_1) FILTER (WHERE mytable.name IS NOT NULL)",
|
|
)
|
|
|
|
def test_funcfilter_compound_criterion(self):
|
|
self.assert_compile(
|
|
func.count(1).filter(
|
|
table1.c.name == None, table1.c.myid > 0 # noqa
|
|
),
|
|
"count(:count_1) FILTER (WHERE mytable.name IS NULL AND "
|
|
"mytable.myid > :myid_1)",
|
|
)
|
|
|
|
def test_funcfilter_arrayagg_subscript(self):
|
|
num = column("q")
|
|
self.assert_compile(
|
|
func.array_agg(num).filter(num % 2 == 0)[1],
|
|
"(array_agg(q) FILTER (WHERE q %% %(q_1)s = "
|
|
"%(param_1)s))[%(param_2)s]",
|
|
dialect="postgresql",
|
|
)
|
|
|
|
def test_funcfilter_label(self):
|
|
self.assert_compile(
|
|
select(
|
|
func.count(1)
|
|
.filter(table1.c.description != None) # noqa
|
|
.label("foo")
|
|
),
|
|
"SELECT count(:count_1) FILTER (WHERE mytable.description "
|
|
"IS NOT NULL) AS foo FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_fromobj_fromfunc(self):
|
|
# test from_obj generation.
|
|
# from func:
|
|
self.assert_compile(
|
|
select(
|
|
func.max(table1.c.name).filter(
|
|
literal_column("description") != None # noqa
|
|
)
|
|
),
|
|
"SELECT max(mytable.name) FILTER (WHERE description "
|
|
"IS NOT NULL) AS anon_1 FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_fromobj_fromcriterion(self):
|
|
# from criterion:
|
|
self.assert_compile(
|
|
select(func.count(1).filter(table1.c.name == "name")),
|
|
"SELECT count(:count_1) FILTER (WHERE mytable.name = :name_1) "
|
|
"AS anon_1 FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_chaining(self):
|
|
# test chaining:
|
|
self.assert_compile(
|
|
select(
|
|
func.count(1)
|
|
.filter(table1.c.name == "name")
|
|
.filter(table1.c.description == "description")
|
|
),
|
|
"SELECT count(:count_1) FILTER (WHERE "
|
|
"mytable.name = :name_1 AND mytable.description = :description_1) "
|
|
"AS anon_1 FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_windowing_orderby(self):
|
|
# test filtered windowing:
|
|
self.assert_compile(
|
|
select(
|
|
func.rank()
|
|
.filter(table1.c.name > "foo")
|
|
.over(order_by=table1.c.name)
|
|
),
|
|
"SELECT rank() FILTER (WHERE mytable.name > :name_1) "
|
|
"OVER (ORDER BY mytable.name) AS anon_1 FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_windowing_orderby_partitionby(self):
|
|
self.assert_compile(
|
|
select(
|
|
func.rank()
|
|
.filter(table1.c.name > "foo")
|
|
.over(order_by=table1.c.name, partition_by=["description"])
|
|
),
|
|
"SELECT rank() FILTER (WHERE mytable.name > :name_1) "
|
|
"OVER (PARTITION BY mytable.description ORDER BY mytable.name) "
|
|
"AS anon_1 FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_windowing_range(self):
|
|
self.assert_compile(
|
|
select(
|
|
func.rank()
|
|
.filter(table1.c.name > "foo")
|
|
.over(range_=(1, 5), partition_by=["description"])
|
|
),
|
|
"SELECT rank() FILTER (WHERE mytable.name > :name_1) "
|
|
"OVER (PARTITION BY mytable.description RANGE BETWEEN :param_1 "
|
|
"FOLLOWING AND :param_2 FOLLOWING) "
|
|
"AS anon_1 FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_windowing_rows(self):
|
|
self.assert_compile(
|
|
select(
|
|
func.rank()
|
|
.filter(table1.c.name > "foo")
|
|
.over(rows=(1, 5), partition_by=["description"])
|
|
),
|
|
"SELECT rank() FILTER (WHERE mytable.name > :name_1) "
|
|
"OVER (PARTITION BY mytable.description ROWS BETWEEN :param_1 "
|
|
"FOLLOWING AND :param_2 FOLLOWING) "
|
|
"AS anon_1 FROM mytable",
|
|
)
|
|
|
|
def test_funcfilter_within_group(self):
|
|
stmt = select(
|
|
table1.c.myid,
|
|
func.percentile_cont(0.5).within_group(table1.c.name),
|
|
)
|
|
self.assert_compile(
|
|
stmt,
|
|
"SELECT mytable.myid, percentile_cont(:percentile_cont_1) "
|
|
"WITHIN GROUP (ORDER BY mytable.name) "
|
|
"AS anon_1 "
|
|
"FROM mytable",
|
|
{"percentile_cont_1": 0.5},
|
|
)
|
|
|
|
def test_funcfilter_within_group_multi(self):
|
|
stmt = select(
|
|
table1.c.myid,
|
|
func.percentile_cont(0.5).within_group(
|
|
table1.c.name, table1.c.description
|
|
),
|
|
)
|
|
self.assert_compile(
|
|
stmt,
|
|
"SELECT mytable.myid, percentile_cont(:percentile_cont_1) "
|
|
"WITHIN GROUP (ORDER BY mytable.name, mytable.description) "
|
|
"AS anon_1 "
|
|
"FROM mytable",
|
|
{"percentile_cont_1": 0.5},
|
|
)
|
|
|
|
def test_funcfilter_within_group_desc(self):
|
|
stmt = select(
|
|
table1.c.myid,
|
|
func.percentile_cont(0.5).within_group(table1.c.name.desc()),
|
|
)
|
|
self.assert_compile(
|
|
stmt,
|
|
"SELECT mytable.myid, percentile_cont(:percentile_cont_1) "
|
|
"WITHIN GROUP (ORDER BY mytable.name DESC) "
|
|
"AS anon_1 "
|
|
"FROM mytable",
|
|
{"percentile_cont_1": 0.5},
|
|
)
|
|
|
|
def test_funcfilter_within_group_w_over(self):
|
|
stmt = select(
|
|
table1.c.myid,
|
|
func.percentile_cont(0.5)
|
|
.within_group(table1.c.name.desc())
|
|
.over(partition_by=table1.c.description),
|
|
)
|
|
self.assert_compile(
|
|
stmt,
|
|
"SELECT mytable.myid, percentile_cont(:percentile_cont_1) "
|
|
"WITHIN GROUP (ORDER BY mytable.name DESC) "
|
|
"OVER (PARTITION BY mytable.description) AS anon_1 "
|
|
"FROM mytable",
|
|
{"percentile_cont_1": 0.5},
|
|
)
|
|
|
|
def test_incorrect_none_type(self):
|
|
from sqlalchemy.sql.expression import FunctionElement
|
|
|
|
class MissingType(FunctionElement):
|
|
name = "mt"
|
|
type = None
|
|
|
|
assert_raises_message(
|
|
TypeError,
|
|
"Object None associated with '.type' attribute is "
|
|
"not a TypeEngine class or object",
|
|
lambda: column("x", MissingType()) == 5,
|
|
)
|
|
|
|
def test_as_comparison(self):
|
|
|
|
fn = func.substring("foo", "foobar").as_comparison(1, 2)
|
|
is_(fn.type._type_affinity, Boolean)
|
|
|
|
self.assert_compile(
|
|
fn.left, ":substring_1", checkparams={"substring_1": "foo"}
|
|
)
|
|
self.assert_compile(
|
|
fn.right, ":substring_1", checkparams={"substring_1": "foobar"}
|
|
)
|
|
|
|
self.assert_compile(
|
|
fn,
|
|
"substring(:substring_1, :substring_2)",
|
|
checkparams={"substring_1": "foo", "substring_2": "foobar"},
|
|
)
|
|
|
|
def test_as_comparison_annotate(self):
|
|
|
|
fn = func.foobar("x", "y", "q", "p", "r").as_comparison(2, 5)
|
|
|
|
from sqlalchemy.sql import annotation
|
|
|
|
fn_annotated = annotation._deep_annotate(fn, {"token": "yes"})
|
|
|
|
eq_(fn.left._annotations, {})
|
|
eq_(fn_annotated.left._annotations, {"token": "yes"})
|
|
|
|
def test_as_comparison_many_argument(self):
|
|
|
|
fn = func.some_comparison("x", "y", "z", "p", "q", "r").as_comparison(
|
|
2, 5
|
|
)
|
|
is_(fn.type._type_affinity, Boolean)
|
|
|
|
self.assert_compile(
|
|
fn.left,
|
|
":some_comparison_1",
|
|
checkparams={"some_comparison_1": "y"},
|
|
)
|
|
self.assert_compile(
|
|
fn.right,
|
|
":some_comparison_1",
|
|
checkparams={"some_comparison_1": "q"},
|
|
)
|
|
|
|
from sqlalchemy.sql import visitors
|
|
|
|
fn_2 = visitors.cloned_traverse(fn, {}, {})
|
|
fn_2.right = literal_column("ABC")
|
|
|
|
self.assert_compile(
|
|
fn,
|
|
"some_comparison(:some_comparison_1, :some_comparison_2, "
|
|
":some_comparison_3, "
|
|
":some_comparison_4, :some_comparison_5, :some_comparison_6)",
|
|
checkparams={
|
|
"some_comparison_1": "x",
|
|
"some_comparison_2": "y",
|
|
"some_comparison_3": "z",
|
|
"some_comparison_4": "p",
|
|
"some_comparison_5": "q",
|
|
"some_comparison_6": "r",
|
|
},
|
|
)
|
|
|
|
self.assert_compile(
|
|
fn_2,
|
|
"some_comparison(:some_comparison_1, :some_comparison_2, "
|
|
":some_comparison_3, "
|
|
":some_comparison_4, ABC, :some_comparison_5)",
|
|
checkparams={
|
|
"some_comparison_1": "x",
|
|
"some_comparison_2": "y",
|
|
"some_comparison_3": "z",
|
|
"some_comparison_4": "p",
|
|
"some_comparison_5": "r",
|
|
},
|
|
)
|
|
|
|
|
|
class ReturnTypeTest(AssertsCompiledSQL, fixtures.TestBase):
|
|
def test_array_agg(self):
|
|
expr = func.array_agg(column("data", Integer))
|
|
is_(expr.type._type_affinity, ARRAY)
|
|
is_(expr.type.item_type._type_affinity, Integer)
|
|
|
|
def test_array_agg_array_datatype(self):
|
|
expr = func.array_agg(column("data", ARRAY(Integer)))
|
|
is_(expr.type._type_affinity, ARRAY)
|
|
is_(expr.type.item_type._type_affinity, Integer)
|
|
|
|
def test_array_agg_array_literal_implicit_type(self):
|
|
from sqlalchemy.dialects.postgresql import array, ARRAY as PG_ARRAY
|
|
|
|
expr = array([column("data", Integer), column("d2", Integer)])
|
|
|
|
assert isinstance(expr.type, PG_ARRAY)
|
|
|
|
agg_expr = func.array_agg(expr)
|
|
assert isinstance(agg_expr.type, PG_ARRAY)
|
|
is_(agg_expr.type._type_affinity, ARRAY)
|
|
is_(agg_expr.type.item_type._type_affinity, Integer)
|
|
|
|
self.assert_compile(
|
|
agg_expr, "array_agg(ARRAY[data, d2])", dialect="postgresql"
|
|
)
|
|
|
|
def test_array_agg_array_literal_explicit_type(self):
|
|
from sqlalchemy.dialects.postgresql import array
|
|
|
|
expr = array([column("data", Integer), column("d2", Integer)])
|
|
|
|
agg_expr = func.array_agg(expr, type_=ARRAY(Integer))
|
|
is_(agg_expr.type._type_affinity, ARRAY)
|
|
is_(agg_expr.type.item_type._type_affinity, Integer)
|
|
|
|
self.assert_compile(
|
|
agg_expr, "array_agg(ARRAY[data, d2])", dialect="postgresql"
|
|
)
|
|
|
|
def test_mode(self):
|
|
expr = func.mode(0.5).within_group(column("data", Integer).desc())
|
|
is_(expr.type._type_affinity, Integer)
|
|
|
|
def test_percentile_cont(self):
|
|
expr = func.percentile_cont(0.5).within_group(column("data", Integer))
|
|
is_(expr.type._type_affinity, Integer)
|
|
|
|
def test_percentile_cont_array(self):
|
|
expr = func.percentile_cont(0.5, 0.7).within_group(
|
|
column("data", Integer)
|
|
)
|
|
is_(expr.type._type_affinity, ARRAY)
|
|
is_(expr.type.item_type._type_affinity, Integer)
|
|
|
|
def test_percentile_cont_array_desc(self):
|
|
expr = func.percentile_cont(0.5, 0.7).within_group(
|
|
column("data", Integer).desc()
|
|
)
|
|
is_(expr.type._type_affinity, ARRAY)
|
|
is_(expr.type.item_type._type_affinity, Integer)
|
|
|
|
def test_cume_dist(self):
|
|
expr = func.cume_dist(0.5).within_group(column("data", Integer).desc())
|
|
is_(expr.type._type_affinity, Numeric)
|
|
|
|
def test_percent_rank(self):
|
|
expr = func.percent_rank(0.5).within_group(column("data", Integer))
|
|
is_(expr.type._type_affinity, Numeric)
|
|
|
|
|
|
class ExecuteTest(fixtures.TestBase):
|
|
__backend__ = True
|
|
|
|
def teardown_test(self):
|
|
pass
|
|
|
|
def test_conn_execute(self, connection):
|
|
from sqlalchemy.sql.expression import FunctionElement
|
|
from sqlalchemy.ext.compiler import compiles
|
|
|
|
class myfunc(FunctionElement):
|
|
type = Date()
|
|
|
|
@compiles(myfunc)
|
|
def compile_(elem, compiler, **kw):
|
|
return compiler.process(func.current_date())
|
|
|
|
x = connection.execute(func.current_date()).scalar()
|
|
y = connection.execute(func.current_date().select()).scalar()
|
|
z = connection.scalar(func.current_date())
|
|
q = connection.scalar(myfunc())
|
|
|
|
assert (x == y == z == q) is True
|
|
|
|
def test_exec_options(self, connection):
|
|
f = func.foo()
|
|
eq_(f._execution_options, {})
|
|
|
|
f = f.execution_options(foo="bar")
|
|
eq_(f._execution_options, {"foo": "bar"})
|
|
s = f.select()
|
|
eq_(s._execution_options, {"foo": "bar"})
|
|
|
|
ret = connection.execute(func.now().execution_options(foo="bar"))
|
|
eq_(ret.context.execution_options, {"foo": "bar"})
|
|
ret.close()
|
|
|
|
@testing.provide_metadata
|
|
def test_update(self, connection):
|
|
"""
|
|
Tests sending functions and SQL expressions to the VALUES and SET
|
|
clauses of INSERT/UPDATE instances, and that column-level defaults
|
|
get overridden.
|
|
"""
|
|
|
|
meta = self.metadata
|
|
t = Table(
|
|
"t1",
|
|
meta,
|
|
Column(
|
|
"id",
|
|
Integer,
|
|
Sequence("t1idseq", optional=True),
|
|
primary_key=True,
|
|
),
|
|
Column("value", Integer),
|
|
)
|
|
t2 = Table(
|
|
"t2",
|
|
meta,
|
|
Column(
|
|
"id",
|
|
Integer,
|
|
Sequence("t2idseq", optional=True),
|
|
primary_key=True,
|
|
),
|
|
Column("value", Integer, default=7),
|
|
Column("stuff", String(20), onupdate="thisisstuff"),
|
|
)
|
|
meta.create_all(connection)
|
|
connection.execute(t.insert().values(value=func.length("one")))
|
|
eq_(connection.execute(t.select()).first().value, 3)
|
|
connection.execute(t.update().values(value=func.length("asfda")))
|
|
eq_(connection.execute(t.select()).first().value, 5)
|
|
|
|
r = connection.execute(
|
|
t.insert().values(value=func.length("sfsaafsda"))
|
|
)
|
|
id_ = r.inserted_primary_key[0]
|
|
eq_(
|
|
connection.execute(t.select().where(t.c.id == id_)).first().value,
|
|
9,
|
|
)
|
|
connection.execute(t.update().values({t.c.value: func.length("asdf")}))
|
|
eq_(connection.execute(t.select()).first().value, 4)
|
|
connection.execute(t2.insert())
|
|
connection.execute(t2.insert().values(value=func.length("one")))
|
|
connection.execute(
|
|
t2.insert().values(value=func.length("asfda") + -19),
|
|
stuff="hi",
|
|
)
|
|
|
|
res = sorted(connection.execute(select(t2.c.value, t2.c.stuff)))
|
|
eq_(res, [(-14, "hi"), (3, None), (7, None)])
|
|
|
|
connection.execute(
|
|
t2.update().values(value=func.length("asdsafasd")),
|
|
stuff="some stuff",
|
|
)
|
|
eq_(
|
|
connection.execute(select(t2.c.value, t2.c.stuff)).fetchall(),
|
|
[(9, "some stuff"), (9, "some stuff"), (9, "some stuff")],
|
|
)
|
|
|
|
connection.execute(t2.delete())
|
|
|
|
connection.execute(t2.insert().values(value=func.length("one") + 8))
|
|
eq_(connection.execute(t2.select()).first().value, 11)
|
|
|
|
connection.execute(t2.update().values(value=func.length("asfda")))
|
|
eq_(
|
|
connection.execute(select(t2.c.value, t2.c.stuff)).first(),
|
|
(5, "thisisstuff"),
|
|
)
|
|
|
|
connection.execute(
|
|
t2.update().values(
|
|
{t2.c.value: func.length("asfdaasdf"), t2.c.stuff: "foo"}
|
|
)
|
|
)
|
|
|
|
eq_(
|
|
connection.execute(select(t2.c.value, t2.c.stuff)).first(),
|
|
(9, "foo"),
|
|
)
|
|
|
|
@testing.fails_on_everything_except("postgresql")
|
|
def test_as_from(self, connection):
|
|
# TODO: shouldn't this work on oracle too ?
|
|
x = connection.execute(func.current_date()).scalar()
|
|
y = connection.execute(func.current_date().select()).scalar()
|
|
z = connection.scalar(func.current_date())
|
|
w = connection.scalar(select("*").select_from(func.current_date()))
|
|
|
|
assert x == y == z == w
|
|
|
|
def test_extract_bind(self, connection):
|
|
"""Basic common denominator execution tests for extract()"""
|
|
|
|
date = datetime.date(2010, 5, 1)
|
|
|
|
def execute(field):
|
|
return connection.execute(select(extract(field, date))).scalar()
|
|
|
|
assert execute("year") == 2010
|
|
assert execute("month") == 5
|
|
assert execute("day") == 1
|
|
|
|
date = datetime.datetime(2010, 5, 1, 12, 11, 10)
|
|
|
|
assert execute("year") == 2010
|
|
assert execute("month") == 5
|
|
assert execute("day") == 1
|
|
|
|
@testing.provide_metadata
|
|
def test_extract_expression(self, connection):
|
|
meta = self.metadata
|
|
table = Table("test", meta, Column("dt", DateTime), Column("d", Date))
|
|
meta.create_all(connection)
|
|
connection.execute(
|
|
table.insert(),
|
|
{
|
|
"dt": datetime.datetime(2010, 5, 1, 12, 11, 10),
|
|
"d": datetime.date(2010, 5, 1),
|
|
},
|
|
)
|
|
rs = connection.execute(
|
|
select(extract("year", table.c.dt), extract("month", table.c.d))
|
|
)
|
|
row = rs.first()
|
|
assert row[0] == 2010
|
|
assert row[1] == 5
|
|
rs.close()
|
|
|
|
|
|
class RegisterTest(fixtures.TestBase, AssertsCompiledSQL):
|
|
__dialect__ = "default"
|
|
|
|
def setup_test(self):
|
|
self._registry = deepcopy(functions._registry)
|
|
|
|
def teardown_test(self):
|
|
functions._registry = self._registry
|
|
|
|
def test_GenericFunction_is_registered(self):
|
|
assert "GenericFunction" not in functions._registry["_default"]
|
|
|
|
def test_register_function(self):
|
|
|
|
# test generic function registering
|
|
class registered_func(GenericFunction):
|
|
_register = True
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
GenericFunction.__init__(self, *args, **kwargs)
|
|
|
|
class registered_func_child(registered_func):
|
|
type = sqltypes.Integer
|
|
|
|
assert "registered_func" in functions._registry["_default"]
|
|
assert isinstance(func.registered_func_child().type, Integer)
|
|
|
|
class not_registered_func(GenericFunction):
|
|
_register = False
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
GenericFunction.__init__(self, *args, **kwargs)
|
|
|
|
class not_registered_func_child(not_registered_func):
|
|
type = sqltypes.Integer
|
|
|
|
assert "not_registered_func" not in functions._registry["_default"]
|
|
assert isinstance(func.not_registered_func_child().type, Integer)
|