firebird-qa/tests/bugs/core_5075_test.py

#coding:utf-8
#
# id:           bugs.core_5075
# title:        Regression. Triger on DISCONNECT with dynamic SQL (ES 'insert into ...'): 1) does not work in 3.0; 2) leads FB to crash when it is recreated
# decription:   
#                  Test does following:
#                  * obtains firebird.log as it was _before_ actions;
#                  * stores initial script for creation DB objects in file <f_sql_init> for futher applying it twice (see ticket);
#                  * open PIPE object and communicates with ISQL issuing commands into STDIN, three times; result is stored in var.'sqlres'
#                  * print variable 'sqlres'; it content must be equal to that we get in FB 2.5;
#                  * again apply initial DDL and then - one again launch ISQL with passing to it commands via STDIN and saving result in 'sqlres'
#                  * print variable 'sqlres';
#                  * obtains firebird.log as it is _after_ actions;
#                  * compare two firebird.log versions - diff must be empty.
#               
#                  Checked on 3.0.0.32281, SS/SC/CS.
#               
#                
# tracker_id:   CORE-5075
# min_versions: ['3.0']
# versions:     3.0
# qmid:         None

import pytest
from firebird.qa import db_factory, isql_act, Action

# version: 3.0
# resources: None

substitutions_1 = []

init_script_1 = """"""

db_1 = db_factory(sql_dialect=3, init=init_script_1)

# test_script_1
#---
# import os
#  import subprocess
#  from subprocess import Popen, PIPE, STDOUT
#  import time
#  import difflib
#  
#  os.environ["ISC_USER"] = user_name
#  os.environ["ISC_PASSWORD"] = user_password
#  db_conn.close()
#  
#  #--------------------------------------------
#  
#  def flush_and_close(file_handle):
#      # https://docs.python.org/2/library/os.html#os.fsync
#      # If you're starting with a Python file object f, 
#      # first do f.flush(), and 
#      # then do os.fsync(f.fileno()), to ensure that all internal buffers associated with f are written to disk.
#      global os
#      
#      file_handle.flush()
#      if file_handle.mode not in ('r', 'rb'):
#          # otherwise: "OSError: [Errno 9] Bad file descriptor"!
#          os.fsync(file_handle.fileno())
#      file_handle.close()
#  
#  #--------------------------------------------
#  
#  def cleanup( f_names_list ):
#      global os
#      for i in range(len( f_names_list )):
#         if type(f_names_list[i]) == file:
#            del_name = f_names_list[i].name
#         elif type(f_names_list[i]) == str:
#            del_name = f_names_list[i]
#         else:
#            print('Unrecognized type of element:', f_names_list[i], ' - can not be treated as file.')
#            del_name = None
#  
#         if del_name and os.path.isfile( del_name ):
#             os.remove( del_name )
#  
#  #--------------------------------------------
#  
#  
#  def svc_get_fb_log( f_fb_log ):
#  
#    import subprocess
#  
#    subprocess.call([ context['fbsvcmgr_path'],
#                      "localhost:service_mgr",
#                      "action_get_fb_log"
#                    ],
#                     stdout=f_fb_log, stderr=subprocess.STDOUT
#                   )
#    return
#  
#  #--------------------------------------------
#  
#  f_fblog_before=open( os.path.join(context['temp_directory'],'tmp_5075_fblog_before.txt'), 'w')
#  svc_get_fb_log( f_fblog_before )
#  flush_and_close( f_fblog_before )
#  
#  sqltxt='''set term ^;
#  create or alter trigger trg_connect active on connect position 0 as
#  begin
#  end
#  ^
#  
#  create or alter trigger trg_disc active on disconnect position 0 as
#  begin
#  end
#  ^
#  set term ;^
#  commit;
#  
#  recreate sequence g;
#  recreate table log(
#      event_id int generated by default as identity constraint pk_log primary key,
#      event_name varchar(20), 
#      when_it_was timestamp default 'now'
#  );
#  commit;
#  
#  set term ^;
#  execute block as
#  begin
#      rdb$set_context('USER_SESSION','INITIAL_DDL','1');
#  end
#  ^
#  
#  create or alter trigger trg_connect active on connect position 0 as
#  begin
#      execute statement 'insert into log(event_name) values(''connect'')'
#      with autonomous transaction;
#  end
#  ^
#  
#  create or alter trigger trg_disc active on disconnect position 0 as
#  begin
#      if ( rdb$get_context('USER_SESSION','INITIAL_DDL') is null ) then
#          execute statement 'insert into log(event_name) values(''disconnect'')'
#          with autonomous transaction;
#  end
#  ^
#  set term ;^
#  commit; 
#  '''
#  
#  f_sql_init=open( os.path.join(context['temp_directory'],'tmp_5075_init.sql'), 'w')
#  f_sql_init.write(sqltxt)
#  flush_and_close( f_sql_init )
#  
#  sqlres=subprocess.check_output([context['isql_path'], dsn, "-nod", "-i", f_sql_init.name], stderr=subprocess.STDOUT)
#  print(sqlres) # Must be empty (no errors)
#  
#  sqltxt='''set list on;set count on; select event_id, event_name from log; 
#  '''
#  
#  # http://stackoverflow.com/questions/8475290/how-do-i-write-to-a-python-subprocess-stdin
#  
#  sqlres=''
#  
#  p = Popen([context['isql_path'], dsn], stdout=PIPE, stdin=PIPE, stderr=subprocess.STDOUT )
#  sqlres += p.communicate(input=sqltxt)[0]
#  
#  
#  p = Popen([context['isql_path'], dsn], stdout=PIPE, stdin=PIPE, stderr=subprocess.STDOUT )
#  sqlres += p.communicate(input=sqltxt)[0]
#  
#  p = Popen([context['isql_path'], dsn], stdout=PIPE, stdin=PIPE, stderr=subprocess.STDOUT )
#  sqlres += p.communicate(input=sqltxt)[0]
#  
#  print(sqlres)
#  
#  sqlres=subprocess.check_output([context['isql_path'], dsn, "-nod", "-i", f_sql_init.name], stderr=subprocess.STDOUT)
#  print(sqlres) # Must be empty (no errors)
#  
#  p = Popen([context['isql_path'], dsn], stdout=PIPE, stdin=PIPE, stderr=subprocess.STDOUT )
#  sqlres = p.communicate(input=sqltxt)[0]
#  print(sqlres)
#  
#  
#  f_fblog_after=open( os.path.join(context['temp_directory'],'tmp_5075_fblog_after.txt'), 'w')
#  svc_get_fb_log( f_fblog_after )
#  flush_and_close( f_fblog_after )
#  
#  time.sleep(1)
#  
#  oldfb=open(f_fblog_before.name, 'r')
#  newfb=open(f_fblog_after.name, 'r')
#  
#  difftext = ''.join(difflib.unified_diff(
#      oldfb.readlines(), 
#      newfb.readlines()
#    ))
#  oldfb.close()
#  newfb.close()
#  
#  f_diff_txt=open( os.path.join(context['temp_directory'],'tmp_5075_diff.txt'), 'w')
#  f_diff_txt.write(difftext)
#  flush_and_close( f_diff_txt )
#  
#  with open( f_diff_txt.name,'r') as f:
#      for line in f:
#          print('new messages in firebird.log: '+line)
#  
#  # Cleanup.
#  ##########
#  # do NOT remove this pause otherwise some of logs will not be enable for deletion and test will finish with 
#  # Exception raised while executing Python test script. exception: WindowsError: 32
#  time.sleep(1)
#  cleanup((f_sql_init,f_fblog_before,f_fblog_after,f_diff_txt))
#    
#---
#act_1 = python_act('db_1', test_script_1, substitutions=substitutions_1)

expected_stdout_1 = """
    EVENT_ID                        1
    EVENT_NAME                      connect
    Records affected: 1
    EVENT_ID                        1
    EVENT_NAME                      connect
    EVENT_ID                        2
    EVENT_NAME                      disconnect
    EVENT_ID                        3
    EVENT_NAME                      connect
    Records affected: 3
    EVENT_ID                        1
    EVENT_NAME                      connect
    EVENT_ID                        2
    EVENT_NAME                      disconnect
    EVENT_ID                        3
    EVENT_NAME                      connect
    EVENT_ID                        4
    EVENT_NAME                      disconnect
    EVENT_ID                        5
    EVENT_NAME                      connect
    Records affected: 5
    EVENT_ID                        1
    EVENT_NAME                      connect
    Records affected: 1
  """

@pytest.mark.version('>=3.0')
@pytest.mark.xfail
def test_1(db_1):
    pytest.fail("Test not IMPLEMENTED")