firebird-qa/tests/functional/trigger/database/test_connect_04.py

#coding:utf-8

"""
ID:          trigger.database.connect-02
TITLE:       Error handling when multiple triggers on database connect exist
DESCRIPTION:
  Test behavoiur when MULTIPLE db-level triggers on CONNECT exist:
  first uncaught exception in the trigger declared with attribute  'position N'
  must prevent execution of any subsequent triggers (i.e. with position > N);
FBTEST:      functional.trigger.database.connect_02
NOTES:
[26.05.2022] pzotov
  Re-implemented for work in firebird-qa suite. 
  Checked on: 3.0.8.33535, 4.0.1.2692, 5.0.0.497
"""

import pytest
from firebird.qa import *

db = db_factory()
tmp_worker = user_factory('db', name='tmp_worker', password='123')
tmp_hacker = user_factory('db', name='tmp_hacker', password='456')

act = python_act('db', substitutions=[('line:.*', '')])


@pytest.mark.version('>=3.0')
def test_1(act: Action, tmp_worker: User, tmp_hacker: User):

    expected_stdout = f"""
        Statement failed, SQLSTATE = HY000
        exception 1
        -EXC_CONNECT
        -Exception in ON CONNECT trigger trg_connect_1 for user {tmp_hacker.name}
        -At trigger 'TRG_CONNECT_1'
        ID                              1
        AUDIT_WHO                       {tmp_worker.name}
        TRG_NAME                        trg_connect_1
        ID                              2
        AUDIT_WHO                       {tmp_worker.name}
        TRG_NAME                        trg_connect_2
        ID                              3
        AUDIT_WHO                       {tmp_worker.name}
        TRG_NAME                        trg_connect_3
        ID                              4
        AUDIT_WHO                       {tmp_hacker.name}
        TRG_NAME                        trg_connect_1
        Records affected: 4
        ID                              1
        SESSION_WHO                     {tmp_worker.name}
        TRG_NAME                        trg_connect_1
        ID                              2
        SESSION_WHO                     {tmp_worker.name}
        TRG_NAME                        trg_connect_2
        ID                              3
        SESSION_WHO                     {tmp_worker.name}
        TRG_NAME                        trg_connect_3
        Records affected: 3
    """

    script = f"""
        create table sessions_in_work(
            id int generated by default as identity constraint pk_ssn_in_work primary key
            ,session_who varchar(31) default current_user
            ,trg_name varchar(31)
        );

        create table sessions_audit(
            id int generated by default as identity constraint pk_ssn_audit primary key
            ,audit_who varchar(31) default current_user
            ,trg_name varchar(31)
        );
        create exception exc_connect 'Exception in ON CONNECT trigger @1 for user @2';

        set term ^;
        create trigger trg_connect_1 on connect position 1 as
        begin
          if ( current_user = '{act.db.user}' ) then
              exit;

          in autonomous transaction do
              insert into sessions_audit(trg_name) values('trg_connect_1');

          insert into sessions_in_work(trg_name) values('trg_connect_1');

          if ( current_user = upper('{tmp_hacker.name}') ) then
              exception exc_connect using('trg_connect_1', current_user);
        end
        ^
        ----------------------------------------------------------------------
        create trigger trg_connect_2 on connect position 2 as
        begin
          if ( current_user = '{act.db.user}' ) then
              exit;

          in autonomous transaction do
              insert into sessions_audit(trg_name) values('trg_connect_2');

          insert into sessions_in_work(trg_name) values('trg_connect_2');

          if ( current_user = upper('{tmp_hacker.name}') ) then
              exception exc_connect using('trg_connect_2', current_user);
        end
        ^
        ----------------------------------------------------------------------
        create trigger trg_connect_3 on connect position 3 as
        begin
          if ( current_user = '{act.db.user}' ) then
              exit;

          in autonomous transaction do
              insert into sessions_audit(trg_name) values('trg_connect_3');

          insert into sessions_in_work(trg_name) values('trg_connect_3');

          if ( current_user = upper('{tmp_hacker.name}') ) then
              exception exc_connect using('trg_connect_3', current_user);
        end
        ^
        set term ;^
        commit;
        -----------------------------------------------------
        connect '{act.db.dsn}' user '{tmp_worker.name}' password '{tmp_worker.password}';
        commit;

        connect '{act.db.dsn}' user '{tmp_hacker.name}' password '{tmp_hacker.password}';
        commit;
        -----------------------------------------------------

        -- Connect as DBA for obtaining results:
        connect '{act.db.dsn}' user '{act.db.user}' password '{act.db.password}';
        set list on;
        set count on;
        select * from sessions_audit;
        select * from sessions_in_work;
    """

    act.expected_stdout = expected_stdout
    act.isql(switches=['-q'], input = script, combine_output=True)
    assert act.clean_stdout == act.clean_expected_stdout