mirror of
https://github.com/FirebirdSQL/firebird-qa.git
synced 2025-01-22 13:33:07 +01:00
Added/Updated functional\gtcs\test_transactions_autocommit_3.py: checked on 4.0.1.2692, 3.0.8.33535
This commit is contained in:
zotov
parent
64fd89305f
commit
02fd131ef9
@ -12,7 +12,7 @@ DESCRIPTION:
|
||||
with passing there input value multiplied by 100.
|
||||
When sp_ins_1 is called with argument = 3 then sp_ins_2 will insert into test_2 table value = 300
|
||||
and sp_ins_3 will insert into test_3 value = 30000.
|
||||
This mean that we can can NOT call sp_ins_1 with values equal or more than 4 because of numeric overflow exception
|
||||
This means that we can can NOT call sp_ins_1 with values equal or more than 4 because of numeric overflow exception
|
||||
that will be raised in sp_ins_3.
|
||||
|
||||
Test calls sp_ins1 two times: with arg=3 and arg=4. Second time must fail and we check that all three tables contain only
|
||||
@ -24,85 +24,70 @@ FBTEST: functional.gtcs.transactions_autocommit_3
|
||||
|
||||
import pytest
|
||||
from firebird.qa import *
|
||||
from firebird.driver import *
|
||||
|
||||
db = db_factory()
|
||||
N_MAX = 3
|
||||
sql_proc = """
|
||||
create table test_%(i)s(x smallint)
|
||||
^
|
||||
create procedure sp_ins_%(i)s(a_x bigint) as
|
||||
begin
|
||||
insert into test_%(i)s(x) values(:a_x);
|
||||
if ( %(i)s != %(N_MAX)s ) then
|
||||
execute statement ( 'execute procedure sp_ins_%(k)s (?)' ) (:a_x * 100);
|
||||
end
|
||||
^
|
||||
commit
|
||||
^
|
||||
"""
|
||||
|
||||
act = python_act('db')
|
||||
init_script = 'set term ^;'
|
||||
for i in range(N_MAX,0,-1):
|
||||
k = i+1
|
||||
init_script += sql_proc % locals()
|
||||
|
||||
db = db_factory(init=init_script)
|
||||
|
||||
act = python_act('db', substitutions=[('[ \t]+', ' ')])
|
||||
|
||||
expected_stdout = """
|
||||
mon$auto_commit: 1
|
||||
x: 3
|
||||
x: 300
|
||||
x: 30000
|
||||
exception occured, gdscodes: (335544321, 335544916, 335544842, 335544842, 335544842)
|
||||
10
|
||||
3
|
||||
300
|
||||
30000
|
||||
"""
|
||||
|
||||
@pytest.mark.skip('FIXME: Not IMPLEMENTED')
|
||||
@pytest.mark.version('>=3')
|
||||
def test_1(act: Action):
|
||||
pytest.fail("Not IMPLEMENTED")
|
||||
custom_tpb = TPB(lock_timeout = 0, auto_commit=True).get_buffer()
|
||||
|
||||
@pytest.mark.version('>=3')
|
||||
def test_1(act: Action, capsys):
|
||||
with act.db.connect() as con:
|
||||
|
||||
con.begin(custom_tpb)
|
||||
with con.cursor() as cur:
|
||||
for r in cur.execute('select mon$auto_commit from mon$transactions where mon$transaction_id = current_transaction').fetchall():
|
||||
print('mon$auto_commit:', r[0])
|
||||
|
||||
try:
|
||||
cur.execute( 'insert into test_1 values(?)', (10,) ) # this leads to PK/UK violation in the table 'test_3'
|
||||
except DatabaseError as e:
|
||||
print('exception occured, gdscodes:', e.gds_codes)
|
||||
|
||||
with con.cursor() as cur:
|
||||
cur.callproc( 'sp_ins_1', (3,) )
|
||||
|
||||
try:
|
||||
# this will cause numeric overflow exception in sp_ins_3:
|
||||
cur.callproc( 'sp_ins_1', (4,) )
|
||||
except DatabaseError as e:
|
||||
print('exception occured, gdscodes:', e.gds_codes)
|
||||
|
||||
con.commit()
|
||||
with con.cursor() as cur:
|
||||
for r in cur.execute('select x from test_1 union all select x from test_2 union all select x from test_3').fetchall():
|
||||
print(r[0])
|
||||
|
||||
assert act.clean_string(capsys.readouterr().out, act.substitutions) == act.clean_string(expected_stdout, act.substitutions)
|
||||
|
||||
# test_script_1
|
||||
#---
|
||||
#
|
||||
# import os
|
||||
# os.environ["ISC_USER"] = user_name
|
||||
# os.environ["ISC_PASSWORD"] = user_password
|
||||
#
|
||||
# import os
|
||||
# import sys
|
||||
# import inspect
|
||||
# import fdb
|
||||
#
|
||||
# N_MAX=3
|
||||
#
|
||||
# #CUSTOM_TX_PARAMS = ( [ fdb.isc_tpb_read_committed, fdb.isc_tpb_no_rec_version, fdb.isc_tpb_nowait, fdb.isc_tpb_autocommit ] )
|
||||
# CUSTOM_TX_PARAMS = ( [ fdb.isc_tpb_nowait, fdb.isc_tpb_autocommit ] )
|
||||
#
|
||||
# db_conn.begin()
|
||||
# cx=db_conn.cursor()
|
||||
# sql_proc='''
|
||||
# create procedure sp_ins_%(i)s(a_x bigint) as
|
||||
# begin
|
||||
# insert into test_%(i)s(x) values(:a_x);
|
||||
# if ( %(i)s != %(N_MAX)s ) then
|
||||
# execute statement ( 'execute procedure sp_ins_%(k)s (?)' ) (:a_x * 100);
|
||||
# end
|
||||
# '''
|
||||
#
|
||||
# for i in range(N_MAX,0,-1):
|
||||
# k = i+1
|
||||
# cx.execute( 'create table test_%(i)s(x smallint)' % locals() )
|
||||
# cx.execute( sql_proc % locals() )
|
||||
#
|
||||
# db_conn.commit()
|
||||
#
|
||||
# tx = db_conn.trans( default_tpb = CUSTOM_TX_PARAMS )
|
||||
#
|
||||
# tx.begin()
|
||||
# cx=tx.cursor()
|
||||
#
|
||||
# cx.execute('select mon$auto_commit from mon$transactions where mon$transaction_id = current_transaction')
|
||||
# for r in cx:
|
||||
# print( 'mon$auto_commit:', r[0] )
|
||||
#
|
||||
# cx.callproc( 'sp_ins_1', (3,) )
|
||||
#
|
||||
# try:
|
||||
# cx.callproc( 'sp_ins_1', (4,) )
|
||||
# except Exception as e:
|
||||
# pass
|
||||
# #print('Unexpected exception in ', inspect.stack()[0][3], ': ', sys.exc_info()[0])
|
||||
# #print(e)
|
||||
#
|
||||
# tx.commit()
|
||||
#
|
||||
# cx = db_conn.cursor()
|
||||
# cx.execute('select x from test_1 union all select x from test_2 union all select x from test_3')
|
||||
# for r in cx:
|
||||
# print( 'x:', r[0])
|
||||
#
|
||||
# cx.close()
|
||||
# tx.close()
|
||||
# db_conn.close()
|
||||
#
|
||||
#---
|
||||
|
||||
Loading…
Reference in New Issue
Block a user