diff --git a/pyproject.toml b/pyproject.toml index 947af11e..985aec28 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,7 +65,7 @@ dependencies = [ "six>=1.9.0", "psutil", "packaging", - "testgres.os_ops>=3.3.0,<4.0.0", + "testgres.os_ops>=3.3.3,<4.0.0", ] [project.urls] diff --git a/tests/conftest.py b/tests/conftest.py index a1adc157..b7d119ec 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,10 @@ # ///////////////////////////////////////////////////////////////////////////// # PyTest Configuration +from .conftest_helpers import TestStartupData +from .conftest_helpers import TestServices +from .conftest_helpers import TestExitStatus + import pluggy import pytest import os @@ -10,19 +14,15 @@ import datetime import typing import enum +import threading import _pytest.outcomes -import _pytest.unittest import _pytest.logging from packaging.version import Version # ///////////////////////////////////////////////////////////////////////////// -C_ROOT_DIR__RELATIVE = ".." - -# ///////////////////////////////////////////////////////////////////////////// - T_TUPLE__str_int = typing.Tuple[str, int] # ///////////////////////////////////////////////////////////////////////////// @@ -58,117 +58,14 @@ class T_TEST_PROCESS_MODE(enum.Enum): # ///////////////////////////////////////////////////////////////////////////// - -g_test_process_kind: typing.Optional[T_TEST_PROCESS_KIND] = None -g_test_process_mode: typing.Optional[T_TEST_PROCESS_MODE] = None - -g_worker_log_is_created: typing.Optional[bool] = None - -# ///////////////////////////////////////////////////////////////////////////// -# TestConfigPropNames - - -class TestConfigPropNames: - TEST_CFG__LOG_DIR = "TEST_CFG__LOG_DIR" - - -# ///////////////////////////////////////////////////////////////////////////// -# TestStartupData__Helper - - -class TestStartupData__Helper: - sm_StartTS = datetime.datetime.now() - - # -------------------------------------------------------------------- - @staticmethod - def GetStartTS() -> datetime.datetime: - assert type(__class__.sm_StartTS) is datetime.datetime - return __class__.sm_StartTS - - # -------------------------------------------------------------------- - @staticmethod - def CalcRootDir() -> str: - r = os.path.abspath(__file__) - r = os.path.dirname(r) - r = os.path.join(r, C_ROOT_DIR__RELATIVE) - r = os.path.abspath(r) - return r - - # -------------------------------------------------------------------- - @staticmethod - def CalcRootLogDir() -> str: - if TestConfigPropNames.TEST_CFG__LOG_DIR in os.environ: - resultPath = os.environ[TestConfigPropNames.TEST_CFG__LOG_DIR] - else: - rootDir = __class__.CalcRootDir() - resultPath = os.path.join(rootDir, "logs") - - assert type(resultPath) is str - return resultPath - - # -------------------------------------------------------------------- - @staticmethod - def CalcCurrentTestWorkerSignature() -> str: - currentPID = os.getpid() - assert type(currentPID) is int - - startTS = __class__.sm_StartTS - assert type(startTS) is datetime.datetime - - result = "pytest-{0:04d}{1:02d}{2:02d}_{3:02d}{4:02d}{5:02d}".format( - startTS.year, - startTS.month, - startTS.day, - startTS.hour, - startTS.minute, - startTS.second, - ) - - gwid = os.environ.get("PYTEST_XDIST_WORKER") - - if gwid is not None: - result += "--xdist_" + str(gwid) - - result += "--" + "pid" + str(currentPID) - return result - - -# ///////////////////////////////////////////////////////////////////////////// -# TestStartupData +# TPDATA -class TestStartupData: - sm_RootDir: str = TestStartupData__Helper.CalcRootDir() - sm_CurrentTestWorkerSignature: str = ( - TestStartupData__Helper.CalcCurrentTestWorkerSignature() - ) +class TPDATA: + test_process_kind: typing.Optional[T_TEST_PROCESS_KIND] = None + test_process_mode: typing.Optional[T_TEST_PROCESS_MODE] = None + worker_log_is_created: typing.Optional[bool] = None - sm_RootLogDir: str = TestStartupData__Helper.CalcRootLogDir() - - # -------------------------------------------------------------------- - @staticmethod - def GetRootDir() -> str: - assert type(__class__.sm_RootDir) is str - return __class__.sm_RootDir - - # -------------------------------------------------------------------- - @staticmethod - def GetRootLogDir() -> str: - assert type(__class__.sm_RootLogDir) is str - return __class__.sm_RootLogDir - - # -------------------------------------------------------------------- - @staticmethod - def GetCurrentTestWorkerSignature() -> str: - assert type(__class__.sm_CurrentTestWorkerSignature) is str - return __class__.sm_CurrentTestWorkerSignature - - -# ///////////////////////////////////////////////////////////////////////////// -# TEST_PROCESS_STATS - - -class TEST_PROCESS_STATS: cTotalTests: int = 0 cNotExecutedTests: int = 0 cExecutedTests: int = 0 @@ -201,6 +98,7 @@ def incrementTotalTestCount() -> None: __class__.cTotalTests += 1 assert __class__.cTotalTests > 0 + return # -------------------------------------------------------------------- @staticmethod @@ -211,6 +109,7 @@ def incrementNotExecutedTestCount() -> None: __class__.cNotExecutedTests += 1 assert __class__.cNotExecutedTests > 0 + return # -------------------------------------------------------------------- @staticmethod @@ -232,6 +131,7 @@ def incrementPassedTestCount() -> None: __class__.cPassedTests += 1 assert __class__.cPassedTests > 0 + return # -------------------------------------------------------------------- @staticmethod @@ -257,6 +157,7 @@ def incrementFailedTestCount(testID: str, errCount: int) -> None: __class__.cTotalErrors += errCount assert __class__.cTotalErrors > 0 + return # -------------------------------------------------------------------- @staticmethod @@ -274,6 +175,7 @@ def incrementXFailedTestCount(testID: str, errCount: int) -> None: assert len(__class__.XFailedTests) > 0 assert __class__.cXFailedTests > 0 assert len(__class__.XFailedTests) == __class__.cXFailedTests + return # -------------------------------------------------------------------- @staticmethod @@ -284,6 +186,7 @@ def incrementSkippedTestCount() -> None: __class__.cSkippedTests += 1 assert __class__.cSkippedTests > 0 + return # -------------------------------------------------------------------- @staticmethod @@ -299,6 +202,7 @@ def incrementNotXFailedTests(testID: str) -> None: assert len(__class__.NotXFailedTests) > 0 assert __class__.cNotXFailedTests > 0 assert len(__class__.NotXFailedTests) == __class__.cNotXFailedTests + return # -------------------------------------------------------------------- @staticmethod @@ -325,6 +229,7 @@ def incrementWarningTestCount(testID: str, warningCount: int) -> None: __class__.cTotalWarnings += warningCount assert __class__.cTotalWarnings > 0 + return # -------------------------------------------------------------------- @staticmethod @@ -335,6 +240,7 @@ def incrementUnexpectedTests() -> None: __class__.cUnexpectedTests += 1 assert __class__.cUnexpectedTests > 0 + return # -------------------------------------------------------------------- @staticmethod @@ -350,6 +256,7 @@ def incrementAchtungTestCount(testID: str) -> None: assert len(__class__.AchtungTests) > 0 assert __class__.cAchtungTests > 0 assert len(__class__.AchtungTests) == __class__.cAchtungTests + return # ///////////////////////////////////////////////////////////////////////////// @@ -397,6 +304,14 @@ def helper__build_test_id(item: pytest.Function) -> str: # ///////////////////////////////////////////////////////////////////////////// +def helper__exc_to_text(exc: BaseException) -> str: + assert isinstance(exc, BaseException) + return TestServices.ExceptionToHumanText(exc) + + +# ///////////////////////////////////////////////////////////////////////////// + + def helper__makereport__setup( item: pytest.Function, call: pytest.CallInfo, outcome: T_PLUGGY_RESULT ): @@ -412,20 +327,20 @@ def helper__makereport__setup( # logging.info("pytest_runtest_makereport - setup") - TEST_PROCESS_STATS.incrementTotalTestCount() + TPDATA.incrementTotalTestCount() rep: pytest.TestReport = outcome.get_result() assert rep is not None assert type(rep) is pytest.TestReport if rep.outcome == "skipped": - TEST_PROCESS_STATS.incrementNotExecutedTestCount() + TPDATA.incrementNotExecutedTestCount() return testID = helper__build_test_id(item) if rep.outcome == "passed": - testNumber = TEST_PROCESS_STATS.incrementExecutedTestCount() + testNumber = TPDATA.incrementExecutedTestCount() logging.info(C_LINE1) logging.info("* START TEST {0}".format(testID)) @@ -437,7 +352,7 @@ def helper__makereport__setup( assert rep.outcome != "passed" - TEST_PROCESS_STATS.incrementAchtungTestCount(testID) + TPDATA.incrementAchtungTestCount(testID) logging.info(C_LINE1) logging.info("* ACHTUNG TEST {0}".format(testID)) @@ -449,22 +364,12 @@ def helper__makereport__setup( assert call.excinfo is not None assert call.excinfo.value is not None logging.info("*") - logging.error(call.excinfo.value) + logging.error(helper__exc_to_text(call.excinfo.value)) logging.info("*") return -# ------------------------------------------------------------------------ -class ExitStatusNames: - FAILED = "FAILED" - PASSED = "PASSED" - XFAILED = "XFAILED" - NOT_XFAILED = "NOT XFAILED" - SKIPPED = "SKIPPED" - UNEXPECTED = "UNEXPECTED" - - # ------------------------------------------------------------------------ def helper__makereport__call( item: pytest.Function, call: pytest.CallInfo, outcome: T_PLUGGY_RESULT @@ -513,7 +418,7 @@ def helper__makereport__call( assert type(testDurration) is datetime.timedelta # -------- - exitStatus = None + exitStatus: typing.Optional[TestExitStatus] = None exitStatusInfo = None if rep.outcome == "skipped": assert call.excinfo is not None # research @@ -522,21 +427,21 @@ def helper__makereport__call( if type(call.excinfo.value) is _pytest.outcomes.Skipped: assert not hasattr(rep, "wasxfail") - exitStatus = ExitStatusNames.SKIPPED + exitStatus = TestExitStatus.SKIPPED reasonText = str(call.excinfo.value) reasonMsgTempl = "SKIP REASON: {0}" - TEST_PROCESS_STATS.incrementSkippedTestCount() + TPDATA.incrementSkippedTestCount() elif type(call.excinfo.value) is _pytest.outcomes.XFailed: - exitStatus = ExitStatusNames.XFAILED + exitStatus = TestExitStatus.XFAILED reasonText = str(call.excinfo.value) reasonMsgTempl = "XFAIL REASON: {0}" - TEST_PROCESS_STATS.incrementXFailedTestCount(testID, item_error_msg_count) + TPDATA.incrementXFailedTestCount(testID, item_error_msg_count) else: - exitStatus = ExitStatusNames.XFAILED + exitStatus = TestExitStatus.XFAILED assert hasattr(rep, "wasxfail") assert rep.wasxfail is not None assert type(rep.wasxfail) is str @@ -547,10 +452,10 @@ def helper__makereport__call( if type(call.excinfo.value) is SIGNAL_EXCEPTION: pass else: - logging.error(call.excinfo.value) + logging.error(helper__exc_to_text(call.excinfo.value)) item_error_msg_count += 1 - TEST_PROCESS_STATS.incrementXFailedTestCount(testID, item_error_msg_count) + TPDATA.incrementXFailedTestCount(testID, item_error_msg_count) assert type(reasonText) is str @@ -567,20 +472,20 @@ def helper__makereport__call( assert item_error_msg_count > 0 pass else: - logging.error(call.excinfo.value) + logging.error(helper__exc_to_text(call.excinfo.value)) item_error_msg_count += 1 assert item_error_msg_count > 0 - TEST_PROCESS_STATS.incrementFailedTestCount(testID, item_error_msg_count) + TPDATA.incrementFailedTestCount(testID, item_error_msg_count) - exitStatus = ExitStatusNames.FAILED + exitStatus = TestExitStatus.FAILED elif rep.outcome == "passed": assert call.excinfo is None if hasattr(rep, "wasxfail"): assert type(rep.wasxfail) is str - TEST_PROCESS_STATS.incrementNotXFailedTests(testID) + TPDATA.incrementNotXFailedTests(testID) warnMsg = "NOTE: Test is marked as xfail" @@ -588,41 +493,44 @@ def helper__makereport__call( warnMsg += " [" + rep.wasxfail + "]" logging.info(warnMsg) - exitStatus = ExitStatusNames.NOT_XFAILED + exitStatus = TestExitStatus.NOT_XFAILED else: assert not hasattr(rep, "wasxfail") - TEST_PROCESS_STATS.incrementPassedTestCount() - exitStatus = ExitStatusNames.PASSED + TPDATA.incrementPassedTestCount() + exitStatus = TestExitStatus.PASSED else: - TEST_PROCESS_STATS.incrementUnexpectedTests() - exitStatus = ExitStatusNames.UNEXPECTED + TPDATA.incrementUnexpectedTests() + exitStatus = TestExitStatus.UNEXPECTED exitStatusInfo = rep.outcome # [2025-03-28] It may create a useless problem in new environment. # assert False # -------- if item_warning_msg_count > 0: - TEST_PROCESS_STATS.incrementWarningTestCount(testID, item_warning_msg_count) + TPDATA.incrementWarningTestCount(testID, item_warning_msg_count) # -------- assert exitStatus is not None - assert type(exitStatus) is str + assert type(exitStatus) is TestExitStatus - if exitStatus == ExitStatusNames.FAILED: - assert item_error_msg_count > 0 - pass + assert exitStatus != TestExitStatus.FAILED or item_error_msg_count > 0 + + TestServices.CleanTestTmpDirBeforeExit( + item, + exitStatus, + ) # -------- - assert type(TEST_PROCESS_STATS.cTotalDuration) is datetime.timedelta + assert type(TPDATA.cTotalDuration) is datetime.timedelta assert type(testDurration) is datetime.timedelta - TEST_PROCESS_STATS.cTotalDuration += testDurration + TPDATA.cTotalDuration += testDurration - assert testDurration <= TEST_PROCESS_STATS.cTotalDuration + assert testDurration <= TPDATA.cTotalDuration # -------- - exitStatusLineData = exitStatus + exitStatusLineData = exitStatus.value if exitStatusInfo is not None: exitStatusLineData += " [{}]".format(exitStatusInfo) @@ -637,6 +545,7 @@ def helper__makereport__call( logging.info("*") logging.info("* STOP TEST {0}".format(testID)) logging.info("*") + return # ///////////////////////////////////////////////////////////////////////////// @@ -689,6 +598,7 @@ def pytest_runtest_makereport(item: pytest.Function, call: pytest.CallInfo): class LogWrapper2: + _guard: threading.Lock _old_method: typing.Any _err_counter: typing.Optional[int] _warn_counter: typing.Optional[int] @@ -697,14 +607,18 @@ class LogWrapper2: # -------------------------------------------------------------------- def __init__(self): + self._guard = threading.Lock() self._old_method = None self._err_counter = None self._warn_counter = None self._critical_counter = None + return # -------------------------------------------------------------------- def __enter__(self): + assert self._guard is not None + # assert isinstance(self._guard, threading.Lock) assert self._old_method is None assert self._err_counter is None assert self._warn_counter is None @@ -725,6 +639,8 @@ def __enter__(self): # -------------------------------------------------------------------- def __exit__(self, exc_type, exc_val, exc_tb): + assert self._guard is not None + # assert isinstance(self._guard, threading.Lock) assert self._old_method is not None assert self._err_counter is not None assert self._warn_counter is not None @@ -746,6 +662,8 @@ def __exit__(self, exc_type, exc_val, exc_tb): def __call__(self, record: logging.LogRecord): assert record is not None assert isinstance(record, logging.LogRecord) + assert self._guard is not None + # assert isinstance(self._guard, threading.Lock) assert self._old_method is not None assert self._err_counter is not None assert self._warn_counter is not None @@ -760,15 +678,16 @@ def __call__(self, record: logging.LogRecord): r = self._old_method(record) - if record.levelno == logging.ERROR: - self._err_counter += 1 - assert self._err_counter > 0 - elif record.levelno == logging.WARNING: - self._warn_counter += 1 - assert self._warn_counter > 0 - elif record.levelno == logging.CRITICAL: - self._critical_counter += 1 - assert self._critical_counter > 0 + with self._guard: + if record.levelno == logging.ERROR: + self._err_counter += 1 + assert self._err_counter > 0 + elif record.levelno == logging.WARNING: + self._warn_counter += 1 + assert self._warn_counter > 0 + elif record.levelno == logging.CRITICAL: + self._critical_counter += 1 + assert self._critical_counter > 0 return r @@ -938,28 +857,24 @@ def pytest_sessionfinish(): # NOTE: It should execute after logging.pytest_sessionfinish # - global g_test_process_kind # noqa: F824 - global g_test_process_mode # noqa: F824 - global g_worker_log_is_created # noqa: F824 - - assert g_test_process_kind is not None - assert type(g_test_process_kind) is T_TEST_PROCESS_KIND + assert TPDATA.test_process_kind is not None + assert type(TPDATA.test_process_kind) is T_TEST_PROCESS_KIND - if g_test_process_kind == T_TEST_PROCESS_KIND.Master: + if TPDATA.test_process_kind == T_TEST_PROCESS_KIND.Master: return - assert g_test_process_kind == T_TEST_PROCESS_KIND.Worker + assert TPDATA.test_process_kind == T_TEST_PROCESS_KIND.Worker - assert g_test_process_mode is not None - assert type(g_test_process_mode) is T_TEST_PROCESS_MODE + assert TPDATA.test_process_mode is not None + assert type(TPDATA.test_process_mode) is T_TEST_PROCESS_MODE - if g_test_process_mode == T_TEST_PROCESS_MODE.Collect: + if TPDATA.test_process_mode == T_TEST_PROCESS_MODE.Collect: return - assert g_test_process_mode == T_TEST_PROCESS_MODE.ExecTests + assert TPDATA.test_process_mode == T_TEST_PROCESS_MODE.ExecTests - assert type(g_worker_log_is_created) is bool - assert g_worker_log_is_created + assert type(TPDATA.worker_log_is_created) is bool + assert TPDATA.worker_log_is_created C_LINE1 = "---------------------------" @@ -1004,67 +919,67 @@ def LOCAL__print_test_list2( # fmt: off LOCAL__print_test_list( "ACHTUNG TESTS", - TEST_PROCESS_STATS.cAchtungTests, - TEST_PROCESS_STATS.AchtungTests, + TPDATA.cAchtungTests, + TPDATA.AchtungTests, ) LOCAL__print_test_list2( "FAILED TESTS", - TEST_PROCESS_STATS.cFailedTests, - TEST_PROCESS_STATS.FailedTests + TPDATA.cFailedTests, + TPDATA.FailedTests ) LOCAL__print_test_list2( "XFAILED TESTS", - TEST_PROCESS_STATS.cXFailedTests, - TEST_PROCESS_STATS.XFailedTests, + TPDATA.cXFailedTests, + TPDATA.XFailedTests, ) LOCAL__print_test_list( "NOT XFAILED TESTS", - TEST_PROCESS_STATS.cNotXFailedTests, - TEST_PROCESS_STATS.NotXFailedTests, + TPDATA.cNotXFailedTests, + TPDATA.NotXFailedTests, ) LOCAL__print_test_list2( "WARNING TESTS", - TEST_PROCESS_STATS.cWarningTests, - TEST_PROCESS_STATS.WarningTests, + TPDATA.cWarningTests, + TPDATA.WarningTests, ) # fmt: on LOCAL__print_line1_with_header("SUMMARY STATISTICS") logging.info("") logging.info("[TESTS]") - logging.info(" TOTAL : {0}".format(TEST_PROCESS_STATS.cTotalTests)) - logging.info(" EXECUTED : {0}".format(TEST_PROCESS_STATS.cExecutedTests)) - logging.info(" NOT EXECUTED : {0}".format(TEST_PROCESS_STATS.cNotExecutedTests)) - logging.info(" ACHTUNG : {0}".format(TEST_PROCESS_STATS.cAchtungTests)) + logging.info(" TOTAL : {0}".format(TPDATA.cTotalTests)) + logging.info(" EXECUTED : {0}".format(TPDATA.cExecutedTests)) + logging.info(" NOT EXECUTED : {0}".format(TPDATA.cNotExecutedTests)) + logging.info(" ACHTUNG : {0}".format(TPDATA.cAchtungTests)) logging.info("") - logging.info(" PASSED : {0}".format(TEST_PROCESS_STATS.cPassedTests)) - logging.info(" FAILED : {0}".format(TEST_PROCESS_STATS.cFailedTests)) - logging.info(" XFAILED : {0}".format(TEST_PROCESS_STATS.cXFailedTests)) - logging.info(" NOT XFAILED : {0}".format(TEST_PROCESS_STATS.cNotXFailedTests)) - logging.info(" SKIPPED : {0}".format(TEST_PROCESS_STATS.cSkippedTests)) - logging.info(" WITH WARNINGS: {0}".format(TEST_PROCESS_STATS.cWarningTests)) - logging.info(" UNEXPECTED : {0}".format(TEST_PROCESS_STATS.cUnexpectedTests)) + logging.info(" PASSED : {0}".format(TPDATA.cPassedTests)) + logging.info(" FAILED : {0}".format(TPDATA.cFailedTests)) + logging.info(" XFAILED : {0}".format(TPDATA.cXFailedTests)) + logging.info(" NOT XFAILED : {0}".format(TPDATA.cNotXFailedTests)) + logging.info(" SKIPPED : {0}".format(TPDATA.cSkippedTests)) + logging.info(" WITH WARNINGS: {0}".format(TPDATA.cWarningTests)) + logging.info(" UNEXPECTED : {0}".format(TPDATA.cUnexpectedTests)) logging.info("") - assert type(TEST_PROCESS_STATS.cTotalDuration) is datetime.timedelta + assert type(TPDATA.cTotalDuration) is datetime.timedelta LOCAL__print_line1_with_header("TIME") logging.info("") logging.info( " TOTAL DURATION: {0}".format( - timedelta_to_human_text(TEST_PROCESS_STATS.cTotalDuration) + timedelta_to_human_text(TPDATA.cTotalDuration) ) ) logging.info("") LOCAL__print_line1_with_header("TOTAL INFORMATION") logging.info("") - logging.info(" TOTAL ERROR COUNT : {0}".format(TEST_PROCESS_STATS.cTotalErrors)) - logging.info(" TOTAL WARNING COUNT: {0}".format(TEST_PROCESS_STATS.cTotalWarnings)) + logging.info(" TOTAL ERROR COUNT : {0}".format(TPDATA.cTotalErrors)) + logging.info(" TOTAL WARNING COUNT: {0}".format(TPDATA.cTotalWarnings)) logging.info("") @@ -1107,7 +1022,9 @@ def helper__pytest_configure__logging(config: pytest.Config) -> None: pathlib.Path(log_dir).mkdir(exist_ok=True) - logging_plugin = config.pluginmanager.get_plugin("logging-plugin") + logging_plugin = config.pluginmanager.get_plugin( + "logging-plugin", + ) assert logging_plugin is not None assert isinstance(logging_plugin, _pytest.logging.LoggingPlugin) @@ -1125,31 +1042,27 @@ def helper__pytest_configure__logging(config: pytest.Config) -> None: def pytest_configure(config: pytest.Config) -> None: assert isinstance(config, pytest.Config) - global g_test_process_kind - global g_test_process_mode - global g_worker_log_is_created - - assert g_test_process_kind is None - assert g_test_process_mode is None - assert g_worker_log_is_created is None + assert TPDATA.test_process_kind is None + assert TPDATA.test_process_mode is None + assert TPDATA.worker_log_is_created is None - g_test_process_mode = helper__detect_test_process_mode(config) - g_test_process_kind = helper__detect_test_process_kind(config) + TPDATA.test_process_mode = helper__detect_test_process_mode(config) + TPDATA.test_process_kind = helper__detect_test_process_kind(config) - assert type(g_test_process_kind) is T_TEST_PROCESS_KIND - assert type(g_test_process_mode) is T_TEST_PROCESS_MODE + assert type(TPDATA.test_process_kind) is T_TEST_PROCESS_KIND + assert type(TPDATA.test_process_mode) is T_TEST_PROCESS_MODE - if g_test_process_kind == T_TEST_PROCESS_KIND.Master: + if TPDATA.test_process_kind == T_TEST_PROCESS_KIND.Master: pass else: - assert g_test_process_kind == T_TEST_PROCESS_KIND.Worker + assert TPDATA.test_process_kind == T_TEST_PROCESS_KIND.Worker - if g_test_process_mode == T_TEST_PROCESS_MODE.Collect: - g_worker_log_is_created = False + if TPDATA.test_process_mode == T_TEST_PROCESS_MODE.Collect: + TPDATA.worker_log_is_created = False else: - assert g_test_process_mode == T_TEST_PROCESS_MODE.ExecTests + assert TPDATA.test_process_mode == T_TEST_PROCESS_MODE.ExecTests helper__pytest_configure__logging(config) - g_worker_log_is_created = True + TPDATA.worker_log_is_created = True return diff --git a/tests/conftest_helpers.py b/tests/conftest_helpers.py new file mode 100644 index 00000000..099823a1 --- /dev/null +++ b/tests/conftest_helpers.py @@ -0,0 +1,802 @@ +# ///////////////////////////////////////////////////////////////////////////// +# PyTest Configuration Helpers + +from __future__ import annotations + +import os +import pytest +import logging +import time +import traceback +import typing +import shutil +import enum +import fnmatch +import datetime + + +# ///////////////////////////////////////////////////////////////////////////// + +C_ROOT_DIR__RELATIVE = ".." + + +# ///////////////////////////////////////////////////////////////////////////// +# TestExitStatus + +class TestExitStatus(enum.Enum): + FAILED = "FAILED" + PASSED = "PASSED" + XFAILED = "XFAILED" + NOT_XFAILED = "NOT XFAILED" + SKIPPED = "SKIPPED" + UNEXPECTED = "UNEXPECTED" + + +# ///////////////////////////////////////////////////////////////////////////// +# TestConfigPropNames + + +class TestConfigPropNames: + TEST_CFG__NO_CLEANUP = "TEST_CFG__NO_CLEANUP" + + TEST_CFG__TEMP_DIR = "TEST_CFG__TEMP_DIR" + + TEST_CFG__LOG_DIR = "TEST_CFG__LOG_DIR" + + TEST_CFG__ENABLE_XFAIL = "TEST_CFG__ENABLE_XFAIL" + + +# ///////////////////////////////////////////////////////////////////////////// +# ThrowError + + +class ThrowError: + @staticmethod + def EnvVarIsNotDefined(envVarName: str) -> typing.NoReturn: + assert type(envVarName) is str + raise RuntimeError("System env variable [{}] is not defined.".format( + envVarName, + )) + + # -------------------------------------------------------------------- + @staticmethod + def EnvVarHasBadValue(envVarName: str) -> typing.NoReturn: + assert type(envVarName) is str + raise RuntimeError("System env variable [{}] has bad value.".format( + envVarName, + )) + + # -------------------------------------------------------------------- + @staticmethod + def EnvVarHasBadValue2(envVarName: str, envVarValue) -> typing.NoReturn: + assert type(envVarName) is str + raise RuntimeError("System env variable [{}] has bad value [{}].".format( + envVarName, + envVarValue, + )) + +# ///////////////////////////////////////////////////////////////////////////// +# TestConfigHelper + + +class TestConfigHelper: + @staticmethod + def NoCleanup( + exit_status: typing.Optional[TestExitStatus], + ) -> bool: + assert exit_status is None or type(exit_status) is TestExitStatus + + v = os.environ.get(TestConfigPropNames.TEST_CFG__NO_CLEANUP) + + if v is None: + return False + + vv = str(v).upper() + + if vv in __class__.sm_NO: + return False + + if vv in __class__.sm_YES: + return True + + if exit_status is None: + return False + + v2 = vv.split(",") + + if exit_status.name in v2: + return True + + return False + + # -------------------------------------------------------------------- + @staticmethod + def EnableXFail() -> bool: + if TestConfigPropNames.TEST_CFG__ENABLE_XFAIL not in os.environ.keys(): + return False + + v = os.environ[TestConfigPropNames.TEST_CFG__ENABLE_XFAIL] + + return __class__.Helper__ToBoolean(v, TestConfigPropNames.TEST_CFG__ENABLE_XFAIL) + + # -------------------------------------------------------------------- + @staticmethod + def GetEnvValue__STR(envName: str) -> typing.Optional[str]: + assert type(envName) is str + return os.getenv(envName) + + # -------------------------------------------------------------------- + @staticmethod + def GetReqEnvValue__STR(envName: str) -> str: + assert type(envName) is str + + v = os.getenv(envName) + + if v is None: + ThrowError.EnvVarIsNotDefined(envName) + assert False + + return v + + # Helper methods ----------------------------------------------------- + sm_YES: list[str] = ["1", "TRUE", "YES", "ON"] + + sm_NO: list[str] = ["0", "FALSE", "NO", "OFF"] + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ToBoolean(v, envVarName: str) -> bool: + assert type(envVarName) is str + + typeV = type(v) + + if typeV is bool: + return v + + if typeV is str: + vv = str(v).upper() + assert type(vv) is str + + if vv in __class__.sm_YES: + return True + + if vv in __class__.sm_NO: + return False + + ThrowError.EnvVarHasBadValue2(envVarName, vv) + return False + + if typeV is int: + if v == 0: + return False + + if v == 1: + return True + + ThrowError.EnvVarHasBadValue2(envVarName, v) + return False + + ThrowError.EnvVarHasBadValue(envVarName) + return False + +# ///////////////////////////////////////////////////////////////////////////// +# TestStartupData__Helper + + +class TestStartupData__Helper: + sm_StartTS = datetime.datetime.now() + + # -------------------------------------------------------------------- + @staticmethod + def GetStartTS() -> datetime.datetime: + assert type(__class__.sm_StartTS) is datetime.datetime + return __class__.sm_StartTS + + # -------------------------------------------------------------------- + @staticmethod + def CalcRootDir() -> str: + r = os.path.abspath(__file__) + r = os.path.dirname(r) + r = os.path.join(r, C_ROOT_DIR__RELATIVE) + r = os.path.abspath(r) + return r + + # -------------------------------------------------------------------- + @staticmethod + def CalcRootTmpDir() -> str: + if TestConfigPropNames.TEST_CFG__TEMP_DIR in os.environ: + resultPath = os.environ[TestConfigPropNames.TEST_CFG__TEMP_DIR] + else: + rootDir = __class__.CalcRootDir() + resultPath = os.path.join(rootDir, "tmp") + + assert type(resultPath) is str + return resultPath + + # -------------------------------------------------------------------- + @staticmethod + def CalcRootLogDir() -> str: + if TestConfigPropNames.TEST_CFG__LOG_DIR in os.environ: + resultPath = os.environ[TestConfigPropNames.TEST_CFG__LOG_DIR] + else: + rootDir = __class__.CalcRootDir() + resultPath = os.path.join(rootDir, "logs") + + assert type(resultPath) is str + return resultPath + + # -------------------------------------------------------------------- + @staticmethod + def CalcCurrentTestWorkerSignature() -> str: + currentPID = os.getpid() + assert type(currentPID) is int + + startTS = __class__.sm_StartTS + assert type(startTS) is datetime.datetime + + result = "pytest-{0:04d}{1:02d}{2:02d}_{3:02d}{4:02d}{5:02d}".format( + startTS.year, + startTS.month, + startTS.day, + startTS.hour, + startTS.minute, + startTS.second, + ) + + gwid = os.environ.get("PYTEST_XDIST_WORKER") + + if gwid is not None: + result += "--xdist_" + str(gwid) + + result += "--" + "pid" + str(currentPID) + return result + + +# ///////////////////////////////////////////////////////////////////////////// +# TestStartupData + + +class TestStartupData: + sm_RootDir: str = TestStartupData__Helper.CalcRootDir() + sm_RootTmpDir: str = TestStartupData__Helper.CalcRootTmpDir() + sm_RootTmpDataDir: str = os.path.join(sm_RootTmpDir, "data") + sm_CurrentTestWorkerSignature: str = ( + TestStartupData__Helper.CalcCurrentTestWorkerSignature() + ) + sm_RootTmpDataDirForCurrentTestWorker: str = os.path.join( + sm_RootTmpDataDir, sm_CurrentTestWorkerSignature + ) + + sm_RootLogDir: str = TestStartupData__Helper.CalcRootLogDir() + + # -------------------------------------------------------------------- + @staticmethod + def GetRootDir() -> str: + assert type(__class__.sm_RootDir) is str + return __class__.sm_RootDir + + # -------------------------------------------------------------------- + @staticmethod + def GetRootLogDir() -> str: + assert type(__class__.sm_RootLogDir) is str + return __class__.sm_RootLogDir + + # -------------------------------------------------------------------- + @staticmethod + def GetCurrentTestWorkerSignature() -> str: + assert type(__class__.sm_CurrentTestWorkerSignature) is str + return __class__.sm_CurrentTestWorkerSignature + + # -------------------------------------------------------------------- + @staticmethod + def GetRootTmpDataDirForCurrentTestWorker() -> str: + assert type(__class__.sm_RootTmpDataDirForCurrentTestWorker) is str + return __class__.sm_RootTmpDataDirForCurrentTestWorker + + +# ///////////////////////////////////////////////////////////////////////////// +# class TestTempDirCleaner2 + + +class TestTempDirCleaner2: + @staticmethod + def exec( + root_path: str, + artifact_patterns: typing.Iterable[str], + ): + assert type(root_path) is str + assert isinstance(artifact_patterns, typing.Iterable) + + stack: typing.List[__class__.tagStackItem] = [] + + __class__.Helper__push(stack, root_path) + + while len(stack) > 0: + head = stack[-1] + assert type(head) is __class__.tagStackItem + + if len(head.data[1]) > 0: + __class__.Helper__push( + stack, + os.path.join(head.path, head.data[1].pop()), + ) + continue + + stack.pop() + + # delete files + cNotDeleted = 0 + for f in head.data[2]: + assert type(f) is str + assert f != "" + + full_file_path = os.path.join(head.path, f) + + # Получаем путь относительно корня очистки для проверки масок (например: "pg_wal/0001.history") + # Это нужно, чтобы работали маски вида "**/pg_wal/*.history" + rel_path = os.path.relpath(full_file_path, root_path) + + # Проверяем, подходит ли файл под какую-либо маску артефактов + is_artifact = False + for pattern in artifact_patterns: + # fnmatch отлично понимает структуры с '/' и '*' + if fnmatch.fnmatch(rel_path, pattern) or fnmatch.fnmatch(f, pattern): + is_artifact = True + break + + if is_artifact: + cNotDeleted += 1 + continue + + if not __class__.Helper__safe_delete_file(full_file_path): + cNotDeleted += 1 + continue + + if cNotDeleted > 0: + continue + + x = __class__.Helper__walk(head.path) + + if x is None: + # ACHTUNG + continue + + if len(x[1]) > 0: + continue + + if len(x[2]) > 0: + continue + + __class__.Helper__safe_delete_dir(head.path) + continue + return + + # -------------------------------------------------------------------- + T_WALK_RESULT = typing.Tuple[ + str, + typing.List[str], + typing.List[str], + ] + + # -------------------------------------------------------------------- + class tagStackItem: + path: str + data: TestTempDirCleaner2.T_WALK_RESULT + + def __init__(self, path, data): + assert type(data) is tuple + assert len(data) == 3 + assert type(data[0]) is str + assert type(data[1]) is list # dirs + assert type(data[2]) is list # files + + self.path = path + self.data = data + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__walk(path: str) -> typing.Optional[T_WALK_RESULT]: + try: + x = os.walk(path).__next__() + except StopIteration: + return None + + assert x is not None + assert type(x) is tuple + assert len(x) == 3 + assert type(x[0]) is str + assert type(x[1]) is list # dirs + assert type(x[2]) is list # files + return x + + # -------------------------------------------------------------------- + @staticmethod + def Helper__push( + stack: typing.List[tagStackItem], + path: str, + ) -> bool: + assert type(path) is str + assert path != "" + x = __class__.Helper__walk(path) + if x is None: + # ACHTUNG + return False + stack.append(__class__.tagStackItem(path, x)) + return True + + # -------------------------------------------------------------------- + @staticmethod + def Helper__safe_delete_file(path: str) -> bool: + assert type(path) is str + assert path != "" + try: + os.remove(path) + except Exception as e: + msg = "File [{}] is not deleted. Reason ({}): {}".format( + path, + type(e).__name__, + e, + ) + logging.info(msg) + return False + return True + + # -------------------------------------------------------------------- + @staticmethod + def Helper__safe_delete_dir(path: str) -> bool: + assert type(path) is str + assert path != "" + try: + os.rmdir(path) + except Exception as e: + msg = "Dir [{}] is not deleted. Reason ({}): {}".format( + path, + type(e).__name__, + e, + ) + logging.info(msg) + return False + return True + +# ///////////////////////////////////////////////////////////////////////////// +# TestServices + + +class TestServices: + C_UNPACKED_TMP_DIR_SIZE_TRESHOLD = 5 * 1024 * 1024 + + # -------------------------------------------------------------------- + @staticmethod + def GetRootDir() -> str: + return TestStartupData.GetRootDir() + + # -------------------------------------------------------------------- + @staticmethod + def GetRootTmpDir() -> str: + return TestStartupData.GetRootTmpDataDirForCurrentTestWorker() + + # -------------------------------------------------------------------- + @staticmethod + def MakeRootTmpDirForGlobalResources(globalResourceID: str) -> str: + assert isinstance(globalResourceID, str) + return os.path.join(__class__.GetRootTmpDir(), ".global", globalResourceID) + + # -------------------------------------------------------------------- + @staticmethod + def GetCurTestTmpDir(request: pytest.FixtureRequest) -> str: + assert isinstance(request, pytest.FixtureRequest) + return __class__.Helper__GetCurTestTmpDir(request.node) + + # -------------------------------------------------------------------- + @staticmethod + def Helper__GetCurTestTmpDir(function: pytest.Function) -> str: + assert isinstance(function, pytest.Function) + + rootDir = TestServices.GetRootDir() + rootTmpDir = TestServices.GetRootTmpDir() + + # [2024-12-18] It is not a fact now. + # assert rootTmpDir.startswith(rootDir) + + testPath = str(function.path) + + if not testPath.startswith(rootDir): + raise Exception( + "Root dir {0} is not found in testPath {1}.".format(rootDir, testPath) + ) + + testPath2 = testPath[len(rootDir) + 1:] + + result = os.path.join(rootTmpDir, testPath2) + + if function.cls is not None: + clsName = function.cls.__name__ + result = os.path.join(result, clsName) + + result = os.path.join(result, function.name) + + return result + + # -------------------------------------------------------------------- + sm_ArtifactRules = { + "**/*.log", + "**/*.conf", + } + + # -------------------------------------------------------------------- + @staticmethod + def CleanTestTmpDirBeforeExit( + function: pytest.Function, + exit_status: TestExitStatus, + ): + assert isinstance(function, pytest.Function) + assert type(exit_status) is TestExitStatus + + tmpDir = __class__.Helper__GetCurTestTmpDir(function) + assert type(tmpDir) is str + + if not os.path.exists(tmpDir): + return + + if TestConfigHelper.NoCleanup(exit_status): + logging.info("A final data cleanup is disabled [test exit status is {}].".format( + exit_status.name, + )) + else: + logging.info("Tmp directory [{}] is cleaned...".format( + tmpDir, + )) + + TestTempDirCleaner2.exec( + tmpDir, + __class__.sm_ArtifactRules, + ) + + if not os.path.exists(tmpDir): + return + + tmpDirSize = __class__.Helper__GetFolderSize(tmpDir) + + if tmpDirSize < __class__.C_UNPACKED_TMP_DIR_SIZE_TRESHOLD: + return + + logging.info("Tmp directory [{}] will be archived [size: {}]...".format( + tmpDir, + __class__.Helper__FormatSize(tmpDirSize), + )) + + shutil.make_archive( + tmpDir, + 'zip', + tmpDir, + ) + + shutil.rmtree(tmpDir) + return + + # -------------------------------------------------------------------- + @staticmethod + def CleanDirBeforeExit(dir_path: str): + assert type(dir_path) is str + + if not os.path.exists(dir_path): + return + + if TestConfigHelper.NoCleanup(None): + logging.info("A final data cleanup is disabled.") + return + + logging.info("Directory [{0}] is cleaned...".format(dir_path)) + + TestTempDirCleaner2.exec(dir_path, __class__.sm_ArtifactRules) + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__FormatSize(bytes_size: int) -> str: + assert type(bytes_size) is int + assert bytes_size >= 0 + + bytes_size_f = float(bytes_size) + for unit in ('B', 'KB', 'MB', 'GB', 'TB'): + if bytes_size_f < 1024: + return __class__.Helper__FormatSizeBuilder(bytes_size_f, unit) + bytes_size_f /= 1024 + continue + return __class__.Helper__FormatSizeBuilder(bytes_size_f, "PB") + + # -------------------------------------------------------------------- + @staticmethod + def Helper__FormatSizeBuilder( + bytes_size_f: float, + unit: str, + ) -> str: + assert type(bytes_size_f) is float + assert bytes_size_f >= 0 + assert type(unit) is str + + return "{:.2f} {}".format( + bytes_size_f, + unit, + ) + + # -------------------------------------------------------------------- + @staticmethod + def Helper__GetFolderSize(folder_path: str) -> int: + total_size = 0 + + for root, _, files in os.walk(folder_path): + for file in files: + file_path = os.path.join(root, file) + # Пропускаем символические ссылки, чтобы избежать ошибок и зацикливания + if not os.path.islink(file_path): + total_size += os.path.getsize(file_path) + continue + continue + + return total_size + + # -------------------------------------------------------------------- + @staticmethod + def PrintExceptionOK(e: Exception): + assert isinstance(e, Exception) + + logging.info( + "OK. We catch an exception. {}".format( + __class__.ExceptionToHumanText(e), + ) + ) + return + + # -------------------------------------------------------------------- + @staticmethod + def ThrowWeWaitAnException() -> typing.NoReturn: + raise Exception("We wait an exception!") + + # -------------------------------------------------------------------- + @staticmethod + def ThrowWeWaitAnXFailPleaseUpdateTest() -> typing.NoReturn: + raise Exception("We wait an xfail, please update test!") + + # -------------------------------------------------------------------- + @staticmethod + def LogCurrentExceptionAndThrowXFailItIsAnExpectedFailure() -> typing.NoReturn: + logging.exception("We catch an expected problem.") + raise pytest.xfail("It is an expected failure.") + + # -------------------------------------------------------------------- + @staticmethod + def SleepWithPrint( + sleepTimeInSec: float, + message: typing.Optional[str] = None, + ): + assert message is None or type(message) is str + + prefix = "" + + if message is not None and message != "": + prefix = message + + if not prefix.endswith("."): + prefix += "." + + prefix += " " + + logging.info("{}Sleep {} second(s).".format( + prefix, + sleepTimeInSec, + )) + time.sleep(sleepTimeInSec) + return + + # -------------------------------------------------------------------- + @staticmethod + def ExceptionToHumanText(exc: BaseException) -> str: + assert isinstance(exc, BaseException) + + if __class__.Helper__GetPrevExc(exc) is None: + return __class__.Helper__ExceptionToHumanText__Single(exc) + + return __class__.Helper__ExceptionToHumanText__Chain(exc) + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ExceptionToHumanText__Single(exc: BaseException) -> str: + assert isinstance(exc, BaseException) + + if isinstance(exc, AssertionError): + err_msg = "Exception ({}).".format(type(exc).__name__) + assert type(err_msg) is str + + exc_info_lines = traceback.format_exception(exc) + assert type(exc_info_lines) is list + err_msg2 = "".join(exc_info_lines).strip() + + if err_msg2 != "": + err_msg += " " + err_msg2 + + assert type(err_msg) is str + return err_msg + + err_msg = "Exception ({})".format(type(exc).__name__) + assert type(err_msg) is str + + err_msg2 = str(exc).strip() + + if err_msg2 == "": + err_msg += "." + else: + err_msg += ": " + err_msg2 + + assert type(err_msg) is str + return err_msg + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ExceptionToHumanText__Chain(exc: BaseException) -> str: + assert isinstance(exc, BaseException) + + chain: typing.List[BaseException] = [] + curr = exc + + processed_exc_ids: typing.Set[int] = set() + + while curr is not None: + assert isinstance(curr, BaseException) + if id(curr) in processed_exc_ids: + logging.error("Cycle in exception chain: {}".format( + " --> ".join( + [type(x).__name__ for x in reversed(chain)], + ) + )) + break + + processed_exc_ids.add(id(curr)) + + chain.append(curr) + curr = __class__.Helper__GetPrevExc(curr) + continue + + lines: typing.List[str] = [] + + lines.append("It is a chain of exceptions (len: {}):".format( + len(chain), + )) + + n = 0 + for e in reversed(chain): + n += 1 + line1 = "---- {}. Exception ({})".format(n, type(e).__name__) + line2 = __class__.Helper__GetExcMsg(e) + + if line2 == "": + lines.append(line1) + else: + lines.append(line1 + ":") + lines.append(line2) + continue + + lines.append("--------") + + return "\n".join(lines) + + # -------------------------------------------------------------------- + @staticmethod + def Helper__GetPrevExc(exc: BaseException) -> typing.Optional[BaseException]: + assert isinstance(exc, BaseException) + return exc.__cause__ or exc.__context__ + + # -------------------------------------------------------------------- + @staticmethod + def Helper__GetExcMsg(exc: BaseException) -> str: + assert isinstance(exc, BaseException) + + if isinstance(exc, AssertionError): + exc_info_lines = traceback.format_exception(exc) + assert type(exc_info_lines) is list + return "".join(exc_info_lines).strip() + + return str(exc) + +# ///////////////////////////////////////////////////////////////////////////// diff --git a/tests/helpers/multi_try_call.py b/tests/helpers/multi_try_call.py new file mode 100755 index 00000000..9fc16ab5 --- /dev/null +++ b/tests/helpers/multi_try_call.py @@ -0,0 +1,433 @@ +# ////////////////////////////////////////////////////////////////////////////// +from __future__ import annotations + +import typing +import logging +import time +import datetime +import dataclasses + + +# ////////////////////////////////////////////////////////////////////////////// +# MultiTryCall + + +class MultiTryCall: + T_SLEEP_SECONDS = typing.Union[int, float] + + # -------------------------------------------------------------------- + @dataclasses.dataclass + class tagSETTINGS: + T_EQUAL_COMPARER = typing.Callable[[typing.Any, typing.Any], bool] + + max_attempts: typing.Optional[int] = None + sleep_seconds: MultiTryCall.T_SLEEP_SECONDS = 0.5 + suppressed_exceptions: typing.Optional[typing.List[type]] = None + end_ts: typing.Optional[datetime.datetime] = None + + stable_count: typing.Optional[int] = None + stable_tester: typing.Optional[T_EQUAL_COMPARER] = None + + # -------------------------------------------------------------------- + @staticmethod + def exec( + method: typing.Callable, + operationDescr: str, + settings: tagSETTINGS, + *args, + **kwargs + ) -> typing.Any: + assert isinstance(method, typing.Callable) + assert type(operationDescr) is str + assert type(settings) is __class__.tagSETTINGS + assert type(settings.sleep_seconds) is int or type(settings.sleep_seconds) is float + assert operationDescr != "" + assert settings.max_attempts is None or settings.max_attempts > 0 + assert type(settings.sleep_seconds) in [int, float] + assert float(settings.sleep_seconds) >= 0 + assert settings.suppressed_exceptions is None or type(settings.suppressed_exceptions) is list + + return __class__.helper__exec__until( + method, + __class__.helper__return_true, + operationDescr, + settings, + *args, + **kwargs + ) + + # -------------------------------------------------------------------- + @dataclasses.dataclass + class tagTryResult: + ok: bool + value: typing.Any + fail_reason: typing.Optional[str] + last_exception: typing.Optional[BaseException] + + # -------------------------------------------------------------------- + @staticmethod + def try_exec__test( + testMethod: typing.Callable[..., bool], + operationDescr: str, + settings: tagSETTINGS, + *args, + **kwargs, + ) -> tagTryResult: + assert isinstance(testMethod, typing.Callable) + assert type(operationDescr) is str + assert type(settings) is __class__.tagSETTINGS + assert type(settings.sleep_seconds) is int or type(settings.sleep_seconds) is float + assert operationDescr != "" + assert settings.max_attempts is None or settings.max_attempts > 0 + assert type(settings.sleep_seconds) in [int, float] + assert float(settings.sleep_seconds) >= 0 + assert settings.suppressed_exceptions is None or type(settings.suppressed_exceptions) is list + + return __class__.helper__try_exec__until( + testMethod, + __class__.helper__is_true, + operationDescr, + settings, + *args, + **kwargs, + ) + + # -------------------------------------------------------------------- + @staticmethod + def try_exec__until( + execMethod: typing.Callable[..., typing.Any], + testMethod: typing.Callable[[typing.Any], bool], + operationDescr: str, + settings: tagSETTINGS, + *args, + **kwargs, + ) -> tagTryResult: + assert isinstance(execMethod, typing.Callable) + assert isinstance(testMethod, typing.Callable) + assert type(operationDescr) is str + assert type(settings) is __class__.tagSETTINGS + assert type(settings.sleep_seconds) is int or type(settings.sleep_seconds) is float + assert operationDescr != "" + assert settings.max_attempts is None or settings.max_attempts > 0 + assert type(settings.sleep_seconds) in [int, float] + assert float(settings.sleep_seconds) >= 0 + assert settings.suppressed_exceptions is None or type(settings.suppressed_exceptions) is list + + return __class__.helper__try_exec__until( + execMethod, + testMethod, + operationDescr, + settings, + *args, + **kwargs, + ) + + # -------------------------------------------------------------------- + @staticmethod + def exec__test( + testMethod: typing.Callable[..., bool], + operationDescr: str, + settings: tagSETTINGS, + *args, + **kwargs, + ) -> None: + assert isinstance(testMethod, typing.Callable) + assert type(operationDescr) is str + assert type(settings) is __class__.tagSETTINGS + assert type(settings.sleep_seconds) is int or type(settings.sleep_seconds) is float + assert operationDescr != "" + assert settings.max_attempts is None or settings.max_attempts > 0 + assert type(settings.sleep_seconds) in [int, float] + assert float(settings.sleep_seconds) >= 0 + assert settings.suppressed_exceptions is None or type(settings.suppressed_exceptions) is list + + __class__.helper__exec__until( + testMethod, + __class__.helper__is_true, + operationDescr, + settings, + *args, + **kwargs, + ) + return + + # -------------------------------------------------------------------- + @staticmethod + def exec__until( + execMethod: typing.Callable[..., typing.Any], + testMethod: typing.Callable[[typing.Any], bool], + operationDescr: str, + settings: tagSETTINGS, + *args, + **kwargs, + ) -> typing.Any: + assert isinstance(execMethod, typing.Callable) + assert isinstance(testMethod, typing.Callable) + assert type(operationDescr) is str + assert type(settings) is __class__.tagSETTINGS + assert type(settings.sleep_seconds) is int or type(settings.sleep_seconds) is float + assert operationDescr != "" + assert settings.max_attempts is None or settings.max_attempts > 0 + assert type(settings.sleep_seconds) in [int, float] + assert float(settings.sleep_seconds) >= 0 + assert settings.suppressed_exceptions is None or type(settings.suppressed_exceptions) is list + + return __class__.helper__exec__until( + execMethod, + testMethod, + operationDescr, + settings, + *args, + **kwargs, + ) + + # -------------------------------------------------------------------- + @staticmethod + def helper__is_true( + value: bool, + ) -> bool: + assert type(value) is bool + return value + + # -------------------------------------------------------------------- + @staticmethod + def helper__exec__until( + execMethod: typing.Callable[..., typing.Any], + testMethod: typing.Callable[[typing.Any], bool], + operationDescr: str, + settings: tagSETTINGS, + *args, + **kwargs, + ) -> typing.Any: + r = __class__.helper__try_exec__until( + execMethod, + testMethod, + operationDescr, + settings, + *args, + **kwargs, + ) + + assert type(r) is __class__.tagTryResult + + if r.ok: + return r.value + + assert type(r.fail_reason) is str + raise RuntimeError(r.fail_reason) from r.last_exception + + # -------------------------------------------------------------------- + @staticmethod + def helper__try_exec__until( + execMethod: typing.Callable[..., typing.Any], + testMethod: typing.Callable[[typing.Any], bool], + operationDescr: str, + settings: tagSETTINGS, + *args, + **kwargs, + ) -> tagTryResult: + assert isinstance(execMethod, typing.Callable) + assert isinstance(testMethod, typing.Callable) + assert type(operationDescr) is str + assert type(settings) is __class__.tagSETTINGS + assert type(settings.sleep_seconds) is int or type(settings.sleep_seconds) is float + assert operationDescr != "" + assert settings.max_attempts is None or settings.max_attempts > 0 + assert settings.sleep_seconds >= 0 + assert settings.suppressed_exceptions is None or type(settings.suppressed_exceptions) is list + + nAttempt = 0 + nStable = 0 + prev_result: typing.Any = None + + while True: + assert settings.max_attempts is None or nAttempt < settings.max_attempts + + if settings.end_ts is not None: + current_ts = datetime.datetime.now(tz=datetime.timezone.utc) + if settings.end_ts < current_ts: + err_msg = "The operation [{}] is timeout. {} attempt(s) made.".format( + operationDescr, + nAttempt, + ) + raise TimeoutError(err_msg) + + nAttempt += 1 + + if nAttempt > 1: + logging.info( + "Sleep [{0}] seconds before the next attempt to do [{1}] ...".format( + settings.sleep_seconds, + operationDescr, + ) + ) + + time.sleep(settings.sleep_seconds) + + logging.info( + "Try to do [{0}]. Attempt {1}/{2} ...".format( + operationDescr, + nAttempt, + settings.max_attempts if settings.max_attempts is not None else "None", + ) + ) + + try: + r = execMethod(*args, **kwargs) + except AssertionError: + raise + except BaseException as e: + nStable = 0 + prev_result = None + + assert settings.max_attempts is None or nAttempt <= settings.max_attempts + + msg = __class__.Helper__build_op_exc_message(operationDescr, e) + logging.info(msg) + + if not __class__.Helper__should_we_suppress_exception( + e, + settings.suppressed_exceptions, + ): + raise + + if settings.max_attempts is None or nAttempt < settings.max_attempts: + continue + + assert nAttempt == settings.max_attempts + + logging.info("It was the last ({}) attempt. Exception will be reraised.".format( + nAttempt + )) + + return __class__.tagTryResult( + ok=False, + value=None, + fail_reason="Operation [{0}] failed. {1} attempts were made.".format( + operationDescr, + nAttempt, + ), + last_exception=e, + ) + + assert nStable >= 0 + + if settings.stable_tester is not None: + assert isinstance(settings.stable_tester, typing.Callable) + + if nStable > 0 and not settings.stable_tester(prev_result, r): + nStable = 0 + + prev_result = r + nStable += 1 + + assert settings.stable_count is not None and settings.stable_count >= 1 + + if nStable < settings.stable_count: + logging.info("Operation [{}] got last_result {} time(s)".format( + operationDescr, + nStable, + )) + continue + + if testMethod(r): + logging.info("Operation [{0}] is succeeded.".format( + operationDescr, + )) + return __class__.tagTryResult( + ok=True, + value=r, + fail_reason=None, + last_exception=None, + ) + + if settings.max_attempts is not None and nAttempt == settings.max_attempts: + return __class__.tagTryResult( + ok=False, + value=None, + fail_reason="Operation [{0}] failed. {1} attempts were made.".format( + operationDescr, + nAttempt, + ), + last_exception=None, + ) + + logging.info("Operation [{0}] still in progress.".format(operationDescr)) + continue + + # -------------------------------------------------------------------- + @staticmethod + def helper__return_true(v: typing.Any) -> bool: + return True + + # Helper methods ----------------------------------------------------- + @staticmethod + def Helper__should_we_suppress_exception( + exc: BaseException, + suppressed_exception: typing.Optional[typing.List[type]], + ) -> bool: + assert isinstance(exc, BaseException) + assert suppressed_exception is None or type(suppressed_exception) is list + + if suppressed_exception is None: + return True + + for t in suppressed_exception: + assert t is not None + assert type(t) is type + assert issubclass(t, BaseException) + + if isinstance(exc, t): + return True + continue + + return False + + # -------------------------------------------------------------------- + @staticmethod + def Helper__build_op_exc_message( + opDescr: str, + exc: BaseException, + ) -> str: + assert type(opDescr) is str + assert isinstance(exc, BaseException) + + if exc.__cause__ is None: + msg = "Operation [{0}] raised the exception ({1}): {2}".format( + opDescr, type(exc).__name__, exc + ) + return msg + + msg = "Operation [{0}] raised the complex exception:".format(opDescr) + + prev = exc + num = 0 + e2: typing.Optional[BaseException] = exc + while e2 is not None: + assert isinstance(e2, BaseException) + assert prev is not None + + num += 1 + + line = "{}. {} - {}".format( + num, + type(e2).__name__, + str(e2).strip() + ) + + msg += "\n" + line + + e2 = e2.__cause__ + + if (num % 2) == 0: + assert prev.__cause__ is not None + prev = prev.__cause__ + + assert e2 is not prev + continue + + assert num > 1 + return msg + + +# ////////////////////////////////////////////////////////////////////////////// diff --git a/tests/requirements.txt b/tests/requirements.txt index fef51be7..fd3d7c75 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -1,8 +1,8 @@ -psutil pytest pytest-env pytest-xdist -psycopg2 +psutil six -testgres.os_ops>=3.3.0,<4.0.0 +psycopg2 +testgres.os_ops>=3.3.3,<4.0.0 testgres.postgres_configuration>=0.2.2,<1.0.0 diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 45c39c6a..d62d606e 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -7,6 +7,9 @@ from tests.helpers.run_conditions import RunConditions from tests.helpers.local_check import LocalCheck from tests.helpers.local_check import OsOpsHelpers +from tests.helpers.multi_try_call import MultiTryCall + +from tests.conftest_helpers import TestServices from testgres.operations.os_ops import OsProcessController from testgres.operations.os_ops import OsCommandResult @@ -2928,7 +2931,10 @@ def test_get_process_children__with_child( assert actual_child_pid == expected_child_pid - child_cmdline = childs[0].cmdline() + child_cmdline = __class__.helper__wait_for_not_empty_proc_cmdline( + childs[0], + ) + assert type(child_cmdline) is list logging.info("child cmdline: {}".format( child_cmdline, @@ -3011,7 +3017,10 @@ def test_get_process_children__with_three_children( assert actual_child_pids == expected_child_pids for i in range(len(childs)): - child_cmdline = childs[i].cmdline() + child_cmdline = __class__.helper__wait_for_not_empty_proc_cmdline( + childs[i], + ) + assert type(child_cmdline) is list logging.info("child cmdline: {}".format( child_cmdline, @@ -4710,7 +4719,98 @@ def test_popen_terminate(self, os_ops_descr: OsOpsDescr): type(os_ops).__name__, )) pass - pass + return + + def test_popen_terminate_mt( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + controller: typing.Optional[OsProcessController] = None + + try: + N_WORKERS = 100 + + logging.info("Process is creating ...") + cmd1 = ["sleep", "100"] + controller = os_ops.popen(cmd1) + assert isinstance(controller, OsProcessController) + + logging.info("Worker are creating ...") + threadPool = ThreadPoolExecutor( + max_workers=N_WORKERS, + thread_name_prefix="ex_creator", + ) + + class tadWorkerData: + future: ThreadFuture + + workerDatas: typing.List[tadWorkerData] = list() + + nErrors = 0 + + try: + for n in range(N_WORKERS): + logging.info("worker #{} is creating ...".format(n)) + + workerDatas.append(tadWorkerData()) + + workerDatas[n].future = threadPool.submit( + controller.terminate, + ) + + assert workerDatas[n].future is not None + + logging.info("OK. All the workers were created!") + except BaseException as e: + nErrors += 1 + logging.error("A problem is detected ({}): {}".format( + type(e).__name__, + TestServices.ExceptionToHumanText(e), + )) + + logging.info("Will wait for stop of all the workers...") + + nWorkers = 0 + + assert type(workerDatas) is list + + for i in range(len(workerDatas)): + worker = workerDatas[i].future + + if worker is None: + break + + nWorkers += 1 + + assert isinstance(worker, ThreadFuture) + + try: + logging.info("Wait for worker #{}".format(i)) + worker.result() + except BaseException as e: + nErrors += 1 + logging.error("Worker #{} finished with error ({}): {}".format( + i, + type(e).__name__, + TestServices.ExceptionToHumanText(e), + )) + continue + + assert nWorkers == N_WORKERS + + if nErrors != 0: + raise RuntimeError("Some problems were detected. Please examine the log messages.") + + finally: + if controller is not None: + controller.close() + return def test_popen_kill(self, os_ops_descr: OsOpsDescr): @@ -6667,3 +6767,28 @@ def helper__bug_check__unknown_os_ops_type( type(os_ops).__name__, ) raise RuntimeError(err_msg) + + @staticmethod + def helper__wait_for_not_empty_proc_cmdline( + proc_info, + ) -> typing.List[str]: + assert proc_info is not None + + def LOCAL__not_empty(v: typing.List[str]) -> bool: + assert v is not None + assert type(v) is list + return len(v) != 0 + + r = MultiTryCall.exec__until( + proc_info.cmdline, + LOCAL__not_empty, + "wait not empty read proc cmdline", + MultiTryCall.tagSETTINGS( + 10, + 1, + ), + ) + + assert type(r) is list + assert len(r) > 0 + return r