mirror of
https://github.com/FirebirdSQL/firebird-qa.git
synced 2025-01-23 14:03:06 +01:00
76 lines
3.0 KiB
Python
76 lines
3.0 KiB
Python
#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
|