bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-09 18:27:52 +01:00
Code Activity

test(bec-atlas-admin-view): complement tests for BECAtlasAdminView, ExperimentSelection, BECAtlasHTTPService

Browse Source
This commit is contained in:
appel_c
2026-02-27 19:59:32 +01:00
parent 2a8f59738e
commit 22d8530744
3 changed files with 285 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_admin_view.py
View File

@@ -175,7 +175,7 @@ class CustomLogoutAction(MaterialIconAction):
toolbar(QToolBar): The toolbar to add the action to.
target(QWidget): The target widget for the action.
"""
create_action_with_text(toolbar_action=self, toolbar=toolbar, min_size=QSize(100, 40))
create_action_with_text(toolbar_action=self, toolbar=toolbar, min_size=QSize(70, 40))
def set_authenticated(self, auth_info: AuthenticatedUserInfo | None):
"""Enable or disable the logout action based on authentication state."""
@@ -202,7 +202,7 @@ class CustomLogoutAction(MaterialIconAction):
def update_label(self):
"""Update the label text of the logout action."""
if self._login_remaining_s > 0:
label_text = f"{self.label_text} ({self._login_remaining_s}s)"
label_text = f"{self.label_text}\n({self._login_remaining_s}s)"
else:
label_text = self.label_text
self.action.setText(label_text)
@@ -220,7 +220,7 @@ class AtlasConnectionInfo(QWidget):
layout = QVBoxLayout(self)
layout.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
layout.setContentsMargins(6, 6, 6, 6)
layout.setContentsMargins(6, 6, 6, 12)
layout.setSpacing(8)
self._bl_info_label = QLabel(self)
self._atlas_url_label = QLabel(self)
@@ -251,8 +251,8 @@ class BECAtlasAdminView(BECWidget, QWidget):
authenticated = Signal(bool)
def __init__(
self, parent=None, atlas_url: str = "http://localhost/api/v1", client=None
): # https://bec-atlas-dev.psi.ch/api/v1
self, parent=None, atlas_url: str = "https://bec-atlas-dev.psi.ch/api/v1", client=None
):
super().__init__(parent=parent, client=client)
@@ -473,11 +473,8 @@ class BECAtlasAdminView(BECWidget, QWidget):
experiments = response.data if isinstance(response.data, list) else []
# Filter experiments to only include those that the user has write access to
self.experiment_selection.set_experiment_infos(experiments)
# self._on_experiment_selection_selected() # Stick to overview
elif ATLAS_ENPOINTS.SET_EXPERIMENT in response.request_url:
self._on_overview_selected() # Reconsider this as the overview is now the login.
# Reconsider once queue is ready
# elif ATLAS_ENPOINTS.REALMS_EXPERIMENTS in response.request_url:
@SafeSlot(dict)
def _on_authenticated(self, auth_info: dict) -> None:
@@ -498,7 +495,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
f"Authenticated user {info.email} does not have access to the current deployment {self._current_deployment_info.name if self._current_deployment_info else '<no deployment>'}."
)
self._authenticated = authenticated
self.authenticated.emit(authenticated)
if authenticated:
self.toolbar.components.get_action("experiment_selection").action.setEnabled(True)
self.toolbar.components.get_action("messaging_services").action.setEnabled(
@@ -517,6 +514,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
self._on_overview_selected() # Switch back to overview on logout
self._atlas_info_widget.clear_login() # Clear login status in atlas info widget on logout
self.toolbar.components.get_action("logout").set_authenticated(None)
self.authenticated.emit(authenticated)
################
## API Methods

26
bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_http_service.py
View File

@@ -184,6 +184,15 @@ class BECAtlasHTTPService(QWidget):
else:
data = {}
if data is None:
data = {}
logger.warning(f"Received empty response for {request_url} with status code {status}.")
if not isinstance(data, dict):
raise BECAtlasHTTPError(
f"Expected response data to be a dict for {request_url}, but got {type(data)}. Response content: {data}"
)
if ATLAS_ENPOINTS.LOGIN.value in request_url:
# If it's a login response, don't forward the token
# but extract the expiration time and emit it
@@ -214,14 +223,15 @@ class BECAtlasHTTPService(QWidget):
)
elif ATLAS_ENPOINTS.DEPLOYMENT_INFO.value in request_url:
owner_groups = data.get("owner_groups", [])
if self.auth_user_info is not None:
if not self.auth_user_info.groups.isdisjoint(owner_groups):
self.authenticated.emit(self.auth_user_info.model_dump())
else:
self._show_warning(
text=f"User {self.auth_user_info.email} does not have access to the active deployment {data.get('name', '<unknown>')}."
)
self.logout() # Logout to clear auth info and stop timer since user does not have access
if self.auth_user_info is not None and not self.auth_user_info.groups.isdisjoint(
owner_groups
):
self.authenticated.emit(self.auth_user_info.model_dump())
else:
self._show_warning(
text=f"User {self.auth_user_info.email} does not have access to the active deployment {data.get('name', '<unknown>')}."
)
self.logout() # Logout to clear auth info and stop timer since user does not have access
response = HTTPResponse(request_url=request_url, headers=headers, status=status, data=data)
self.http_response.emit(response.model_dump())

323
tests/unit_tests/test_atlas_admin_view.py
View File

@@ -5,15 +5,23 @@ from unittest import mock
import jwt
import pytest
from bec_lib.messages import ExperimentInfoMessage
from bec_lib.messages import (
DeploymentInfoMessage,
ExperimentInfoMessage,
MessagingConfig,
MessagingServiceScopeConfig,
SessionInfoMessage,
)
from qtpy.QtCore import QByteArray, QUrl
from qtpy.QtNetwork import QNetworkRequest
from bec_widgets.widgets.services.bec_atlas_admin_view.bec_atlas_admin_view import BECAtlasAdminView
from bec_widgets.widgets.services.bec_atlas_admin_view.bec_atlas_http_service import (
ATLAS_ENPOINTS,
AuthenticatedUserInfo,
BECAtlasHTTPError,
BECAtlasHTTPService,
HTTPResponse,
)
from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.experiment_mat_card import (
ExperimentMatCard,
@@ -289,36 +297,71 @@ class TestBECAtlasHTTPService:
http_service._handle_response(reply, _override_slot_params={"raise_error": True})
class TestBECAtlasExperimentSelection:
@pytest.fixture
def experiment_info_message() -> ExperimentInfoMessage:
data = {
"_id": "p22622",
"owner_groups": ["admin"],
"access_groups": ["unx-sls_xda_bs", "p22622"],
"realm_id": "TestBeamline",
"proposal": "12345967",
"title": "Test Experiment for Mat Card Widget",
"firstname": "John",
"lastname": "Doe",
"email": "john.doe@psi.ch",
"account": "doe_j",
"pi_firstname": "Jane",
"pi_lastname": "Smith",
"pi_email": "jane.smith@psi.ch",
"pi_account": "smith_j",
"eaccount": "e22622",
"pgroup": "p22622",
"abstract": "This is a test abstract for the experiment mat card widget. It should be long enough to test text wrapping and display in the card. The abstract provides a brief overview of the experiment, its goals, and its significance. This text is meant to simulate a real abstract that might be associated with an experiment in the BEC Atlas system. The card should be able to handle abstracts of varying lengths without any issues, ensuring that the user can read the full abstract even if it is quite long.",
"schedule": [{"start": "01/01/2025 08:00:00", "end": "03/01/2025 18:00:00"}],
"proposal_submitted": "15/12/2024",
"proposal_expire": "31/12/2025",
"proposal_status": "Scheduled",
"delta_last_schedule": 30,
"mainproposal": "",
}
return ExperimentInfoMessage.model_validate(data)
@pytest.fixture
def experiment_info_message(self):
data = {
"_id": "p22622",
"owner_groups": ["admin"],
"access_groups": ["unx-sls_xda_bs", "p22622"],
"realm_id": "TestBeamline",
"proposal": "12345967",
"title": "Test Experiment for Mat Card Widget",
"firstname": "John",
"lastname": "Doe",
"email": "john.doe@psi.ch",
"account": "doe_j",
"pi_firstname": "Jane",
"pi_lastname": "Smith",
"pi_email": "jane.smith@psi.ch",
"pi_account": "smith_j",
"eaccount": "e22622",
"pgroup": "p22622",
"abstract": "This is a test abstract for the experiment mat card widget. It should be long enough to test text wrapping and display in the card. The abstract provides a brief overview of the experiment, its goals, and its significance. This text is meant to simulate a real abstract that might be associated with an experiment in the BEC Atlas system. The card should be able to handle abstracts of varying lengths without any issues, ensuring that the user can read the full abstract even if it is quite long.",
"schedule": [{"start": "01/01/2025 08:00:00", "end": "03/01/2025 18:00:00"}],
"proposal_submitted": "15/12/2024",
"proposal_expire": "31/12/2025",
"proposal_status": "Scheduled",
"delta_last_schedule": 30,
"mainproposal": "",
}
return ExperimentInfoMessage.model_validate(data)
@pytest.fixture
def experiment_info_list(experiment_info_message: ExperimentInfoMessage) -> list[dict]:
"""Fixture to provide a list of experiment info dictionaries."""
another_experiment_info = {
"_id": "p22623",
"owner_groups": ["admin"],
"access_groups": ["unx-sls_xda_bs", "p22623"],
"realm_id": "TestBeamline",
"proposal": "",
"title": "Experiment without Proposal",
"firstname": "Alice",
"lastname": "Johnson",
"email": "alice.johnson@psi.ch",
"account": "johnson_a",
"pi_firstname": "Bob",
"pi_lastname": "Brown",
"pi_email": "bob.brown@psi.ch",
"pi_account": "brown_b",
"eaccount": "e22623",
"pgroup": "p22623",
"abstract": "",
"schedule": [],
"proposal_submitted": "",
"proposal_expire": "",
"proposal_status": "",
"delta_last_schedule": None,
"mainproposal": "",
}
return [
experiment_info_message.model_dump(),
ExperimentInfoMessage.model_validate(another_experiment_info).model_dump(),
]
class TestBECAtlasExperimentSelection:
def test_format_name(self, experiment_info_message: ExperimentInfoMessage):
"""Test utils format name"""
@@ -404,39 +447,6 @@ class TestBECAtlasExperimentSelection:
qtbot.waitExposed(selection)
return selection
@pytest.fixture
def experiment_info_list(self, experiment_info_message: ExperimentInfoMessage):
"""Fixture to provide a list of experiment info dictionaries."""
another_experiment_info = {
"_id": "p22623",
"owner_groups": ["admin"],
"access_groups": ["unx-sls_xda_bs", "p22623"],
"realm_id": "TestBeamline",
"proposal": "",
"title": "Experiment without Proposal",
"firstname": "Alice",
"lastname": "Johnson",
"email": "alice.johnson@psi.ch",
"account": "johnson_a",
"pi_firstname": "Bob",
"pi_lastname": "Brown",
"pi_email": "bob.brown@psi.ch",
"pi_account": "brown_b",
"eaccount": "e22623",
"pgroup": "p22623",
"abstract": "",
"schedule": [],
"proposal_submitted": "",
"proposal_expire": "",
"proposal_status": "",
"delta_last_schedule": None,
"mainproposal": "",
}
return [
experiment_info_message.model_dump(),
ExperimentInfoMessage.model_validate(another_experiment_info).model_dump(),
]
def test_set_experiments(
self, experiment_selection: ExperimentSelection, experiment_info_list: list[dict]
):
@@ -518,3 +528,190 @@ class TestBECAtlasExperimentSelection:
with qtbot.waitSignal(wid.experiment_selected, timeout=1000) as blocker:
wid._side_card._activate_button.click()
assert blocker.args == [exp]
class TestBECAtlasAdminView:
@pytest.fixture
def admin_view(self, qtbot):
"""Fixture to create a BECAtlasAdminView instance."""
view = BECAtlasAdminView()
qtbot.addWidget(view)
qtbot.waitExposed(view)
return view
def test_init_and_login(self, admin_view: BECAtlasAdminView, qtbot):
"""Test that the BECAtlasAdminView initializes correctly."""
# Check that the atlas URL is set correctly
assert admin_view._atlas_url == "https://bec-atlas-dev.psi.ch/api/v1"
# Test that clicking the login button emits the credentials_entered signal with the correct username and password
with mock.patch.object(admin_view.atlas_http_service, "login") as mock_login:
with qtbot.waitSignal(admin_view.overview_widget.login_requested, timeout=1000):
admin_view.overview_widget._login.username.setText("alice")
admin_view.overview_widget._login.password.setText("password")
admin_view.overview_widget._login._emit_credentials()
mock_login.assert_called_once_with(username="alice", password="password")
mock_login.reset_mock()
admin_view._authenticated = True
with mock.patch.object(admin_view, "logout") as mock_logout:
with qtbot.waitSignal(admin_view.overview_widget.login_requested, timeout=1000):
admin_view.overview_widget._login.password.setText("password")
admin_view.overview_widget._login._emit_credentials()
mock_logout.assert_called_once()
mock_login.assert_called_once_with(username="alice", password="password")
def test_on_experiment_selected(
self, admin_view: BECAtlasAdminView, deployment_info: DeploymentInfoMessage, qtbot
):
"""Test that selecting an experiment in the overview widget correctly calls the HTTP service to set the experiment and updates the current experiment view."""
# First we need to simulate that we are authenticated and have deployment info
admin_view._update_deployment_info(deployment_info.model_dump(), {})
with mock.patch.object(
admin_view.atlas_http_service, "set_experiment"
) as mock_set_experiment:
with qtbot.waitSignal(
admin_view.experiment_selection.experiment_selected, timeout=1000
):
admin_view.experiment_selection.experiment_selected.emit(
deployment_info.active_session.experiment.model_dump()
)
mock_set_experiment.assert_called_once_with(
deployment_info.active_session.experiment.pgroup, deployment_info.deployment_id
)
@pytest.fixture
def deployment_info(
self, experiment_info_message: ExperimentInfoMessage
) -> DeploymentInfoMessage:
"""Fixture to provide a DeploymentInfoMessage instance."""
return DeploymentInfoMessage(
deployment_id="dep-1",
name="Test Deployment",
messaging_config=MessagingConfig(
signal=MessagingServiceScopeConfig(enabled=False),
teams=MessagingServiceScopeConfig(enabled=False),
scilog=MessagingServiceScopeConfig(enabled=False),
),
active_session=SessionInfoMessage(
experiment=experiment_info_message, name="Test Session"
),
)
def test_on_authenticated(
self, admin_view: BECAtlasAdminView, deployment_info: DeploymentInfoMessage, qtbot
):
"""Test that the on_authenticated method correctly updates the UI based on authentication state."""
# Simulate successful authentication
auth_info = AuthenticatedUserInfo(
email="alice@example.com",
exp=time.time() + 300,
groups={"operators"},
deployment_id="dep-1",
)
# First check that deployment info updates all fields correctly
admin_view._update_deployment_info(deployment_info.model_dump(), {})
assert admin_view.atlas_http_service._current_deployment_info == deployment_info
assert (
admin_view._atlas_info_widget._bl_info_label.text()
== f"{deployment_info.active_session.experiment.realm_id} @ {deployment_info.name}"
)
assert admin_view._atlas_info_widget._atlas_url_label.text() == admin_view._atlas_url
# Now run on_authenticated, this enables all toolbar buttons
# and calls fetch experiments. It also switches the overview widget
# to the current experiment view.
# Default should be on the overview widget
assert admin_view.stacked_layout.currentWidget() == admin_view.overview_widget
with mock.patch.object(
admin_view, "_fetch_available_experiments"
) as mock_fetch_experiments:
with qtbot.waitSignal(admin_view.authenticated, timeout=1000) as blocker:
admin_view._on_authenticated(auth_info.model_dump())
# Fetch experiments should be called
mock_fetch_experiments.assert_called_once()
assert blocker.args[0] is True
assert (
admin_view._atlas_info_widget._atlas_url_label.text()
== f"{admin_view._atlas_info_widget._atlas_url_text} | {auth_info.email}"
)
assert (
admin_view.toolbar.components.get_action("messaging_services").action.isEnabled()
is False
)
# Logout timer is running
logout_action = admin_view.toolbar.components.get_action("logout")
assert logout_action.action.isEnabled() is True
assert logout_action._tick_timer.isActive() is True
# Current Experiment widget should be visible in the overview
assert (
admin_view.overview_widget.stacked_layout.currentWidget()
== admin_view.overview_widget._experiment_overview_widget
)
# Click toolbar to switch to experiment selection
exp_select = admin_view.toolbar.components.get_action("experiment_selection")
assert exp_select.action.isEnabled() is True
with qtbot.waitSignal(exp_select.action.triggered, timeout=1000):
exp_select.action.trigger()
assert admin_view.stacked_layout.currentWidget() == admin_view.experiment_selection
# Now we simulate that the authentication expires
# This deactivates buttons, resets the overview widget
# and emits authenticated signal with False
with qtbot.waitSignal(admin_view.authenticated, timeout=1000) as blocker:
admin_view._on_authenticated({}) # Simulate not authenticated anymore
assert blocker.args[0] is False
assert logout_action._tick_timer.isActive() is False
assert admin_view._atlas_info_widget._atlas_url_label.text() == admin_view._atlas_url
assert (
admin_view.overview_widget.stacked_layout.currentWidget()
== admin_view.overview_widget._login_widget
)
# View should switch back to overview
assert admin_view.stacked_layout.currentWidget() == admin_view.overview_widget
def test_fetch_experiments(
self, admin_view: BECAtlasAdminView, deployment_info: DeploymentInfoMessage, qtbot
):
"""Test that _fetch_available_experiments correctly calls the HTTP service and updates the experiment selection widget."""
admin_view._update_deployment_info(deployment_info.model_dump(), {})
with mock.patch.object(
admin_view.atlas_http_service, "get_experiments_for_realm"
) as mock_get_experiments:
admin_view._fetch_available_experiments()
mock_get_experiments.assert_called_once_with(
deployment_info.active_session.experiment.realm_id
)
def test_on_http_response_received(
self, experiment_info_list: list[dict], admin_view: BECAtlasAdminView, qtbot
):
"""Test that _on_http_response_received correctly handles HTTP responses and updates the UI accordingly."""
realms = HTTPResponse(
request_url=f"{admin_view._atlas_url}/{ATLAS_ENPOINTS.REALMS_EXPERIMENTS}/experiments?realm_id=TestBeamline",
status=200,
headers={"content-type": "application/json"},
data=experiment_info_list,
)
with mock.patch.object(
admin_view.experiment_selection, "set_experiment_infos"
) as mock_set_experiment_infos:
admin_view._on_http_response_received(realms.model_dump())
mock_set_experiment_infos.assert_called_once_with(experiment_info_list)
set_experiment = HTTPResponse(
request_url=f"{admin_view._atlas_url}/{ATLAS_ENPOINTS.SET_EXPERIMENT}",
status=200,
headers={"content-type": "application/json"},
data={},
)
with mock.patch.object(admin_view, "_on_overview_selected") as mock_on_overview_selected:
admin_view._on_http_response_received(set_experiment.model_dump())
mock_on_overview_selected.assert_called_once()