diff --git a/environment.yml b/environment.yml index aa8fa17..c107c39 100644 --- a/environment.yml +++ b/environment.yml @@ -3,6 +3,7 @@ channels: - conda-forge - mantid - oncat + - neutrons dependencies: - anaconda-client - boa @@ -21,5 +22,6 @@ dependencies: - sphinx-rtd-theme - python-build - pyoncat + - pyoncatqt - gtk3 - libstdcxx-ng diff --git a/src/shiver/views/oncat.py b/src/shiver/views/oncat.py index 21bd7ae..d1969cc 100644 --- a/src/shiver/views/oncat.py +++ b/src/shiver/views/oncat.py @@ -1,220 +1,20 @@ """PyQt widget for the OnCat widget in General tab.""" import os -import json -import oauthlib import numpy as np import pyoncat +from pyoncatqt.login import ONCatLogin from qtpy.QtWidgets import ( QGridLayout, QGroupBox, - QPushButton, QLabel, QComboBox, - QDialog, - QLineEdit, QDoubleSpinBox, ) from qtpy.QtCore import QTimer from shiver.configuration import get_data -class OncatToken: - """Class to hold OnCat token""" - - def __init__(self, token_path: str): - self.token_path = token_path - - def read_token(self): - """Read token from file""" - # If there is not a token stored, return - # None from the callback. - if not os.path.exists(self.token_path): - return None - - with open(self.token_path, encoding="UTF-8") as storage: - return json.load(storage) - - def write_token(self, token): - """Write token to file""" - # check if directory exists - if not os.path.exists(os.path.dirname(self.token_path)): - os.makedirs(os.path.dirname(self.token_path)) - # write token to file - with open(self.token_path, "w", encoding="UTF-8") as storage: - json.dump(token, storage) - # change permissions to read-only by user - os.chmod(self.token_path, 0o600) - - -class OnCatAgent: - """Agent to interface with OnCat""" - - def __init__(self, use_notes=False) -> None: - """Initialize OnCat agent""" - # get configuration settings - self._use_notes = use_notes - self._oncat_url = get_data("generate_tab.oncat", "oncat_url") - self._client_id = get_data("generate_tab.oncat", "client_id") - - user_home_dir = os.path.expanduser("~") - self._token = OncatToken( - os.path.abspath(f"{user_home_dir}/.shiver/oncat_token.json"), - ) - self._agent = pyoncat.ONCat( - self._oncat_url, - client_id=self._client_id, - # Pass in token getter/setter callbacks here: - token_getter=self._token.read_token, - token_setter=self._token.write_token, - flow=pyoncat.RESOURCE_OWNER_CREDENTIALS_FLOW, - ) - - def login(self, username: str, password: str): - """Connect to OnCat with given username and password. - - Parameters - ---------- - username : str - Username for OnCat - password : str - Password for OnCat - """ - self._agent.login(username, password) - - @property - def has_token(self): - """Check if token exists""" - return self._token.read_token() is not None - - @property - def is_connected(self): - """Check if connected to OnCat""" - try: - self._agent.Facility.list() # pylint: disable=no-member - return True - except pyoncat.InvalidRefreshTokenError: - return False - except pyoncat.LoginRequiredError: - return False - except Exception: # pylint: disable=W0718 - return False - - def get_ipts(self, facility: str, instrument: str) -> list: - """Get IPTS numbers for given facility and instrument. - - Parameters - ---------- - facility : str - Facility name - instrument : str - Instrument name - - Returns - ------- - list - List of IPTS numbers in string format (IPTS-XXXXX) - """ - # return empty list if not connected - if not self.is_connected: - return [] - # get list of experiments - experiments = self._agent.Experiment.list( # pylint: disable=no-member - facility=facility, - instrument=instrument, - projection=["facility", "instrument"], - ) - return list({exp["id"] for exp in experiments}) - - def get_datasets(self, facility: str, instrument: str, ipts: int) -> list: - """Get datasets for given facility, instrument, and IPTS. - - Parameters - ---------- - facility : str - Facility name - instrument : str - Instrument name - ipts : int - IPTS number - - Returns - ------- - list - List of datasets - """ - # return empty list if not connected - if not self.is_connected: - return [] - # get list of datasets - return get_dataset_names( - self._agent, - ipts_number=ipts, - instrument=instrument, - use_notes=self._use_notes, - facility=facility, - ) - - def get_agent_instance(self): - """Get OnCat agent instance""" - return self._agent - - -class OnCatLogin(QDialog): - """OnCat login dialog""" - - def __init__(self, parent=None, error_message_callback=None): - super().__init__(parent) - self.setWindowTitle("Use U/XCAM to connect to OnCat") - self.resize(350, 200) - layout = QGridLayout() - label1 = QLabel("UserId") - self.user_obj = QLineEdit() - layout.addWidget(label1, 0, 0) - layout.addWidget(self.user_obj, 0, 1) - label2 = QLabel("Password") - self.user_pwd = QLineEdit() - layout.addWidget(label2, 1, 0) - layout.addWidget(self.user_pwd, 1, 1) - self.user_pwd.setEchoMode(QLineEdit.Password) - self.button_login = QPushButton("Login") - layout.addWidget(self.button_login, 2, 0, 2, 2) - self.setLayout(layout) - - # connect signals and slots - self.button_login.clicked.connect(self.accept) - - self.error_message_callback = error_message_callback - - def accept(self): - """Accept""" - # login to OnCat - try: - self.parent().oncat_agent.login( - self.user_obj.text(), - self.user_pwd.text(), - ) - except oauthlib.oauth2.rfc6749.errors.InvalidGrantError: - if self.error_message_callback is not None: - self.error_message_callback( - "Invalid username or password. Please try again.", - ) - self.user_pwd.setText("") - return - except pyoncat.LoginRequiredError: - if self.error_message_callback is not None: - self.error_message_callback( - "A username and/or password was not provided when logging in.", - ) - self.user_pwd.setText("") - return - - # ask parent to sync - self.parent().sync_with_remote() - # close dialog - self.close() - - class Oncat(QGroupBox): """ONCat widget""" @@ -258,29 +58,19 @@ def __init__(self, parent=None): self.oncat_options_layout.addWidget(self.angle_target_label, 3, 0) self.oncat_options_layout.addWidget(self.angle_target, 3, 1) - # status indicator (disconnected: red, connected: green) - self.status_label = QLabel("") - self.status_label.setToolTip("ONCat connection status.") - self.oncat_options_layout.addWidget(self.status_label, 4, 0) - - # connect to OnCat button - self.oncat_button = QPushButton("&Connect to ONCat") - self.oncat_button.setFixedWidth(300) - self.oncat_button.setToolTip("Connect to ONCat (requires login credentials).") - self.oncat_options_layout.addWidget(self.oncat_button, 4, 1) + self.oncat_login = ONCatLogin(key="shiver", parent=self) + self.oncat_login.connection_updated.connect(self.connect_to_oncat) + self.oncat_options_layout.addWidget(self.oncat_login, 4, 0, 1, 2) # set layout self.setLayout(self.oncat_options_layout) - # connect signals and slots - self.oncat_button.clicked.connect(self.connect_to_oncat) - # error message callback self.error_message_callback = None self.use_notes = get_data("generate_tab.oncat", "use_notes") # OnCat agent - self.oncat_agent = OnCatAgent(self.use_notes) + self.oncat_agent = self.oncat_login.get_agent_instance() # Sync with remote self.sync_with_remote(refresh=True) @@ -307,7 +97,7 @@ def __init__(self, parent=None): @property def connected_to_oncat(self) -> bool: """Check if connected to OnCat""" - return self.oncat_agent.is_connected + return self.oncat_login.is_connected def get_suggested_path(self) -> str: """Return a suggested path based on current selection.""" @@ -327,7 +117,7 @@ def get_suggested_selected_files(self) -> list: group_by_angle = self.angle_target.value() > 0 return get_dataset_info( - login=self.oncat_agent.get_agent_instance(), + login=self.oncat_agent, ipts_number=self.get_ipts_number(), instrument=self.get_instrument(), use_notes=self.use_notes, @@ -343,39 +133,15 @@ def set_dataset_to_custom(self): def connect_to_oncat(self): """Connect to OnCat""" - # check if connection exists - if self.connected_to_oncat: - return - - # check if token exists - if self.oncat_agent.has_token: - if self.error_message_callback: - self.error_message_callback("Session expired. Please login again.") - - # prompt for username and password - oncat_login = OnCatLogin(self, self.error_message_callback) - oncat_login.show() - # update connection status self.sync_with_remote(refresh=True) def sync_with_remote(self, refresh=False): """Update all items within OnCat widget.""" - self.update_connection_status() - if self.connected_to_oncat and refresh: self.update_ipts() self.update_datasets() - def update_connection_status(self): - """Update connection status""" - if self.oncat_agent.is_connected: - self.status_label.setText("ONCat: Connected") - self.status_label.setStyleSheet("color: green") - else: - self.status_label.setText("ONCat: Disconnected") - self.status_label.setStyleSheet("color: red") - def connect_error_callback(self, callback): """Connect error message callback""" self.error_message_callback = callback @@ -433,7 +199,7 @@ def get_facility(self): def update_ipts(self): """Update IPTS list""" # get IPTS list from OnCat - ipts_list = self.oncat_agent.get_ipts( + ipts_list = self.get_oncat_ipts( self.get_facility(), self.get_instrument(), ) @@ -444,15 +210,45 @@ def update_ipts(self): def update_datasets(self): """Update dataset list""" # get dataset list from OnCat - dataset_list = ["custom"] + self.oncat_agent.get_datasets( - self.get_facility(), - self.get_instrument(), - self.get_ipts_number(), - ) + dataset_list = [] + if self.connected_to_oncat: + dataset_list = ["custom"] + get_dataset_names( + self.oncat_agent, + facility=self.get_facility(), + instrument=self.get_instrument(), + ipts_number=self.get_ipts_number(), + use_notes=self.use_notes, + ) # update dataset list self.dataset.clear() self.dataset.addItems(sorted(dataset_list)) + def get_oncat_ipts(self, facility: str, instrument: str) -> list: + """Get IPTS numbers for given facility and instrument. + + Parameters + ---------- + facility : str + Facility name + instrument : str + Instrument name + + Returns + ------- + list + List of IPTS numbers in string format (IPTS-XXXXX) + """ + # return empty list if not connected + if not self.connected_to_oncat: + return [] + # get list of experiments + experiments = self.oncat_agent.Experiment.list( # pylint: disable=no-member + facility=facility, + instrument=instrument, + projection=["facility", "instrument"], + ) + return list({exp["id"] for exp in experiments}) + # The following are utility functions migrated from the original # oncat_util.py diff --git a/tests/conftest.py b/tests/conftest.py index 1f111ca..db71dce 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -33,3 +33,8 @@ def user_conf_file(tmp_path_factory, request): with open(user_path, "w", encoding="utf8") as config_file: user_config.write(config_file) return user_path + + +@pytest.fixture(autouse=True) +def _get_login(monkeypatch: pytest.fixture) -> None: + monkeypatch.setattr(os, "getlogin", lambda: "test") diff --git a/tests/views/test_oncat.py b/tests/views/test_oncat.py index 8d818c0..5dbe985 100644 --- a/tests/views/test_oncat.py +++ b/tests/views/test_oncat.py @@ -1,17 +1,10 @@ #!/usr/bin/env python # pylint: disable=all """Test the views for the ONCat application.""" -import os -from pathlib import Path -from configparser import ConfigParser - -import pytest -import pyoncat from qtpy.QtWidgets import QGroupBox +from qtpy.QtCore import Signal +import pytest from shiver.views.oncat import ( - OncatToken, - OnCatAgent, - OnCatLogin, Oncat, get_data_from_oncat, get_dataset_names, @@ -19,218 +12,34 @@ ) -def get_configuration_settings(): - """get configuration settings from the configuration_template file""" - template_config = ConfigParser() - project_directory = Path(__file__).resolve().parent.parent.parent - template_file_path = os.path.join(project_directory, "src", "shiver", "configuration_template.ini") - template_config.read(template_file_path) - return template_config - - -class MockRecord: - def __init__(self, *args, **kwargs) -> None: - pass - - def list(self, *args, **kwargs) -> list: - return [{"id": "test_id"}, {"id": "test_id2"}] - - -class MockONcat: - """Mock Oncat instance for testing""" - - def __init__(self, *args, **kwargs) -> None: - self.Facility = MockRecord() - self.Experiment = MockRecord() - - def login(self, *args, **kwargs) -> None: - """Mock login""" - pass - - -def test_oncat_token(tmp_path): - """Test the OncatToken class.""" - token_path = str(tmp_path / "shiver_test") - - token = OncatToken(token_path) - - # test write - token.write_token("test_token") - - # test read - assert token.read_token() == "test_token" - - -def test_oncat_agent(monkeypatch): - """Test the OnCatAgent class.""" - - # mockey patch get_dataset_names - def mock_get_dataset_names(*args, **kwargs): - return ["test1", "test2"] - - monkeypatch.setattr("shiver.views.oncat.get_dataset_names", mock_get_dataset_names) - - # mockey patch class pyoncat.ONCat - monkeypatch.setattr("pyoncat.ONCat", MockONcat) - - # mockey patch get_settings data - def mock_get_data(*args, **kwargs): - # get configuration setting from the configuration_template file - template_config = get_configuration_settings() - return template_config[args[1], args[2]] - - monkeypatch.setattr("shiver.configuration.get_data", mock_get_data) - - # test the class - agent = OnCatAgent() - # test configuration settings are stored from template configuration file - assert agent._oncat_url == "https://oncat.ornl.gov" - assert agent._client_id == "99025bb3-ce06-4f4b-bcf2-36ebf925cd1d" - assert agent._use_notes is False - # test login - agent.login("test_login", "test_password") - # test is_connected - assert agent.is_connected is True - # test get_ipts - assert set(agent.get_ipts("facility", "inst")) == {"test_id2", "test_id"} - # test get_dataset_names - assert set(agent.get_datasets("facility", "inst", 1)) == {"test1", "test2"} - # test get agent instance - assert agent.get_agent_instance() is not None - - -def test_oncat_login(qtbot): - """Test the login window""" - - # fake oncat agent - class FakeOnCatAgent: - def __init__(self) -> None: - pass - - def login(self, *args, **kwargs) -> None: - pass - - class DummyOnCatAgent: - def __init__(self) -> None: - template_config = get_configuration_settings() - oncat_url = template_config["generate_tab.oncat"]["oncat_url"] - client_id = template_config["generate_tab.oncat"]["client_id"] - self._agent = pyoncat.ONCat( - oncat_url, - client_id=client_id, - flow=pyoncat.RESOURCE_OWNER_CREDENTIALS_FLOW, - ) - - def login(self, username: str, password: str) -> None: - self._agent.login(username, password) - - # fake parent widget - class FakeParent(QGroupBox): - def __init__(self, parent=None) -> None: - super().__init__(parent) - self.oncat_agent = FakeOnCatAgent() - - def sync_with_remote(self, *args, **kwargs): - pass - - class DummyParent(QGroupBox): - def __init__(self, parent=None) -> None: - super().__init__(parent) - self.oncat_agent = DummyOnCatAgent() - - def sync_with_remote(self, *args, **kwargs): - pass - - err_msgs = [] - - def error_message_callback(msg): - err_msgs.append(msg) - - # Use Fake ones to test happy path - # case: login successfully (based on fake oncat agent) - fake_parent = FakeParent() - qtbot.addWidget(fake_parent) - oncat_login_wgt = OnCatLogin(fake_parent, error_message_callback) - qtbot.addWidget(oncat_login_wgt) - oncat_login_wgt.show() - # - oncat_login_wgt.user_obj.setText("test_login") - oncat_login_wgt.user_pwd.setText("test_password") - oncat_login_wgt.button_login.click() - assert len(err_msgs) == 0 - - # Use Dummy ones to test unhappy path - dummy_parent = DummyParent() - qtbot.addWidget(dummy_parent) - # case: login without password - oncat_login_wgt1 = OnCatLogin(dummy_parent, error_message_callback) - qtbot.addWidget(oncat_login_wgt1) - # - oncat_login_wgt1.user_obj.setText("test_login") - oncat_login_wgt1.user_pwd.setText("") - oncat_login_wgt1.button_login.click() - assert err_msgs[-1] == "A username and/or password was not provided when logging in." - # - oncat_login_wgt2 = OnCatLogin(dummy_parent, error_message_callback) - qtbot.addWidget(oncat_login_wgt2) - oncat_login_wgt2.show() - # case: login with wrong password - oncat_login_wgt2.user_obj.setText("test_login") - oncat_login_wgt2.user_pwd.setText("wrong_password") - oncat_login_wgt2.button_login.click() - assert err_msgs[-1] == "Invalid username or password. Please try again." - - -@pytest.mark.parametrize( - "user_conf_file", - [ - """ - [generate_tab.oncat] - oncat_url = test_url - client_id = 0000-0000 - use_notes = False - """ - ], - indirect=True, -) -def test_oncat(monkeypatch, user_conf_file, qtbot): +def test_oncat(monkeypatch, qtbot): """Test the Oncat class.""" - # mockpatch OnCatAgent - class MockOnCatAgent: - def __init__(self, use_notes) -> None: - pass - - def login(self, *args, **kwargs) -> None: - pass + class MockLogin(QGroupBox): + connection_updated = Signal(bool) - @property - def is_connected(self) -> bool: - return True - - def get_ipts(self, *args, **kwargs) -> list: - return ["IPTS-1111", "IPTS-2222"] - - def get_datasets(self, *args, **kwargs) -> list: - return ["test_dataset1", "test_dataset2"] + def __init__(self, *args, parent, **kwargs): + super().__init__(parent=parent) + self.is_connected = True def get_agent_instance(self): - return "test_agent" + return None - @property - def has_token(self): - return True + def mock_dataset(login, ipts_number, instrument, use_notes, facility): + return ["test_dataset1", "test_dataset2"] - monkeypatch.setattr("shiver.views.oncat.OnCatAgent", MockOnCatAgent) + def mock_ipts(self, facility, instrument): + return ["IPTS-1111", "IPTS-2222"] + + monkeypatch.setattr("shiver.views.oncat.get_dataset_names", mock_dataset) + monkeypatch.setattr(Oncat, "get_oncat_ipts", mock_ipts) # mock get_dataset_info def mock_get_dataset_info(*args, **kwargs): return ["test_file1"] monkeypatch.setattr("shiver.views.oncat.get_dataset_info", mock_get_dataset_info) - - # mock get_oncat_url, client_id and use_notes info - monkeypatch.setattr("shiver.configuration.CONFIG_PATH_FILE", user_conf_file) + monkeypatch.setattr("shiver.views.oncat.ONCatLogin", MockLogin) err_msgs = []