bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-04 16:02:51 +01:00
Code Activity

refactor: address review comments

Browse Source
This commit is contained in:
appel_c
2026-02-27 22:34:04 +01:00
parent 2cad86158f
commit b6101c391f
4 changed files with 45 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,7 +31,7 @@ from bec_widgets.utils.toolbars.actions import (
from bec_widgets.utils.toolbars.bundles import ToolbarBundle
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
from bec_widgets.widgets.services.bec_atlas_admin_view.bec_atlas_http_service import (
ATLAS_ENPOINTS,
ATLAS_ENDPOINTS,
AuthenticatedUserInfo,
BECAtlasHTTPService,
HTTPResponse,
@@ -471,12 +471,11 @@ class BECAtlasAdminView(BECWidget, QWidget):
logger.debug(
f"HTTP Response received: {response.request_url} with status {response.status}"
)
if ATLAS_ENPOINTS.REALMS_EXPERIMENTS in response.request_url:
if ATLAS_ENDPOINTS.REALMS_EXPERIMENTS in response.request_url:
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)
elif ATLAS_ENPOINTS.SET_EXPERIMENT in response.request_url:
self._on_overview_selected() # Reconsider this as the overview is now the login.
elif ATLAS_ENDPOINTS.SET_EXPERIMENT in response.request_url:
self._on_overview_selected()
@SafeSlot(dict)
def _on_authenticated(self, auth_info: dict) -> None:
@@ -522,7 +521,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
## API Methods
################
@SafeSlot(dict, popup_error=True)
@SafeSlot(str, str, popup_error=True)
def set_experiment(self, experiment_id: str, deployment_id: str) -> None:
"""Set the experiment information for the current experiment."""
self.atlas_http_service.set_experiment(experiment_id, deployment_id)

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

@@ -1,6 +1,15 @@
"""
This module is a QWidget-based HTTP service, responsible for interacting with the
BEC Atlas API using a QNetworkAccessManager. It prov"""
HTTP service widget for interacting with the BEC Atlas API.
This module defines Qt-based classes that wrap ``QNetworkAccessManager`` to perform
authenticated HTTP requests against the BEC Atlas backend, manage login and logout
flows, and track authentication token expiry for the BEC Atlas Admin View.
It also provides Pydantic models that describe HTTP responses and authenticated user
information, as well as an ``AuthenticatedTimer`` helper used to signal when
authentication tokens expire.
"""
import json
import time
@@ -20,7 +29,7 @@ from bec_widgets.utils.error_popups import SafeSlot
logger = bec_logger.logger
class ATLAS_ENPOINTS(StrEnum):
class ATLAS_ENDPOINTS(StrEnum):
"""Constants for BEC Atlas API endpoints."""
LOGIN = "/user/login"
@@ -188,12 +197,12 @@ class BECAtlasHTTPService(QWidget):
data = {}
logger.warning(f"Received empty response for {request_url} with status code {status}.")
if not isinstance(data, dict):
if not isinstance(data, (dict, list, str)):
raise BECAtlasHTTPError(
f"Expected response data to be a dict for {request_url}, but got {type(data)}. Response content: {data}"
f"Expected response data to be a dict, list, or str for {request_url}, but got {type(data)}. Response content: {data}"
)
if ATLAS_ENPOINTS.LOGIN.value in request_url:
if ATLAS_ENDPOINTS.LOGIN.value in request_url:
# If it's a login response, don't forward the token
# but extract the expiration time and emit it
token = data.get("access_token")
@@ -201,13 +210,12 @@ class BECAtlasHTTPService(QWidget):
self.authentication_expires.emit(data.get("exp", 0))
# Now we set the auth info, and then fetch the user info to get the groups
self.__set_auth_info(data)
# Fetch information about the deployment info
self.get_user_info() # Fetch groups, then emit authenticated once groups are set on auth_user
elif ATLAS_ENPOINTS.LOGOUT.value in request_url:
elif ATLAS_ENDPOINTS.LOGOUT.value in request_url:
self._auth_timer.stop() # Stop the timer if it was running
self.__clear_login_info(skip_logout=True) # Skip calling logout again
self.authenticated.emit({})
elif ATLAS_ENPOINTS.USER_INFO.value in request_url:
elif ATLAS_ENDPOINTS.USER_INFO.value in request_url:
groups = data.get("groups", [])
email = data.get("email", "")
# Second step of authentication: We also have all groups now
@@ -221,7 +229,7 @@ class BECAtlasHTTPService(QWidget):
self.get_deployment_info(
deployment_id=self._current_deployment_info.deployment_id
)
elif ATLAS_ENPOINTS.DEPLOYMENT_INFO.value in request_url:
elif ATLAS_ENDPOINTS.DEPLOYMENT_INFO.value in request_url:
owner_groups = data.get("owner_groups", [])
if self.auth_user_info is not None and not self.auth_user_info.groups.isdisjoint(
owner_groups
@@ -256,7 +264,6 @@ class BECAtlasHTTPService(QWidget):
endpoint (str): The API endpoint to send the GET request to.
query_parameters (dict | None): Optional query parameters to include in the URL.
"""
logger.info(f"Sending GET request to {endpoint}.")
url = QUrl(self._base_url + endpoint)
if query_parameters:
query = QUrlQuery()
@@ -279,7 +286,6 @@ class BECAtlasHTTPService(QWidget):
payload (dict): The JSON payload to include in the POST request.
query_parameters (dict | None): Optional query parameters to include in the URL.
"""
logger.info(f"Sending GET request to {endpoint}.")
if payload is None:
payload = {}
url = QUrl(self._base_url + endpoint)
@@ -323,14 +329,14 @@ class BECAtlasHTTPService(QWidget):
password (str): The password for authentication.
"""
self._post_request(
endpoint=ATLAS_ENPOINTS.LOGIN.value,
endpoint=ATLAS_ENDPOINTS.LOGIN.value,
payload={"username": username, "password": password},
)
@SafeSlot(popup_error=True)
def logout(self):
"""Logout from BEC Atlas."""
self._post_request(endpoint=ATLAS_ENPOINTS.LOGOUT.value)
self._post_request(endpoint=ATLAS_ENDPOINTS.LOGOUT.value)
@SafeSlot(str, popup_error=True)
def get_experiments_for_realm(self, realm_id: str):
@@ -341,7 +347,7 @@ class BECAtlasHTTPService(QWidget):
realm_id (str): The ID of the realm to retrieve experiments for.
"""
self._get_request(
endpoint=ATLAS_ENPOINTS.REALMS_EXPERIMENTS.value,
endpoint=ATLAS_ENDPOINTS.REALMS_EXPERIMENTS.value,
query_parameters={"realm_id": realm_id},
)
@@ -355,14 +361,14 @@ class BECAtlasHTTPService(QWidget):
deployment_id (str): The ID of the deployment associated with the experiment.
"""
self._post_request(
endpoint=ATLAS_ENPOINTS.SET_EXPERIMENT.value,
endpoint=ATLAS_ENDPOINTS.SET_EXPERIMENT.value,
query_parameters={"experiment_id": experiment_id, "deployment_id": deployment_id},
)
@SafeSlot(popup_error=True)
def get_user_info(self):
"""Get the current user information from BEC Atlas. Requires authentication."""
self._get_request(endpoint=ATLAS_ENPOINTS.USER_INFO.value)
self._get_request(endpoint=ATLAS_ENDPOINTS.USER_INFO.value)
@SafeSlot(str, popup_error=True)
def get_deployment_info(self, deployment_id: str):
@@ -373,6 +379,6 @@ class BECAtlasHTTPService(QWidget):
deployment_id (str): The ID of the deployment to retrieve information for.
"""
self._get_request(
endpoint=ATLAS_ENPOINTS.DEPLOYMENT_INFO.value,
endpoint=ATLAS_ENDPOINTS.DEPLOYMENT_INFO.value,
query_parameters={"deployment_id": deployment_id},
)

6
bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/experiment_selection.py
View File

@@ -266,9 +266,8 @@ class ExperimentSelection(QWidget):
self._side_card.set_experiment_info(self._table_infos[index.row()])
def _emit_selected_experiment(self):
if self._tabs.currentWidget() is self._card_tab:
if self._tabs.currentWidget() is self._card_tab and self._next_experiment:
self.experiment_selected.emit(self._next_experiment)
logger.info(f"Emitting next experiment signal with info: {self._next_experiment}")
return
selected = self._table.selectionModel().selectedRows()
if not selected:
@@ -297,7 +296,8 @@ class ExperimentSelection(QWidget):
def _on_tab_changed(self, index):
if self._tabs.widget(index) is self._table_tab:
self._table.resizeRowsToContents()
self._side_card.set_experiment_info(self._next_experiment)
if self._next_experiment:
self._side_card.set_experiment_info(self._next_experiment)
self._apply_table_filters()
def _get_column_data(self, row) -> dict[str, str]:

24
tests/unit_tests/test_atlas_admin_view.py
View File

@@ -17,7 +17,7 @@ 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,
ATLAS_ENDPOINTS,
AuthenticatedUserInfo,
BECAtlasHTTPError,
BECAtlasHTTPService,
@@ -102,7 +102,7 @@ class TestBECAtlasHTTPService:
with mock.patch.object(http_service.network_manager, "get") as mock_get:
http_service._get_request(
endpoint=ATLAS_ENPOINTS.REALMS_EXPERIMENTS.value,
endpoint=ATLAS_ENDPOINTS.REALMS_EXPERIMENTS.value,
query_parameters={"realm_id": "realm-1"},
)
@@ -118,7 +118,8 @@ class TestBECAtlasHTTPService:
with mock.patch.object(http_service.network_manager, "post") as mock_post:
http_service._post_request(
endpoint=ATLAS_ENPOINTS.LOGIN.value, payload={"username": "alice", "password": "pw"}
endpoint=ATLAS_ENDPOINTS.LOGIN.value,
payload={"username": "alice", "password": "pw"},
)
mock_post.assert_called_once()
@@ -132,13 +133,13 @@ class TestBECAtlasHTTPService:
with mock.patch.object(http_service, "_get_request") as mock_get:
# User info
http_service.get_user_info()
mock_get.assert_called_once_with(endpoint=ATLAS_ENPOINTS.USER_INFO.value)
mock_get.assert_called_once_with(endpoint=ATLAS_ENDPOINTS.USER_INFO.value)
mock_get.reset_mock()
# Deployment info
http_service.get_deployment_info("dep-1")
mock_get.assert_called_once_with(
endpoint=ATLAS_ENPOINTS.DEPLOYMENT_INFO.value,
endpoint=ATLAS_ENDPOINTS.DEPLOYMENT_INFO.value,
query_parameters={"deployment_id": "dep-1"},
)
@@ -146,27 +147,28 @@ class TestBECAtlasHTTPService:
# Realms experiments
http_service.get_experiments_for_realm("realm-1")
mock_get.assert_called_once_with(
endpoint=ATLAS_ENPOINTS.REALMS_EXPERIMENTS.value,
endpoint=ATLAS_ENDPOINTS.REALMS_EXPERIMENTS.value,
query_parameters={"realm_id": "realm-1"},
)
with mock.patch.object(http_service, "_post_request") as mock_post:
# Logout
http_service.logout()
mock_post.assert_called_once_with(endpoint=ATLAS_ENPOINTS.LOGOUT.value)
mock_post.assert_called_once_with(endpoint=ATLAS_ENDPOINTS.LOGOUT.value)
mock_post.reset_mock()
# Login
http_service.login("alice", "pw")
mock_post.assert_called_once_with(
endpoint=ATLAS_ENPOINTS.LOGIN.value, payload={"username": "alice", "password": "pw"}
endpoint=ATLAS_ENDPOINTS.LOGIN.value,
payload={"username": "alice", "password": "pw"},
)
mock_post.reset_mock()
# Set experiment
http_service.set_experiment("exp-1", "dep-1")
mock_post.assert_called_once_with(
endpoint=ATLAS_ENPOINTS.SET_EXPERIMENT.value,
endpoint=ATLAS_ENDPOINTS.SET_EXPERIMENT.value,
query_parameters={"experiment_id": "exp-1", "deployment_id": "dep-1"},
)
@@ -698,7 +700,7 @@ class TestBECAtlasAdminView:
):
"""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",
request_url=f"{admin_view._atlas_url}/{ATLAS_ENDPOINTS.REALMS_EXPERIMENTS}/experiments?realm_id=TestBeamline",
status=200,
headers={"content-type": "application/json"},
data=experiment_info_list,
@@ -710,7 +712,7 @@ class TestBECAtlasAdminView:
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}",
request_url=f"{admin_view._atlas_url}/{ATLAS_ENDPOINTS.SET_EXPERIMENT}",
status=200,
headers={"content-type": "application/json"},
data={},