summaryrefslogtreecommitdiff
path: root/test/dialect/postgresql/test_query.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2014-07-25 16:04:35 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2014-07-25 16:04:35 -0400
commit31178db91455ead5bfd4269658073c745e090569 (patch)
tree1db071c0e830d28e03e5ca55d07a7e2131b5935a /test/dialect/postgresql/test_query.py
parent6b60d3a9e6ba93d177ac777bfaae8269c18ddee6 (diff)
downloadsqlalchemy-31178db91455ead5bfd4269658073c745e090569.tar.gz
- flake8 all of test/dialect/postgresql
- add __backend__ to most tests so that pg8000 can start coming in
Diffstat (limited to 'test/dialect/postgresql/test_query.py')
-rw-r--r--test/dialect/postgresql/test_query.py454
1 files changed, 255 insertions, 199 deletions
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index 064f0c372..a512b56fa 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -1,23 +1,23 @@
# coding: utf-8
-from sqlalchemy.testing.assertions import eq_, assert_raises, \
- assert_raises_message, is_, AssertsExecutionResults, \
- AssertsCompiledSQL, ComparesTables
+from sqlalchemy.testing import AssertsExecutionResults, eq_, \
+ assert_raises_message, AssertsCompiledSQL
+from sqlalchemy import Table, Column, MetaData, Integer, String, bindparam, \
+ Sequence, ForeignKey, text, select, func, extract, literal_column, \
+ tuple_, DateTime, Time, literal, and_, Date, or_
from sqlalchemy.testing import engines, fixtures
from sqlalchemy import testing
-from sqlalchemy import Table, Column, select, MetaData, text, Integer, \
- String, Sequence, ForeignKey, join, Numeric, \
- PrimaryKeyConstraint, DateTime, tuple_, Float, BigInteger, \
- func, literal_column, literal, bindparam, cast, extract, \
- SmallInteger, Enum, REAL, update, insert, Index, delete, \
- and_, Date, TypeDecorator, Time, Unicode, Interval, or_, Text
from sqlalchemy import exc
from sqlalchemy.dialects import postgresql
import datetime
+metadata = matchtable = cattable = None
+
+
class InsertTest(fixtures.TestBase, AssertsExecutionResults):
__only_on__ = 'postgresql'
+ __backend__ = True
@classmethod
def setup_class(cls):
@@ -32,20 +32,32 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
self.engine.dispose()
def test_compiled_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- primary_key=True), Column('data', String(30)))
+ table = Table(
+ 'testtable', metadata, Column(
+ 'id', Integer, primary_key=True),
+ Column(
+ 'data', String(30)))
metadata.create_all()
- ins = table.insert(inline=True, values={'data': bindparam('x'
- )}).compile()
+ ins = table.insert(
+ inline=True,
+ values={'data': bindparam('x')}).compile()
ins.execute({'x': 'five'}, {'x': 'seven'})
- assert table.select().execute().fetchall() == [(1, 'five'), (2,
- 'seven')]
+ eq_(
+ table.select().execute().fetchall(),
+ [(1, 'five'), (2, 'seven')]
+ )
def test_foreignkey_missing_insert(self):
t1 = Table('t1', metadata, Column('id', Integer,
- primary_key=True))
- t2 = Table('t2', metadata, Column('id', Integer,
- ForeignKey('t1.id'), primary_key=True))
+ primary_key=True))
+ t2 = Table(
+ 't2',
+ metadata,
+ Column(
+ 'id',
+ Integer,
+ ForeignKey('t1.id'),
+ primary_key=True))
metadata.create_all()
# want to ensure that "null value in column "id" violates not-
@@ -55,68 +67,107 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
# the latter corresponds to autoincrement behavior, which is not
# the case here due to the foreign key.
- for eng in [engines.testing_engine(options={'implicit_returning'
- : False}),
- engines.testing_engine(options={'implicit_returning'
- : True})]:
+ for eng in [
+ engines.testing_engine(options={'implicit_returning': False}),
+ engines.testing_engine(options={'implicit_returning': True})
+ ]:
assert_raises_message(exc.DBAPIError,
'violates not-null constraint',
eng.execute, t2.insert())
def test_sequence_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- Sequence('my_seq'), primary_key=True),
- Column('data', String(30)))
+ table = Table(
+ 'testtable',
+ metadata,
+ Column(
+ 'id',
+ Integer,
+ Sequence('my_seq'),
+ primary_key=True),
+ Column(
+ 'data',
+ String(30)))
metadata.create_all()
self._assert_data_with_sequence(table, 'my_seq')
@testing.requires.returning
def test_sequence_returning_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- Sequence('my_seq'), primary_key=True),
- Column('data', String(30)))
+ table = Table(
+ 'testtable',
+ metadata,
+ Column(
+ 'id',
+ Integer,
+ Sequence('my_seq'),
+ primary_key=True),
+ Column(
+ 'data',
+ String(30)))
metadata.create_all()
self._assert_data_with_sequence_returning(table, 'my_seq')
def test_opt_sequence_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- Sequence('my_seq', optional=True),
- primary_key=True), Column('data', String(30)))
+ table = Table(
+ 'testtable', metadata,
+ Column(
+ 'id', Integer, Sequence(
+ 'my_seq', optional=True), primary_key=True),
+ Column(
+ 'data', String(30)))
metadata.create_all()
self._assert_data_autoincrement(table)
@testing.requires.returning
def test_opt_sequence_returning_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- Sequence('my_seq', optional=True),
- primary_key=True), Column('data', String(30)))
+ table = Table(
+ 'testtable', metadata,
+ Column(
+ 'id', Integer, Sequence(
+ 'my_seq', optional=True), primary_key=True),
+ Column(
+ 'data', String(30)))
metadata.create_all()
self._assert_data_autoincrement_returning(table)
def test_autoincrement_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- primary_key=True), Column('data', String(30)))
+ table = Table(
+ 'testtable', metadata,
+ Column(
+ 'id', Integer, primary_key=True),
+ Column(
+ 'data', String(30)))
metadata.create_all()
self._assert_data_autoincrement(table)
@testing.requires.returning
def test_autoincrement_returning_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- primary_key=True), Column('data', String(30)))
+ table = Table(
+ 'testtable', metadata,
+ Column(
+ 'id', Integer, primary_key=True),
+ Column(
+ 'data', String(30)))
metadata.create_all()
self._assert_data_autoincrement_returning(table)
def test_noautoincrement_insert(self):
- table = Table('testtable', metadata, Column('id', Integer,
- primary_key=True, autoincrement=False),
- Column('data', String(30)))
+ table = Table(
+ 'testtable',
+ metadata,
+ Column(
+ 'id',
+ Integer,
+ primary_key=True,
+ autoincrement=False),
+ Column(
+ 'data',
+ String(30)))
metadata.create_all()
self._assert_data_noautoincrement(table)
def _assert_data_autoincrement(self, table):
self.engine = \
- engines.testing_engine(options={'implicit_returning'
- : False})
+ engines.testing_engine(options={'implicit_returning': False})
metadata.bind = self.engine
def go():
@@ -134,7 +185,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
# executemany with explicit ids
table.insert().execute({'id': 31, 'data': 'd3'}, {'id': 32,
- 'data': 'd4'})
+ 'data': 'd4'})
# executemany, uses SERIAL
@@ -158,13 +209,12 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
{'id': 1, 'data': 'd2'}),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 31, 'data': 'd3'}, {'id': 32, 'data': 'd4'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd5'}, {'data': 'd6'}]),
+ ('INSERT INTO testtable (data) VALUES (:data)',
+ [{'data': 'd5'}, {'data': 'd6'}]),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 33, 'data': 'd7'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd8'}]),
- ])
+ ('INSERT INTO testtable (data) VALUES (:data)', [{'data': 'd8'}]),
+ ])
assert table.select().execute().fetchall() == [
(30, 'd1'),
(1, 'd2'),
@@ -174,7 +224,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
(3, 'd6'),
(33, 'd7'),
(4, 'd8'),
- ]
+ ]
table.delete().execute()
# test the same series of events using a reflected version of
@@ -188,7 +238,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
r = table.insert().execute({'data': 'd2'})
assert r.inserted_primary_key == [5]
table.insert().execute({'id': 31, 'data': 'd3'}, {'id': 32,
- 'data': 'd4'})
+ 'data': 'd4'})
table.insert().execute({'data': 'd5'}, {'data': 'd6'})
table.insert(inline=True).execute({'id': 33, 'data': 'd7'})
table.insert(inline=True).execute({'data': 'd8'})
@@ -200,13 +250,12 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
{'id': 5, 'data': 'd2'}),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 31, 'data': 'd3'}, {'id': 32, 'data': 'd4'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd5'}, {'data': 'd6'}]),
+ ('INSERT INTO testtable (data) VALUES (:data)',
+ [{'data': 'd5'}, {'data': 'd6'}]),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 33, 'data': 'd7'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd8'}]),
- ])
+ ('INSERT INTO testtable (data) VALUES (:data)', [{'data': 'd8'}]),
+ ])
assert table.select().execute().fetchall() == [
(30, 'd1'),
(5, 'd2'),
@@ -216,7 +265,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
(7, 'd6'),
(33, 'd7'),
(8, 'd8'),
- ]
+ ]
table.delete().execute()
def _assert_data_autoincrement_returning(self, table):
@@ -239,7 +288,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
# executemany with explicit ids
table.insert().execute({'id': 31, 'data': 'd3'}, {'id': 32,
- 'data': 'd4'})
+ 'data': 'd4'})
# executemany, uses SERIAL
@@ -260,13 +309,12 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
'testtable.id', {'data': 'd2'}),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 31, 'data': 'd3'}, {'id': 32, 'data': 'd4'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd5'}, {'data': 'd6'}]),
+ ('INSERT INTO testtable (data) VALUES (:data)',
+ [{'data': 'd5'}, {'data': 'd6'}]),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 33, 'data': 'd7'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd8'}]),
- ])
+ ('INSERT INTO testtable (data) VALUES (:data)', [{'data': 'd8'}]),
+ ])
assert table.select().execute().fetchall() == [
(30, 'd1'),
(1, 'd2'),
@@ -276,7 +324,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
(3, 'd6'),
(33, 'd7'),
(4, 'd8'),
- ]
+ ]
table.delete().execute()
# test the same series of events using a reflected version of
@@ -290,7 +338,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
r = table.insert().execute({'data': 'd2'})
assert r.inserted_primary_key == [5]
table.insert().execute({'id': 31, 'data': 'd3'}, {'id': 32,
- 'data': 'd4'})
+ 'data': 'd4'})
table.insert().execute({'data': 'd5'}, {'data': 'd6'})
table.insert(inline=True).execute({'id': 33, 'data': 'd7'})
table.insert(inline=True).execute({'data': 'd8'})
@@ -302,13 +350,12 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
'testtable.id', {'data': 'd2'}),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 31, 'data': 'd3'}, {'id': 32, 'data': 'd4'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd5'}, {'data': 'd6'}]),
+ ('INSERT INTO testtable (data) VALUES (:data)',
+ [{'data': 'd5'}, {'data': 'd6'}]),
('INSERT INTO testtable (id, data) VALUES (:id, :data)',
[{'id': 33, 'data': 'd7'}]),
- ('INSERT INTO testtable (data) VALUES (:data)', [{'data'
- : 'd8'}]),
- ])
+ ('INSERT INTO testtable (data) VALUES (:data)', [{'data': 'd8'}]),
+ ])
assert table.select().execute().fetchall() == [
(30, 'd1'),
(5, 'd2'),
@@ -318,20 +365,19 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
(7, 'd6'),
(33, 'd7'),
(8, 'd8'),
- ]
+ ]
table.delete().execute()
def _assert_data_with_sequence(self, table, seqname):
self.engine = \
- engines.testing_engine(options={'implicit_returning'
- : False})
+ engines.testing_engine(options={'implicit_returning': False})
metadata.bind = self.engine
def go():
table.insert().execute({'id': 30, 'data': 'd1'})
table.insert().execute({'data': 'd2'})
table.insert().execute({'id': 31, 'data': 'd3'}, {'id': 32,
- 'data': 'd4'})
+ 'data': 'd4'})
table.insert().execute({'data': 'd5'}, {'data': 'd6'})
table.insert(inline=True).execute({'id': 33, 'data': 'd7'})
table.insert(inline=True).execute({'data': 'd8'})
@@ -349,7 +395,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
[{'id': 33, 'data': 'd7'}]),
("INSERT INTO testtable (id, data) VALUES (nextval('%s'), "
":data)" % seqname, [{'data': 'd8'}]),
- ])
+ ])
assert table.select().execute().fetchall() == [
(30, 'd1'),
(1, 'd2'),
@@ -359,7 +405,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
(3, 'd6'),
(33, 'd7'),
(4, 'd8'),
- ]
+ ]
# cant test reflection here since the Sequence must be
# explicitly specified
@@ -373,7 +419,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
table.insert().execute({'id': 30, 'data': 'd1'})
table.insert().execute({'data': 'd2'})
table.insert().execute({'id': 31, 'data': 'd3'}, {'id': 32,
- 'data': 'd4'})
+ 'data': 'd4'})
table.insert().execute({'data': 'd5'}, {'data': 'd6'})
table.insert(inline=True).execute({'id': 33, 'data': 'd7'})
table.insert(inline=True).execute({'data': 'd8'})
@@ -392,7 +438,7 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
[{'id': 33, 'data': 'd7'}]),
("INSERT INTO testtable (id, data) VALUES (nextval('%s'), "
":data)" % seqname, [{'data': 'd8'}]),
- ])
+ ])
assert table.select().execute().fetchall() == [
(30, 'd1'),
(1, 'd2'),
@@ -402,15 +448,14 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
(3, 'd6'),
(33, 'd7'),
(4, 'd8'),
- ]
+ ]
# cant test reflection here since the Sequence must be
# explicitly specified
def _assert_data_noautoincrement(self, table):
self.engine = \
- engines.testing_engine(options={'implicit_returning'
- : False})
+ engines.testing_engine(options={'implicit_returning': False})
metadata.bind = self.engine
table.insert().execute({'id': 30, 'data': 'd1'})
if self.engine.driver == 'pg8000':
@@ -434,10 +479,13 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
table.insert().execute, {'data': 'd2'},
{'data': 'd3'})
table.insert().execute({'id': 31, 'data': 'd2'}, {'id': 32,
- 'data': 'd3'})
+ 'data': 'd3'})
table.insert(inline=True).execute({'id': 33, 'data': 'd4'})
- assert table.select().execute().fetchall() == [(30, 'd1'), (31,
- 'd2'), (32, 'd3'), (33, 'd4')]
+ assert table.select().execute().fetchall() == [
+ (30, 'd1'),
+ (31, 'd2'),
+ (32, 'd3'),
+ (33, 'd4')]
table.delete().execute()
# test the same series of events using a reflected version of
@@ -454,10 +502,13 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
table.insert().execute, {'data': 'd2'},
{'data': 'd3'})
table.insert().execute({'id': 31, 'data': 'd2'}, {'id': 32,
- 'data': 'd3'})
+ 'data': 'd3'})
table.insert(inline=True).execute({'id': 33, 'data': 'd4'})
- assert table.select().execute().fetchall() == [(30, 'd1'), (31,
- 'd2'), (32, 'd3'), (33, 'd4')]
+ assert table.select().execute().fetchall() == [
+ (30, 'd1'),
+ (31, 'd2'),
+ (32, 'd3'),
+ (33, 'd4')]
class ServerSideCursorsTest(fixtures.TestBase, AssertsExecutionResults):
@@ -466,8 +517,8 @@ class ServerSideCursorsTest(fixtures.TestBase, AssertsExecutionResults):
def _fixture(self, server_side_cursors):
self.engine = engines.testing_engine(
- options={'server_side_cursors':server_side_cursors}
- )
+ options={'server_side_cursors': server_side_cursors}
+ )
return self.engine
def tearDown(self):
@@ -507,15 +558,14 @@ class ServerSideCursorsTest(fixtures.TestBase, AssertsExecutionResults):
assert result.cursor.name
-
def test_conn_option(self):
engine = self._fixture(False)
# and this one
result = \
engine.connect().execution_options(stream_results=True).\
- execute('select 1'
- )
+ execute('select 1'
+ )
assert result.cursor.name
def test_stmt_enabled_conn_option_disabled(self):
@@ -526,7 +576,7 @@ class ServerSideCursorsTest(fixtures.TestBase, AssertsExecutionResults):
# not this one
result = \
engine.connect().execution_options(stream_results=False).\
- execute(s)
+ execute(s)
assert not result.cursor.name
def test_stmt_option_disabled(self):
@@ -581,12 +631,13 @@ class ServerSideCursorsTest(fixtures.TestBase, AssertsExecutionResults):
nextid = engine.execute(Sequence('test_table_id_seq'))
test_table.insert().execute(id=nextid, data='data2')
eq_(test_table.select().execute().fetchall(), [(1, 'data1'
- ), (2, 'data2')])
- test_table.update().where(test_table.c.id
- == 2).values(data=test_table.c.data + ' updated'
- ).execute()
- eq_(test_table.select().execute().fetchall(), [(1, 'data1'
- ), (2, 'data2 updated')])
+ ), (2, 'data2')])
+ test_table.update().where(
+ test_table.c.id == 2).values(
+ data=test_table.c.data +
+ ' updated').execute()
+ eq_(test_table.select().execute().fetchall(),
+ [(1, 'data1'), (2, 'data2 updated')])
test_table.delete().execute()
eq_(test_table.count().scalar(), 0)
finally:
@@ -596,36 +647,37 @@ class ServerSideCursorsTest(fixtures.TestBase, AssertsExecutionResults):
class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
__only_on__ = 'postgresql >= 8.3'
+ __backend__ = True
@classmethod
def setup_class(cls):
global metadata, cattable, matchtable
metadata = MetaData(testing.db)
- cattable = Table('cattable', metadata, Column('id', Integer,
- primary_key=True), Column('description',
- String(50)))
- matchtable = Table('matchtable', metadata, Column('id',
- Integer, primary_key=True), Column('title',
- String(200)), Column('category_id', Integer,
- ForeignKey('cattable.id')))
+ cattable = Table(
+ 'cattable', metadata,
+ Column(
+ 'id', Integer, primary_key=True),
+ Column(
+ 'description', String(50)))
+ matchtable = Table(
+ 'matchtable', metadata,
+ Column(
+ 'id', Integer, primary_key=True),
+ Column(
+ 'title', String(200)),
+ Column(
+ 'category_id', Integer, ForeignKey('cattable.id')))
metadata.create_all()
cattable.insert().execute([{'id': 1, 'description': 'Python'},
- {'id': 2, 'description': 'Ruby'}])
- matchtable.insert().execute([{'id': 1, 'title'
- : 'Agile Web Development with Rails'
- , 'category_id': 2},
- {'id': 2,
- 'title': 'Dive Into Python',
- 'category_id': 1},
- {'id': 3, 'title'
- : "Programming Matz's Ruby",
- 'category_id': 2},
- {'id': 4, 'title'
- : 'The Definitive Guide to Django',
- 'category_id': 1},
- {'id': 5, 'title'
- : 'Python in a Nutshell',
- 'category_id': 1}])
+ {'id': 2, 'description': 'Ruby'}])
+ matchtable.insert().execute(
+ [{'id': 1, 'title': 'Agile Web Development with Rails',
+ 'category_id': 2},
+ {'id': 2, 'title': 'Dive Into Python', 'category_id': 1},
+ {'id': 3, 'title': "Programming Matz's Ruby", 'category_id': 2},
+ {'id': 4, 'title': 'The Definitive Guide to Django',
+ 'category_id': 1},
+ {'id': 5, 'title': 'Python in a Nutshell', 'category_id': 1}])
@classmethod
def teardown_class(cls):
@@ -646,58 +698,58 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
'matchtable.title @@ to_tsquery(%s)')
def test_simple_match(self):
- results = \
- matchtable.select().where(matchtable.c.title.match('python'
- )).order_by(matchtable.c.id).execute().fetchall()
+ results = matchtable.select().where(
+ matchtable.c.title.match('python')).order_by(
+ matchtable.c.id).execute().fetchall()
eq_([2, 5], [r.id for r in results])
def test_simple_match_with_apostrophe(self):
- results = \
- matchtable.select().where(matchtable.c.title.match("Matz's"
- )).execute().fetchall()
+ results = matchtable.select().where(
+ matchtable.c.title.match("Matz's")).execute().fetchall()
eq_([3], [r.id for r in results])
def test_simple_derivative_match(self):
- results = \
- matchtable.select().where(matchtable.c.title.match('nutshells'
- )).execute().fetchall()
+ results = matchtable.select().where(
+ matchtable.c.title.match('nutshells')).execute().fetchall()
eq_([5], [r.id for r in results])
def test_or_match(self):
- results1 = \
- matchtable.select().where(or_(matchtable.c.title.match('nutshells'
- ), matchtable.c.title.match('rubies'
- ))).order_by(matchtable.c.id).execute().fetchall()
+ results1 = matchtable.select().where(
+ or_(
+ matchtable.c.title.match('nutshells'),
+ matchtable.c.title.match('rubies'))).order_by(
+ matchtable.c.id).execute().fetchall()
eq_([3, 5], [r.id for r in results1])
- results2 = \
- matchtable.select().where(
- matchtable.c.title.match('nutshells | rubies'
- )).order_by(matchtable.c.id).execute().fetchall()
+ results2 = matchtable.select().where(
+ matchtable.c.title.match('nutshells | rubies')).order_by(
+ matchtable.c.id).execute().fetchall()
eq_([3, 5], [r.id for r in results2])
def test_and_match(self):
- results1 = \
- matchtable.select().where(and_(matchtable.c.title.match('python'
- ), matchtable.c.title.match('nutshells'
- ))).execute().fetchall()
+ results1 = matchtable.select().where(
+ and_(
+ matchtable.c.title.match('python'),
+ matchtable.c.title.match('nutshells'))).execute().fetchall()
eq_([5], [r.id for r in results1])
results2 = \
matchtable.select().where(
matchtable.c.title.match('python & nutshells'
- )).execute().fetchall()
+ )).execute().fetchall()
eq_([5], [r.id for r in results2])
def test_match_across_joins(self):
- results = matchtable.select().where(and_(cattable.c.id
- == matchtable.c.category_id,
- or_(cattable.c.description.match('Ruby'),
- matchtable.c.title.match('nutshells'
- )))).order_by(matchtable.c.id).execute().fetchall()
+ results = matchtable.select().where(
+ and_(
+ cattable.c.id == matchtable.c.category_id, or_(
+ cattable.c.description.match('Ruby'),
+ matchtable.c.title.match('nutshells')))).order_by(
+ matchtable.c.id).execute().fetchall()
eq_([1, 3, 5], [r.id for r in results])
class TupleTest(fixtures.TestBase):
__only_on__ = 'postgresql'
+ __backend__ = True
def test_tuple_containment(self):
@@ -710,24 +762,24 @@ class TupleTest(fixtures.TestBase):
eq_(
testing.db.execute(
select([
- tuple_(
- literal_column("'a'"),
- literal_column("'b'")
- ).\
- in_([
- tuple_(*[
- literal_column("'%s'" % letter)
- for letter in elem
- ]) for elem in test
- ])
- ])
+ tuple_(
+ literal_column("'a'"),
+ literal_column("'b'")
+ ).
+ in_([
+ tuple_(*[
+ literal_column("'%s'" % letter)
+ for letter in elem
+ ]) for elem in test
+ ])
+ ])
).scalar(),
exp
)
-
class ExtractTest(fixtures.TablesTest):
+
"""The rationale behind this test is that for many years we've had a system
of embedding type casts into the expressions rendered by visit_extract()
on the postgreql platform. The reason for this cast is not clear.
@@ -736,6 +788,7 @@ class ExtractTest(fixtures.TablesTest):
"""
__only_on__ = 'postgresql'
+ __backend__ = True
run_inserts = 'once'
run_deletes = None
@@ -743,19 +796,20 @@ class ExtractTest(fixtures.TablesTest):
@classmethod
def define_tables(cls, metadata):
Table('t', metadata,
- Column('id', Integer, primary_key=True),
- Column('dtme', DateTime),
- Column('dt', Date),
- Column('tm', Time),
- Column('intv', postgresql.INTERVAL),
- Column('dttz', DateTime(timezone=True))
- )
+ Column('id', Integer, primary_key=True),
+ Column('dtme', DateTime),
+ Column('dt', Date),
+ Column('tm', Time),
+ Column('intv', postgresql.INTERVAL),
+ Column('dttz', DateTime(timezone=True))
+ )
@classmethod
def insert_data(cls):
# TODO: why does setting hours to anything
# not affect the TZ in the DB col ?
class TZ(datetime.tzinfo):
+
def utcoffset(self, dt):
return datetime.timedelta(hours=4)
@@ -780,18 +834,18 @@ class ExtractTest(fixtures.TablesTest):
if field == "all":
fields = {"year": 2012, "month": 5, "day": 10,
- "epoch": 1336652125.0,
- "hour": 12, "minute": 15}
+ "epoch": 1336652125.0,
+ "hour": 12, "minute": 15}
elif field == "time":
fields = {"hour": 12, "minute": 15, "second": 25}
elif field == 'date':
fields = {"year": 2012, "month": 5, "day": 10}
elif field == 'all+tz':
fields = {"year": 2012, "month": 5, "day": 10,
- "epoch": 1336637725.0,
- "hour": 8,
- "timezone": 0
- }
+ "epoch": 1336637725.0,
+ "hour": 8,
+ "timezone": 0
+ }
else:
fields = field
@@ -800,7 +854,7 @@ class ExtractTest(fixtures.TablesTest):
for field in fields:
result = testing.db.scalar(
- select([extract(field, expr)]).select_from(t))
+ select([extract(field, expr)]).select_from(t))
eq_(result, fields[field])
def test_one(self):
@@ -810,45 +864,45 @@ class ExtractTest(fixtures.TablesTest):
def test_two(self):
t = self.tables.t
self._test(t.c.dtme + t.c.intv,
- overrides={"epoch": 1336652695.0, "minute": 24})
+ overrides={"epoch": 1336652695.0, "minute": 24})
def test_three(self):
t = self.tables.t
actual_ts = testing.db.scalar(func.current_timestamp()) - \
- datetime.timedelta(days=5)
+ datetime.timedelta(days=5)
self._test(func.current_timestamp() - datetime.timedelta(days=5),
- {"hour": actual_ts.hour, "year": actual_ts.year,
- "month": actual_ts.month}
- )
+ {"hour": actual_ts.hour, "year": actual_ts.year,
+ "month": actual_ts.month}
+ )
def test_four(self):
t = self.tables.t
self._test(datetime.timedelta(days=5) + t.c.dt,
- overrides={"day": 15, "epoch": 1337040000.0, "hour": 0,
- "minute": 0}
- )
+ overrides={"day": 15, "epoch": 1337040000.0, "hour": 0,
+ "minute": 0}
+ )
def test_five(self):
t = self.tables.t
self._test(func.coalesce(t.c.dtme, func.current_timestamp()),
- overrides={"epoch": 1336652125.0})
+ overrides={"epoch": 1336652125.0})
def test_six(self):
t = self.tables.t
self._test(t.c.tm + datetime.timedelta(seconds=30), "time",
- overrides={"second": 55})
+ overrides={"second": 55})
def test_seven(self):
self._test(literal(datetime.timedelta(seconds=10))
- - literal(datetime.timedelta(seconds=10)), "all",
- overrides={"hour": 0, "minute": 0, "month": 0,
- "year": 0, "day": 0, "epoch": 0})
+ - literal(datetime.timedelta(seconds=10)), "all",
+ overrides={"hour": 0, "minute": 0, "month": 0,
+ "year": 0, "day": 0, "epoch": 0})
def test_eight(self):
t = self.tables.t
self._test(t.c.tm + datetime.timedelta(seconds=30),
- {"hour": 12, "minute": 15, "second": 55})
+ {"hour": 12, "minute": 15, "second": 55})
def test_nine(self):
self._test(text("t.dt + t.tm"))
@@ -859,19 +913,21 @@ class ExtractTest(fixtures.TablesTest):
def test_eleven(self):
self._test(func.current_timestamp() - func.current_timestamp(),
- {"year": 0, "month": 0, "day": 0, "hour": 0}
- )
+ {"year": 0, "month": 0, "day": 0, "hour": 0}
+ )
def test_twelve(self):
t = self.tables.t
actual_ts = testing.db.scalar(
- func.current_timestamp()).replace(tzinfo=None) - \
- datetime.datetime(2012, 5, 10, 12, 15, 25)
+ func.current_timestamp()).replace(tzinfo=None) - \
+ datetime.datetime(2012, 5, 10, 12, 15, 25)
- self._test(func.current_timestamp() - func.coalesce(t.c.dtme,
- func.current_timestamp()),
- {"day": actual_ts.days}
- )
+ self._test(
+ func.current_timestamp() - func.coalesce(
+ t.c.dtme,
+ func.current_timestamp()
+ ),
+ {"day": actual_ts.days})
def test_thirteen(self):
t = self.tables.t
@@ -884,5 +940,5 @@ class ExtractTest(fixtures.TablesTest):
def test_fifteen(self):
t = self.tables.t
self._test(datetime.timedelta(days=5) + t.c.dtme,
- overrides={"day": 15, "epoch": 1337084125.0}
- )
+ overrides={"day": 15, "epoch": 1337084125.0}
+ )