1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
import testbase
import datetime
from sqlalchemy import *
from sqlalchemy import exceptions, sql
from sqlalchemy.sql.compiler import BIND_TEMPLATES
from sqlalchemy.engine import default
from sqlalchemy import types as sqltypes
from testlib import *
# TODO: add a helper function to testlib for this
from sqlalchemy.databases import sqlite, postgres, mysql, oracle, firebird, mssql
dialects = [x.dialect() for x in [sqlite, postgres, mysql, oracle, firebird, mssql]]
class CompileTest(SQLCompileTest):
def test_compile(self):
for dialect in dialects:
bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
self.assert_compile(func.current_timestamp(), "CURRENT_TIMESTAMP", dialect=dialect)
self.assert_compile(func.localtime(), "LOCALTIME", dialect=dialect)
if isinstance(dialect, firebird.dialect):
self.assert_compile(func.nosuchfunction(), "nosuchfunction", dialect=dialect)
else:
self.assert_compile(func.nosuchfunction(), "nosuchfunction()", dialect=dialect)
self.assert_compile(func.char_length('foo'), "char_length(%s)" % bindtemplate % {'name':'param_1', 'position':1}, dialect=dialect)
def test_constructor(self):
try:
func.current_timestamp('somearg')
assert False
except TypeError:
assert True
try:
func.char_length('a', 'b')
assert False
except TypeError:
assert True
try:
func.char_length()
assert False
except TypeError:
assert True
def test_typing(self):
assert isinstance(func.coalesce(datetime.date(2007, 10, 5), datetime.date(2005, 10, 15)).type, sqltypes.Date)
assert isinstance(func.coalesce(None, datetime.date(2005, 10, 15)).type, sqltypes.Date)
assert isinstance(func.concat("foo", "bar").type, sqltypes.String)
class ExecuteTest(PersistTest):
def test_standalone_execute(self):
x = testbase.db.func.current_date().execute().scalar()
y = testbase.db.func.current_date().select().execute().scalar()
z = testbase.db.func.current_date().scalar()
assert (x == y == z) is True
# ansi func
x = testbase.db.func.current_date()
assert isinstance(x.type, Date)
assert isinstance(x.execute().scalar(), datetime.date)
def test_conn_execute(self):
conn = testbase.db.connect()
try:
x = conn.execute(func.current_date()).scalar()
y = conn.execute(func.current_date().select()).scalar()
z = conn.scalar(func.current_date())
finally:
conn.close()
assert (x == y == z) is True
def test_update(self):
"""
Tests sending functions and SQL expressions to the VALUES and SET
clauses of INSERT/UPDATE instances, and that column-level defaults
get overridden.
"""
meta = MetaData(testbase.db)
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()
try:
t.insert(values=dict(value=func.length("one"))).execute()
assert t.select().execute().fetchone()['value'] == 3
t.update(values=dict(value=func.length("asfda"))).execute()
assert t.select().execute().fetchone()['value'] == 5
r = t.insert(values=dict(value=func.length("sfsaafsda"))).execute()
id = r.last_inserted_ids()[0]
assert t.select(t.c.id==id).execute().fetchone()['value'] == 9
t.update(values={t.c.value:func.length("asdf")}).execute()
assert t.select().execute().fetchone()['value'] == 4
print "--------------------------"
t2.insert().execute()
t2.insert(values=dict(value=func.length("one"))).execute()
t2.insert(values=dict(value=func.length("asfda") + -19)).execute(stuff="hi")
res = exec_sorted(select([t2.c.value, t2.c.stuff]))
self.assertEquals(res, [(-14, 'hi'), (3, None), (7, None)])
t2.update(values=dict(value=func.length("asdsafasd"))).execute(stuff="some stuff")
assert select([t2.c.value, t2.c.stuff]).execute().fetchall() == [(9,"some stuff"), (9,"some stuff"), (9,"some stuff")]
t2.delete().execute()
t2.insert(values=dict(value=func.length("one") + 8)).execute()
assert t2.select().execute().fetchone()['value'] == 11
t2.update(values=dict(value=func.length("asfda"))).execute()
assert select([t2.c.value, t2.c.stuff]).execute().fetchone() == (5, "thisisstuff")
t2.update(values={t2.c.value:func.length("asfdaasdf"), t2.c.stuff:"foo"}).execute()
print "HI", select([t2.c.value, t2.c.stuff]).execute().fetchone()
assert select([t2.c.value, t2.c.stuff]).execute().fetchone() == (9, "foo")
finally:
meta.drop_all()
@testing.supported('postgres')
def test_as_from(self):
# TODO: shouldnt this work on oracle too ?
x = testbase.db.func.current_date().execute().scalar()
y = testbase.db.func.current_date().select().execute().scalar()
z = testbase.db.func.current_date().scalar()
w = select(['*'], from_obj=[testbase.db.func.current_date()]).scalar()
# construct a column-based FROM object out of a function, like in [ticket:172]
s = select([sql.column('date', type_=DateTime)], from_obj=[testbase.db.func.current_date()])
q = s.execute().fetchone()[s.c.date]
r = s.alias('datequery').select().scalar()
assert x == y == z == w == q == r
def exec_sorted(statement, *args, **kw):
"""Executes a statement and returns a sorted list plain tuple rows."""
return sorted([tuple(row)
for row in statement.execute(*args, **kw).fetchall()])
if __name__ == '__main__':
testbase.main()
|
