6
0
mirror of https://github.com/FirebirdSQL/firebird-qa.git synced 2025-01-22 21:43:06 +01:00

Added/Updated tests\functional\trigger\database\test_disconnect_03.py: Checked on 5.0.0.961

This commit is contained in:
pavel-zotov 2023-02-28 12:34:15 +03:00
parent 1aaa28db16
commit c3a4b71047

View File

@ -0,0 +1,75 @@
#coding:utf-8
"""
ID: trigger.database.trigger_on_disconnect_03
TITLE: ON DISCONNECT trigger must fire if attachment is deleted from MON$ATTACHMENTS by another attachment.
DESCRIPTION:
Improvement was introduced 06-apr-2023, commit:
https://github.com/FirebirdSQL/firebird/commit/d64868cdcd91450e3c58aa9bfa9c1090912ac358
Test creates table 'detach_audit' which is fulfilled from ON DISCONNECT trigger if current user not SYSDBA.
We create non-privileged user ('tmp_worker') and make connection to database.
During that, we create 'admin' connection which issues 'DELETE FROM MON$ATTACHMENTS WHERE ...'
This *must* cause firing trigger with adding record to 'detach_audit' table (and, according to doc, engine
always starts autonomous transaction for that).
Finally, we check that table 'detach_audit' has ONE record (containing name of non-priv user).
NOTES:
[28.02.2023] pzotov
Confirmed missed trigger firing on 5.0.0.459 (05-apr-2023): table 'detach_audit' remained empty.
Checked on 5.0.0.961 - all OK.
"""
import pytest
from firebird.qa import *
from firebird.driver import ShutdownMode, ShutdownMethod, DatabaseError
import time
tmp_worker = user_factory('db', name='tmp_worker', password='123')
db = db_factory()
act = python_act('db')
expected_stdout_chk = """
ID 1
WHO TMP_WORKER
Records affected: 1
"""
@pytest.mark.version('>=5.0')
def test_1(act: Action, tmp_worker: User):
init_sql = f"""
recreate table detach_audit(
id int generated always as identity
,dts timestamp default 'now'
,who varchar(31) default current_user
);
commit;
grant select, insert on detach_audit to public;
set term ^;
create trigger trg_disconnect on disconnect as
begin
if ( current_user != '{act.db.user}' ) then
insert into detach_audit default values;
end
^
set term ;^
commit;
"""
act.expected_stdout = ''
act.isql(switches=['-q'], input = init_sql, combine_output = True)
assert act.clean_stdout == act.clean_expected_stdout
act.reset()
worker_remains_alive = False
with act.db.connect() as con_admin, act.db.connect(user = tmp_worker.name, password = tmp_worker.password) as con_worker:
con_admin.execute_immediate(f'delete from mon$attachments where mon$attachment_id = {con_worker.info.id}')
con_admin.commit()
cur_admin = con_admin.cursor()
cur_admin.execute(f'select mon$attachment_id from mon$attachments where mon$attachment_id = {con_worker.info.id}')
for r in cur_admin:
worker_remains_alive = True
assert not worker_remains_alive, '### FAIL ### Could not delete attachment of non-privileged user'
act.expected_stdout = expected_stdout_chk
act.isql(switches=['-q'], input = 'set count on;set list on;select id,who from detach_audit;', combine_output = True)
assert act.clean_stdout == act.clean_expected_stdout