bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 11:41:49 +02:00
Code Activity

fix(tests): BECDispatcher fixture putted back

Browse Source
This commit is contained in:
wyzula-jan
2024-02-26 14:27:22 +01:00
parent fd711b475f
commit 644f1031f6
1 changed files with 35 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
tests/conftest.py
View File

@ -1,35 +1,35 @@
# import pytest import pytest
# import threading import threading
#
# from bec_lib.bec_service import BECService from bec_lib.bec_service import BECService
# from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
#
#
# @pytest.fixture() @pytest.fixture()
# def threads_check(): def threads_check():
# current_threads = set( current_threads = set(
# th th
# for th in threading.enumerate() for th in threading.enumerate()
# if "loguru" not in th.name and th is not threading.main_thread() if "loguru" not in th.name and th is not threading.main_thread()
# ) )
# yield yield
# threads_after = set( threads_after = set(
# th th
# for th in threading.enumerate() for th in threading.enumerate()
# if "loguru" not in th.name and th is not threading.main_thread() if "loguru" not in th.name and th is not threading.main_thread()
# ) )
# additional_threads = threads_after - current_threads additional_threads = threads_after - current_threads
# assert ( assert (
# len(additional_threads) == 0 len(additional_threads) == 0
# ), f"Test creates {len(additional_threads)} threads that are not cleaned: {additional_threads}" ), f"Test creates {len(additional_threads)} threads that are not cleaned: {additional_threads}"
#
#
# @pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
# def bec_dispatcher(threads_check): def bec_dispatcher(threads_check):
# bec_dispatcher = bec_dispatcher_module.BECDispatcher() bec_dispatcher = bec_dispatcher_module.BECDispatcher()
# yield bec_dispatcher yield bec_dispatcher
# bec_dispatcher.disconnect_all() bec_dispatcher.disconnect_all()
# # clean BEC client # clean BEC client
# BECService.shutdown(bec_dispatcher.client) BECService.shutdown(bec_dispatcher.client)
# # reinitialize singleton for next test # reinitialize singleton for next test
# bec_dispatcher_module._bec_dispatcher = None bec_dispatcher_module._bec_dispatcher = None