from sqlalchemy.testing import eq_, assert_raises_message from sqlalchemy import exc from sqlalchemy.databases import firebird from sqlalchemy.exc import ProgrammingError from sqlalchemy.sql import table, column from sqlalchemy import types as sqltypes from sqlalchemy.testing import fixtures, AssertsExecutionResults, AssertsCompiledSQL from sqlalchemy import testing from sqlalchemy.testing import engines from sqlalchemy import String, VARCHAR, NVARCHAR, Unicode, Integer,\ func, insert, update, MetaData, select, Table, Column, text,\ Sequence, Float from sqlalchemy import schema class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): "Test Firebird domains" __only_on__ = 'firebird' @classmethod def setup_class(cls): con = testing.db.connect() try: con.execute('CREATE DOMAIN int_domain AS INTEGER DEFAULT ' '42 NOT NULL') con.execute('CREATE DOMAIN str_domain AS VARCHAR(255)') con.execute('CREATE DOMAIN rem_domain AS BLOB SUB_TYPE TEXT' ) con.execute('CREATE DOMAIN img_domain AS BLOB SUB_TYPE ' 'BINARY') except ProgrammingError as e: if not 'attempt to store duplicate value' in str(e): raise e con.execute('''CREATE GENERATOR gen_testtable_id''') con.execute('''CREATE TABLE testtable (question int_domain, answer str_domain DEFAULT 'no answer', remark rem_domain DEFAULT '', photo img_domain, d date, t time, dt timestamp, redundant str_domain DEFAULT NULL)''') con.execute("ALTER TABLE testtable " "ADD CONSTRAINT testtable_pk PRIMARY KEY " "(question)") con.execute("CREATE TRIGGER testtable_autoid FOR testtable " " ACTIVE BEFORE INSERT AS" " BEGIN" " IF (NEW.question IS NULL) THEN" " NEW.question = gen_id(gen_testtable_id, 1);" " END") @classmethod def teardown_class(cls): con = testing.db.connect() con.execute('DROP TABLE testtable') con.execute('DROP DOMAIN int_domain') con.execute('DROP DOMAIN str_domain') con.execute('DROP DOMAIN rem_domain') con.execute('DROP DOMAIN img_domain') con.execute('DROP GENERATOR gen_testtable_id') def test_table_is_reflected(self): from sqlalchemy.types import Integer, Text, BLOB, String, Date, \ Time, DateTime metadata = MetaData(testing.db) table = Table('testtable', metadata, autoload=True) eq_(set(table.columns.keys()), set([ 'question', 'answer', 'remark', 'photo', 'd', 't', 'dt', 'redundant', ]), "Columns of reflected table didn't equal expected " "columns") eq_(table.c.question.primary_key, True) # disabled per http://www.sqlalchemy.org/trac/ticket/1660 # eq_(table.c.question.sequence.name, 'gen_testtable_id') assert isinstance(table.c.question.type, Integer) eq_(table.c.question.server_default.arg.text, '42') assert isinstance(table.c.answer.type, String) assert table.c.answer.type.length == 255 eq_(table.c.answer.server_default.arg.text, "'no answer'") assert isinstance(table.c.remark.type, Text) eq_(table.c.remark.server_default.arg.text, "''") assert isinstance(table.c.photo.type, BLOB) assert table.c.redundant.server_default is None # The following assume a Dialect 3 database assert isinstance(table.c.d.type, Date) assert isinstance(table.c.t.type, Time) assert isinstance(table.c.dt.type, DateTime) class BuggyDomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): """Test Firebird domains (and some other reflection bumps), see [ticket:1663] and http://tracker.firebirdsql.org/browse/CORE-356""" __only_on__ = 'firebird' # NB: spacing and newlines are *significant* here! # PS: this test is superfluous on recent FB, where the issue 356 is probably fixed... AUTOINC_DM = """\ CREATE DOMAIN AUTOINC_DM AS NUMERIC(18,0) """ MONEY_DM = """\ CREATE DOMAIN MONEY_DM AS NUMERIC(15,2) DEFAULT 0 CHECK (VALUE BETWEEN - 9999999999999.99 AND +9999999999999.99) """ NOSI_DM = """\ CREATE DOMAIN NOSI_DM AS CHAR(1) DEFAULT 'N' NOT NULL CHECK (VALUE IN ('S', 'N')) """ RIT_TESORERIA_CAPITOLO_DM = """\ CREATE DOMAIN RIT_TESORERIA_CAPITOLO_DM AS VARCHAR(6) CHECK ((VALUE IS NULL) OR (VALUE = UPPER(VALUE))) """ DEF_ERROR_TB = """\ CREATE TABLE DEF_ERROR ( RITENUTAMOV_ID AUTOINC_DM NOT NULL, RITENUTA MONEY_DM, INTERESSI MONEY_DM DEFAULT 0, STAMPATO_MODULO NOSI_DM DEFAULT 'S', TESORERIA_CAPITOLO RIT_TESORERIA_CAPITOLO_DM) """ DEF_ERROR_NODOM_TB = """\ CREATE TABLE DEF_ERROR_NODOM ( RITENUTAMOV_ID INTEGER NOT NULL, RITENUTA NUMERIC(15,2) DEFAULT 0, INTERESSI NUMERIC(15,2) DEFAULT 0, STAMPATO_MODULO CHAR(1) DEFAULT 'S', TESORERIA_CAPITOLO CHAR(1)) """ DOM_ID = """ CREATE DOMAIN DOM_ID INTEGER NOT NULL """ TABLE_A = """\ CREATE TABLE A ( ID DOM_ID /* INTEGER NOT NULL */ DEFAULT 0 ) """ # the 'default' keyword is lower case here TABLE_B = """\ CREATE TABLE B ( ID DOM_ID /* INTEGER NOT NULL */ default 0 ) """ @classmethod def setup_class(cls): con = testing.db.connect() con.execute(cls.AUTOINC_DM) con.execute(cls.MONEY_DM) con.execute(cls.NOSI_DM) con.execute(cls.RIT_TESORERIA_CAPITOLO_DM) con.execute(cls.DEF_ERROR_TB) con.execute(cls.DEF_ERROR_NODOM_TB) con.execute(cls.DOM_ID) con.execute(cls.TABLE_A) con.execute(cls.TABLE_B) @classmethod def teardown_class(cls): con = testing.db.connect() con.execute('DROP TABLE a') con.execute("DROP TABLE b") con.execute('DROP DOMAIN dom_id') con.execute('DROP TABLE def_error_nodom') con.execute('DROP TABLE def_error') con.execute('DROP DOMAIN rit_tesoreria_capitolo_dm') con.execute('DROP DOMAIN nosi_dm') con.execute('DROP DOMAIN money_dm') con.execute('DROP DOMAIN autoinc_dm') def test_tables_are_reflected_same_way(self): metadata = MetaData(testing.db) table_dom = Table('def_error', metadata, autoload=True) table_nodom = Table('def_error_nodom', metadata, autoload=True) eq_(table_dom.c.interessi.server_default.arg.text, table_nodom.c.interessi.server_default.arg.text) eq_(table_dom.c.ritenuta.server_default.arg.text, table_nodom.c.ritenuta.server_default.arg.text) eq_(table_dom.c.stampato_modulo.server_default.arg.text, table_nodom.c.stampato_modulo.server_default.arg.text) def test_intermixed_comment(self): metadata = MetaData(testing.db) table_a = Table('a', metadata, autoload=True) eq_(table_a.c.id.server_default.arg.text, "0") def test_lowercase_default_name(self): metadata = MetaData(testing.db) table_b = Table('b', metadata, autoload=True) eq_(table_b.c.id.server_default.arg.text, "0") class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = firebird.FBDialect() def test_alias(self): t = table('sometable', column('col1'), column('col2')) s = select([t.alias()]) self.assert_compile(s, 'SELECT sometable_1.col1, sometable_1.col2 ' 'FROM sometable AS sometable_1') dialect = firebird.FBDialect() dialect._version_two = False self.assert_compile(s, 'SELECT sometable_1.col1, sometable_1.col2 ' 'FROM sometable sometable_1', dialect=dialect) def test_varchar_raise(self): for type_ in ( String, VARCHAR, String(), VARCHAR(), Unicode, Unicode(), ): type_ = sqltypes.to_instance(type_) assert_raises_message( exc.CompileError, "VARCHAR requires a length on dialect firebird", type_.compile, dialect=firebird.dialect()) t1 = Table('sometable', MetaData(), Column('somecolumn', type_) ) assert_raises_message( exc.CompileError, r"\(in table 'sometable', column 'somecolumn'\)\: " r"(?:N)?VARCHAR requires a length on dialect firebird", schema.CreateTable(t1).compile, dialect=firebird.dialect() ) def test_function(self): self.assert_compile(func.foo(1, 2), 'foo(:foo_1, :foo_2)') self.assert_compile(func.current_time(), 'CURRENT_TIME') self.assert_compile(func.foo(), 'foo') m = MetaData() t = Table('sometable', m, Column('col1', Integer), Column('col2' , Integer)) self.assert_compile(select([func.max(t.c.col1)]), 'SELECT max(sometable.col1) AS max_1 FROM ' 'sometable') def test_substring(self): self.assert_compile(func.substring('abc', 1, 2), 'SUBSTRING(:substring_1 FROM :substring_2 ' 'FOR :substring_3)') self.assert_compile(func.substring('abc', 1), 'SUBSTRING(:substring_1 FROM :substring_2)') def test_update_returning(self): table1 = table('mytable', column('myid', Integer), column('name' , String(128)), column('description', String(128))) u = update(table1, values=dict(name='foo' )).returning(table1.c.myid, table1.c.name) self.assert_compile(u, 'UPDATE mytable SET name=:name RETURNING ' 'mytable.myid, mytable.name') u = update(table1, values=dict(name='foo')).returning(table1) self.assert_compile(u, 'UPDATE mytable SET name=:name RETURNING ' 'mytable.myid, mytable.name, ' 'mytable.description') u = update(table1, values=dict(name='foo' )).returning(func.length(table1.c.name)) self.assert_compile(u, 'UPDATE mytable SET name=:name RETURNING ' 'char_length(mytable.name) AS length_1') def test_insert_returning(self): table1 = table('mytable', column('myid', Integer), column('name' , String(128)), column('description', String(128))) i = insert(table1, values=dict(name='foo' )).returning(table1.c.myid, table1.c.name) self.assert_compile(i, 'INSERT INTO mytable (name) VALUES (:name) ' 'RETURNING mytable.myid, mytable.name') i = insert(table1, values=dict(name='foo')).returning(table1) self.assert_compile(i, 'INSERT INTO mytable (name) VALUES (:name) ' 'RETURNING mytable.myid, mytable.name, ' 'mytable.description') i = insert(table1, values=dict(name='foo' )).returning(func.length(table1.c.name)) self.assert_compile(i, 'INSERT INTO mytable (name) VALUES (:name) ' 'RETURNING char_length(mytable.name) AS ' 'length_1') def test_charset(self): """Exercise CHARACTER SET options on string types.""" columns = [(firebird.CHAR, [1], {}, 'CHAR(1)'), (firebird.CHAR, [1], {'charset': 'OCTETS'}, 'CHAR(1) CHARACTER SET OCTETS'), (firebird.VARCHAR, [1], {}, 'VARCHAR(1)'), (firebird.VARCHAR, [1], {'charset': 'OCTETS'}, 'VARCHAR(1) CHARACTER SET OCTETS')] for type_, args, kw, res in columns: self.assert_compile(type_(*args, **kw), res) def test_quoting_initial_chars(self): self.assert_compile( column("_somecol"), '"_somecol"' ) self.assert_compile( column("$somecol"), '"$somecol"' ) class TypesTest(fixtures.TestBase): __only_on__ = 'firebird' @testing.provide_metadata def test_infinite_float(self): metadata = self.metadata t = Table('t', metadata, Column('data', Float) ) metadata.create_all() t.insert().execute(data=float('inf')) eq_(t.select().execute().fetchall(), [(float('inf'),)] ) class MiscTest(fixtures.TestBase): __only_on__ = 'firebird' @testing.provide_metadata def test_strlen(self): metadata = self.metadata # On FB the length() function is implemented by an external UDF, # strlen(). Various SA tests fail because they pass a parameter # to it, and that does not work (it always results the maximum # string length the UDF was declared to accept). This test # checks that at least it works ok in other cases. t = Table('t1', metadata, Column('id', Integer, Sequence('t1idseq'), primary_key=True), Column('name' , String(10))) metadata.create_all() t.insert(values=dict(name='dante')).execute() t.insert(values=dict(name='alighieri')).execute() select([func.count(t.c.id)], func.length(t.c.name) == 5).execute().first()[0] == 1 def test_version_parsing(self): for string, result in [ ("WI-V1.5.0.1234 Firebird 1.5", (1, 5, 1234, 'firebird')), ("UI-V6.3.2.18118 Firebird 2.1", (2, 1, 18118, 'firebird')), ("LI-V6.3.3.12981 Firebird 2.0", (2, 0, 12981, 'firebird')), ("WI-V8.1.1.333", (8, 1, 1, 'interbase')), ("WI-V8.1.1.333 Firebird 1.5", (1, 5, 333, 'firebird')), ]: eq_( testing.db.dialect._parse_version_info(string), result ) @testing.provide_metadata def test_rowcount_flag(self): metadata = self.metadata engine = engines.testing_engine( options={'enable_rowcount': True}) assert engine.dialect.supports_sane_rowcount metadata.bind = engine t = Table('t1', metadata, Column('data', String(10))) metadata.create_all() r = t.insert().execute({'data': 'd1'}, {'data': 'd2'}, {'data' : 'd3'}) r = t.update().where(t.c.data == 'd2').values(data='d3' ).execute() eq_(r.rowcount, 1) r = t.delete().where(t.c.data == 'd3').execute() eq_(r.rowcount, 2) r = \ t.delete().execution_options(enable_rowcount=False).execute() eq_(r.rowcount, -1) engine.dispose() engine = engines.testing_engine(options={'enable_rowcount' : False}) assert not engine.dialect.supports_sane_rowcount metadata.bind = engine r = t.insert().execute({'data': 'd1'}, {'data': 'd2'}, {'data' : 'd3'}) r = t.update().where(t.c.data == 'd2').values(data='d3' ).execute() eq_(r.rowcount, -1) r = t.delete().where(t.c.data == 'd3').execute() eq_(r.rowcount, -1) r = t.delete().execution_options(enable_rowcount=True).execute() eq_(r.rowcount, 1) r.close() engine.dispose() def test_percents_in_text(self): for expr, result in (text("select '%' from rdb$database"), '%' ), (text("select '%%' from rdb$database"), '%%'), \ (text("select '%%%' from rdb$database"), '%%%'), \ (text("select 'hello % world' from rdb$database"), 'hello % world'): eq_(testing.db.scalar(expr), result) from sqlalchemy.testing.mock import Mock, call class ArgumentTest(fixtures.TestBase): def _dbapi(self): return Mock( paramstyle='qmark', connect=Mock( return_value=Mock( server_version="UI-V6.3.2.18118 Firebird 2.1", cursor=Mock(return_value=Mock()) ) ) ) def _engine(self, type_, **kw): dbapi = self._dbapi() kw.update( dict( module=dbapi, _initialize=False ) ) engine = engines.testing_engine("firebird+%s://" % type_, options=kw) return engine def test_retaining_flag_default_kinterbasdb(self): engine = self._engine("kinterbasdb") self._assert_retaining(engine, False) def test_retaining_flag_true_kinterbasdb(self): engine = self._engine("kinterbasdb", retaining=True) self._assert_retaining(engine, True) def test_retaining_flag_false_kinterbasdb(self): engine = self._engine("kinterbasdb", retaining=False) self._assert_retaining(engine, False) def test_retaining_flag_default_fdb(self): engine = self._engine("fdb") self._assert_retaining(engine, False) def test_retaining_flag_true_fdb(self): engine = self._engine("fdb", retaining=True) self._assert_retaining(engine, True) def test_retaining_flag_false_fdb(self): engine = self._engine("fdb", retaining=False) self._assert_retaining(engine, False) def _assert_retaining(self, engine, flag): conn = engine.connect() trans = conn.begin() trans.commit() eq_( engine.dialect.dbapi.connect.return_value.commit.mock_calls, [call(flag)] ) trans = conn.begin() trans.rollback() eq_( engine.dialect.dbapi.connect.return_value.rollback.mock_calls, [call(flag)] )