diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-07-25 16:04:35 -0400 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-07-25 16:04:35 -0400 |
| commit | 31178db91455ead5bfd4269658073c745e090569 (patch) | |
| tree | 1db071c0e830d28e03e5ca55d07a7e2131b5935a /test/dialect/postgresql/test_query.py | |
| parent | 6b60d3a9e6ba93d177ac777bfaae8269c18ddee6 (diff) | |
| download | sqlalchemy-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.py | 454 |
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} + ) |
