diff --git a/tests/bugs/core_5077_test.py b/tests/bugs/core_5077_test.py index 5bad2b65..81dbacdb 100644 --- a/tests/bugs/core_5077_test.py +++ b/tests/bugs/core_5077_test.py @@ -19,7 +19,7 @@ DESCRIPTION: We open connection to DB and run 'ALTER DATABASE ENCRYPT.../DECRYPT'. One need to keep connection opened for several seconds in order to give encryption thread be fully completed. Duration of this delay depends on concurrent workload, usually it is almost zero. - But in this test it can be tuned - see variable 'MAX_ENCRYPT_DECRYPT_MS'. + But in this test it can be tuned - see variable 'MAX_WAITING_ENCR_FINISH'. Immediately after launch encryption/decryption, we run isql and ask it to give result of 'SHOW DATABASE' command. If this output contains text 'Database [not] encrypted' and *not* contains phrase 'not complete' then we can assume that encryption/decryption thread completed. Otherwise we loop until such conditions will raise or timeout expired. @@ -31,6 +31,7 @@ NOTES: Checked on 4.0.1.2692, 3.0.8.33535 - both on Linux and Windows. """ +import os import time import datetime as py_dt from datetime import timedelta @@ -39,7 +40,22 @@ import pytest from firebird.qa import * from firebird.driver import DatabaseError -MAX_ENCRYPT_DECRYPT_MS = 5000 +########################### +### S E T T I N G S ### +########################### + +# QA_GLOBALS -- dict, is defined in qa/plugin.py, obtain settings +# from act.files_dir/'test_config.ini': +enc_settings = QA_GLOBALS['encryption'] + +# ACHTUNG: this must be carefully tuned on every new host: +# +MAX_WAITING_ENCR_FINISH = int(enc_settings['MAX_WAIT_FOR_ENCR_FINISH_WIN' if os.name == 'nt' else 'MAX_WAIT_FOR_ENCR_FINISH_NIX']) +assert MAX_WAITING_ENCR_FINISH > 0 + +ENCRYPTION_PLUGIN = enc_settings['encryption_plugin'] # fbSampleDbCrypt +ENCRYPTION_KEY = enc_settings['encryption_key'] # Red + db = db_factory() act = python_act('db') @@ -54,7 +70,7 @@ def test_1(act: Action, capsys): with act.db.connect() as con: t1=py_dt.datetime.now() d1 = t1-t1 - sttm = 'alter database ' + ('encrypt with "fbSampleDbCrypt" key "Red"' if m == 'encryption' else 'decrypt') + sttm = 'alter database ' + ( f'encrypt with "{ENCRYPTION_PLUGIN}" key "{ENCRYPTION_KEY}"' if m == 'encryption' else 'decrypt' ) try: con.execute_immediate(sttm) con.commit() @@ -70,7 +86,7 @@ def test_1(act: Action, capsys): while True: t2=py_dt.datetime.now() d1=t2-t1 - if d1.seconds*1000 + d1.microseconds//1000 > MAX_ENCRYPT_DECRYPT_MS: + if d1.seconds*1000 + d1.microseconds//1000 > MAX_WAITING_ENCR_FINISH: break # Possible output: @@ -84,10 +100,10 @@ def test_1(act: Action, capsys): break act.reset() - if d1.seconds*1000 + d1.microseconds//1000 <= MAX_ENCRYPT_DECRYPT_MS: + if d1.seconds*1000 + d1.microseconds//1000 <= MAX_WAITING_ENCR_FINISH: print(expected_stdout_console) else: - print(f'BREAK ON TIMEOUT EXPIRATION: {m.upper()} took {d1.seconds*1000 + d1.microseconds//1000} ms which exceeds limit = {MAX_ENCRYPT_DECRYPT_MS} ms.') + print(f'BREAK ON TIMEOUT EXPIRATION: {m.upper()} took {d1.seconds*1000 + d1.microseconds//1000} ms which exceeds limit = {MAX_WAITING_ENCR_FINISH} ms.') act.expected_stdout = expected_stdout_console act.stdout = capsys.readouterr().out