0
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 03:31:50 +02:00

feat: add rpc broadcast

This commit is contained in:
2025-03-13 10:06:34 +01:00
committed by wyzula-jan
parent 9f2a083abb
commit 2ba9b4cb23
14 changed files with 780 additions and 635 deletions

View File

@ -1,81 +1,59 @@
"""This module contains fixtures that are used in the end-2-end tests."""
import random
import time
from contextlib import contextmanager
import pytest
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.cli.client_utils import BECGuiClient, _start_plot_process
from bec_widgets.utils import BECDispatcher
from bec_widgets.widgets.containers.figure import BECFigure
from bec_widgets.cli.client_utils import BECGuiClient
# pylint: disable=unused-argument
# pylint: disable=redefined-outer-name
# make threads check in autouse, **will be executed at the end**; better than
# having it in fixtures for each test, since it prevents from needing to
# 'manually' shutdown bec_client_lib (for example) to make it happy, whereas
# whereas in fact bec_client_lib makes its on cleanup
@pytest.fixture(autouse=True)
def threads_check_fixture(threads_check):
"""
Fixture to check if threads are still alive at the end of the test.
This should always run to avoid leaked threads within our application.
The fixture is set to autouse, meaning it will run for every test.
"""
return
@pytest.fixture
def gui_id():
return f"figure_{random.randint(0,100)}" # make a new gui id each time, to ensure no 'gui is alive' zombie key can perturbate
@contextmanager
def plot_server(gui_id, klass, client_lib):
dispatcher = BECDispatcher(client=client_lib) # Has to init singleton with fixture client
process, _ = _start_plot_process(
gui_id, klass, gui_class_id="bec", config=client_lib._client._service_config.config_path
)
try:
while client_lib._client.connector.get(MessageEndpoints.gui_heartbeat(gui_id)) is None:
time.sleep(0.3)
yield gui_id
finally:
process.terminate()
process.wait()
dispatcher.disconnect_all()
dispatcher.reset_singleton()
@pytest.fixture
def connected_client_figure(gui_id, bec_client_lib):
with plot_server(gui_id, BECFigure, bec_client_lib) as server:
yield server
"""New gui id each time, to ensure no 'gui is alive' zombie key can perturbate"""
return f"figure_{random.randint(0,100)}"
@pytest.fixture
def connected_client_gui_obj(gui_id, bec_client_lib):
"""
Fixture to create a new BECGuiClient object and start a server in the background.
This fixture should be used if a new gui instance is needed for each test.
"""
gui = BECGuiClient(gui_id=gui_id)
try:
gui.start(wait=True)
# gui._start_server(wait=True)
yield gui
finally:
gui.kill_server()
@pytest.fixture
def connected_client_dock(gui_id, bec_client_lib):
@pytest.fixture(scope="session")
def connected_gui_with_scope_session(gui_id, bec_client_lib):
"""
Fixture to create a new BECGuiClient object and start a server in the background.
This fixture is scoped to the session, meaning it remains alive for all tests in the session.
We can use this fixture to create a gui object that is used across multiple tests, and
simulate a real-world scenario where the gui is not restarted for each test.
"""
gui = BECGuiClient(gui_id=gui_id)
gui._auto_updates_enabled = False
try:
gui.start(wait=True)
gui.window_list[0]
yield gui.window_list[0]
finally:
gui.kill_server()
@pytest.fixture
def connected_client_dock_w_auto_updates(gui_id, bec_client_lib):
gui = BECGuiClient(gui_id=gui_id)
try:
gui._start_server(wait=True)
yield gui, gui.window_list[0]
yield gui
finally:
gui.kill_server()