diff --git a/.streamlit/config.toml b/.streamlit/config.toml index 00c6abba..e3d442ef 100644 --- a/.streamlit/config.toml +++ b/.streamlit/config.toml @@ -8,6 +8,8 @@ developmentMode = false address = "0.0.0.0" maxUploadSize = 200 #MB port = 8501 # should be same as configured in deployment repo +enableCORS = false +enableXsrfProtection = false [theme] diff --git a/src/workflow/CommandExecutor.py b/src/workflow/CommandExecutor.py index 86f265b2..14d28a32 100644 --- a/src/workflow/CommandExecutor.py +++ b/src/workflow/CommandExecutor.py @@ -216,7 +216,7 @@ def read_stderr(): stdout_thread.join() stderr_thread.join() - def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> bool: + def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}, tool_instance_name: str = None) -> bool: """ Constructs and executes commands for the specified tool OpenMS TOPP tool based on the given input and output configurations. Ensures that all input/output file lists @@ -234,6 +234,9 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b tool (str): The executable name or path of the tool. input_output (dict): A dictionary specifying the input/output parameter names (as key) and their corresponding file paths (as value). custom_params (dict): A dictionary of custom parameters to pass to the tool. + tool_instance_name (str, optional): A unique instance name for this tool + invocation, used for parameter lookup when multiple instances of the + same tool exist. If not provided, defaults to the tool name. Returns: bool: True if all commands succeeded, False if any failed. @@ -242,6 +245,8 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b ValueError: If the lengths of input/output file lists are inconsistent, except for single string inputs. """ + # Use tool_instance_name for parameter lookup, fall back to tool name + params_key = tool_instance_name if tool_instance_name else tool # check input: any input lists must be same length, other items can be a single string # e.g. input_mzML : [list of n mzML files], output_featureXML : [list of n featureXML files], input_database : database.tsv io_lengths = [len(v) for v in input_output.values() if len(v) > 1] @@ -261,8 +266,8 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b commands = [] - # Load parameters for non-defaults - params = self.parameter_manager.get_parameters_from_json() + # Load merged parameters (_defaults + user overrides) for this tool instance + merged_params = self.parameter_manager.get_merged_params(params_key) # Construct commands for each process for i in range(n_processes): command = [tool] @@ -281,17 +286,16 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b # standard case, files was a list of strings, take the file name at index else: command += [value[i]] - # Add non-default TOPP tool parameters - if tool in params.keys(): - for k, v in params[tool].items(): - command += [f"-{k}"] - # Skip only empty strings (pass flag with no value) - # Note: 0 and 0.0 are valid values, so use explicit check - if v != "" and v is not None: - if isinstance(v, str) and "\n" in v: - command += v.split("\n") - else: - command += [str(v)] + # Add merged TOPP tool parameters (_defaults + user overrides) + for k, v in merged_params.items(): + command += [f"-{k}"] + # Skip only empty strings (pass flag with no value) + # Note: 0 and 0.0 are valid values, so use explicit check + if v != "" and v is not None: + if isinstance(v, str) and "\n" in v: + command += v.split("\n") + else: + command += [str(v)] # Add custom parameters for k, v in custom_params.items(): command += [f"-{k}"] @@ -306,11 +310,6 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}) -> b command += ["-threads", str(threads_per_command)] commands.append(command) - # check if a ini file has been written, if yes use it (contains custom defaults) - ini_path = Path(self.parameter_manager.ini_dir, tool + ".ini") - if ini_path.exists(): - command += ["-ini", str(ini_path)] - # Run command(s) if len(commands) == 1: return self.run_command(commands[0]) diff --git a/src/workflow/ParameterManager.py b/src/workflow/ParameterManager.py index b0c36263..c4eaf3ec 100644 --- a/src/workflow/ParameterManager.py +++ b/src/workflow/ParameterManager.py @@ -65,7 +65,7 @@ def save_parameters(self) -> None: # Advanced parameters are only in session state if the view is active json_params = self.get_parameters_from_json() | json_params - # get a list of TOPP tools which are in session state + # get a list of TOPP tools (or tool instance names) which are in session state current_topp_tools = list( set( [ @@ -75,12 +75,16 @@ def save_parameters(self) -> None: ] ) ) - # for each TOPP tool, open the ini file + # Retrieve the instance-name → real-tool-name mapping (set by input_TOPP) + tool_instance_map = st.session_state.get("_topp_tool_instance_map", {}) + # for each TOPP tool (or instance name), open the ini file for tool in current_topp_tools: - if not self.create_ini(tool): + # Resolve instance name to real tool name for create_ini / ini loading + real_tool = tool_instance_map.get(tool, tool) + if not self.create_ini(real_tool): # Could not create ini file - skip this tool continue - ini_path = Path(self.ini_dir, f"{tool}.ini") + ini_path = Path(self.ini_dir, f"{real_tool}.ini") if tool not in json_params: json_params[tool] = {} # load the param object @@ -92,19 +96,26 @@ def save_parameters(self) -> None: # Skip display keys used by multiselect widgets if key.endswith("_display"): continue - # get ini_key - ini_key = key.replace(self.topp_param_prefix, "").encode() + # get ini_key – map instance name back to real tool name + ini_key = key.replace(self.topp_param_prefix, "") + if tool != real_tool: + ini_key = ini_key.replace(f"{tool}:1:", f"{real_tool}:1:", 1) + ini_key = ini_key.encode() # get ini (default) value by ini_key ini_value = param.getValue(ini_key) is_list_param = isinstance(ini_value, list) - # check if value is different from default OR is an empty list parameter + # Effective default: _defaults value if present, else ini value + short_key = key.split(":1:")[1] + defaults = json_params.get("_defaults", {}).get(tool, {}) + default_value = defaults.get(short_key, ini_value) + # check if value is different from effective default OR is an empty list parameter if ( - (ini_value != value) - or (key.split(":1:")[1] in json_params[tool]) + (default_value != value) + or (short_key in json_params[tool]) or (is_list_param and not value) # Always save empty list params ): # store non-default value - json_params[tool][key.split(":1:")[1]] = value + json_params[tool][short_key] = value # Save to json file with open(self.params_file, "w", encoding="utf-8") as f: json.dump(json_params, f, indent=4) @@ -130,17 +141,44 @@ def get_parameters_from_json(self) -> dict: st.error("**ERROR**: Attempting to load an invalid JSON parameter file. Reset to defaults.") return {} - def get_topp_parameters(self, tool: str) -> dict: + def get_merged_params(self, tool_instance_name: str, ini_params: dict = None) -> dict: + """ + Three-layer parameter merge: ini defaults < _defaults < user overrides. + + Args: + tool_instance_name: Instance name (or tool name) to look up in params.json. + ini_params: Base parameters from the .ini file. Optional — callers that + don't need the ini layer (e.g., run_topp, which passes -ini separately) + can omit this. + + Returns: + Merged dict with the effective value for each parameter. + """ + params = self.get_parameters_from_json() + defaults = params.get("_defaults", {}).get(tool_instance_name, {}) + user = params.get(tool_instance_name, {}) + + merged = {} + if ini_params: + merged.update(ini_params) + merged.update(defaults) + merged.update(user) + return merged + + def get_topp_parameters(self, tool: str, tool_instance_name: str = None) -> dict: """ Get all parameters for a TOPP tool, merging defaults with user values. Args: - tool: Name of the TOPP tool (e.g., "CometAdapter") + tool: Name of the TOPP tool executable (e.g., "CometAdapter") + tool_instance_name: Optional instance name used for parameter storage + (e.g., "IDFilter_step1"). If not provided, defaults to tool name. Returns: Dict with parameter names as keys (without tool prefix) and their values. Returns empty dict if ini file doesn't exist. """ + instance_name = tool_instance_name or tool ini_path = Path(self.ini_dir, f"{tool}.ini") if not ini_path.exists(): return {} @@ -151,18 +189,14 @@ def get_topp_parameters(self, tool: str) -> dict: # Build dict from ini (extract short key names) prefix = f"{tool}:1:" - full_params = {} + ini_params = {} for key in param.keys(): key_str = key.decode() if isinstance(key, bytes) else str(key) if prefix in key_str: short_key = key_str.split(prefix, 1)[1] - full_params[short_key] = param.getValue(key) - - # Override with user-modified values from JSON - user_params = self.get_parameters_from_json().get(tool, {}) - full_params.update(user_params) + ini_params[short_key] = param.getValue(key) - return full_params + return self.get_merged_params(instance_name, ini_params=ini_params) def reset_to_default_parameters(self) -> None: """ diff --git a/src/workflow/StreamlitUI.py b/src/workflow/StreamlitUI.py index 3cca528b..c0880966 100644 --- a/src/workflow/StreamlitUI.py +++ b/src/workflow/StreamlitUI.py @@ -616,6 +616,7 @@ def input_TOPP( display_subsections: bool = True, display_subsection_tabs: bool = False, custom_defaults: dict = {}, + tool_instance_name: str = None, ) -> None: """ Generates input widgets for TOPP tool parameters dynamically based on the tool's @@ -631,29 +632,43 @@ def input_TOPP( display_subsections (bool, optional): Whether to split parameters into subsections based on the prefix. Defaults to True. display_subsection_tabs (bool, optional): Whether to display main subsections in separate tabs (if more than one main section). Defaults to False. custom_defaults (dict, optional): Dictionary of custom defaults to use. Defaults to an empty dict. + tool_instance_name (str, optional): A unique instance name for this tool + invocation. Allows multiple instances of the same TOPP tool with + independent parameters (e.g., two IDFilter calls). If not provided, + defaults to topp_tool_name. The instance name is used for session + state keys and parameter storage, while topp_tool_name is used for + the actual tool executable and ini file creation. """ + # Default instance name to the tool name when not provided + if tool_instance_name is None: + tool_instance_name = topp_tool_name + + # Register instance-name → real-tool-name mapping in session state + if "_topp_tool_instance_map" not in st.session_state: + st.session_state["_topp_tool_instance_map"] = {} + st.session_state["_topp_tool_instance_map"][tool_instance_name] = topp_tool_name if not display_subsections: display_subsection_tabs = False if display_subsection_tabs: display_subsections = True - # write defaults ini files + # Create pristine ini file (never mutated with custom defaults) ini_file_path = Path(self.parameter_manager.ini_dir, f"{topp_tool_name}.ini") - ini_existed = ini_file_path.exists() if not self.parameter_manager.create_ini(topp_tool_name): st.error(f"TOPP tool **'{topp_tool_name}'** not found.") return - if not ini_existed: - # update custom defaults if necessary - if custom_defaults: - param = poms.Param() - poms.ParamXMLFile().load(str(ini_file_path), param) - for key, value in custom_defaults.items(): - encoded_key = f"{topp_tool_name}:1:{key}".encode() - if encoded_key in param.keys(): - param.setValue(encoded_key, value) - poms.ParamXMLFile().store(str(ini_file_path), param) + + # Seed custom defaults into params.json under _defaults key + if custom_defaults: + params = self.parameter_manager.get_parameters_from_json() + if "_defaults" not in params: + params["_defaults"] = {} + params["_defaults"][tool_instance_name] = custom_defaults + with open(self.parameter_manager.params_file, "w", encoding="utf-8") as f: + json.dump(params, f, indent=4) + # Refresh self.params so widget resolution sees _defaults + self.params = self.parameter_manager.get_parameters_from_json() # read into Param object param = poms.Param() @@ -730,18 +745,18 @@ def _matches_parameter(pattern: str, key: bytes) -> bool: ) params.append(p) - # for each parameter in params_decoded - # if a parameter with custom default value exists, use that value - # else check if the parameter is already in self.params, if yes take the value from self.params + # Build ini_params dict for three-layer merge + ini_params = {} + for p in params: + name = p["key"].decode().split(":1:")[1] + ini_params[name] = p["value"] + + # Resolve effective values: ini < _defaults < user overrides + merged = self.parameter_manager.get_merged_params(tool_instance_name, ini_params=ini_params) for p in params: name = p["key"].decode().split(":1:")[1] - if topp_tool_name in self.params: - if name in self.params[topp_tool_name]: - p["value"] = self.params[topp_tool_name][name] - elif name in custom_defaults: - p["value"] = custom_defaults[name] - elif name in custom_defaults: - p["value"] = custom_defaults[name] + if name in merged: + p["value"] = merged[name] # Ensure list parameters stay as lists after loading from JSON # (JSON may store single-item lists as strings) if p["original_is_list"] and isinstance(p["value"], str): @@ -775,7 +790,7 @@ def _matches_parameter(pattern: str, key: bytes) -> bool: # Display tool name if required if display_tool_name: - st.markdown(f"**{topp_tool_name}**") + st.markdown(f"**{tool_instance_name}**") tab_names = [k for k in param_sections.keys() if ":" not in k] tabs = None @@ -803,8 +818,11 @@ def display_TOPP_params(params: dict, num_cols): cols = st.columns(num_cols) i = 0 for p in params: - # get key and name - key = f"{self.parameter_manager.topp_param_prefix}{p['key'].decode()}" + # get key and name – use tool_instance_name in session state key + key_str = p['key'].decode() + if tool_instance_name != topp_tool_name: + key_str = key_str.replace(f"{topp_tool_name}:1:", f"{tool_instance_name}:1:", 1) + key = f"{self.parameter_manager.topp_param_prefix}{key_str}" name = p["name"] try: # sometimes strings with newline, handle as list @@ -1371,7 +1389,8 @@ def remove_full_paths(d: dict) -> dict: general = {} for k, v in params.items(): - # skip if v is a file path + if k == "_defaults": + continue if isinstance(v, dict): topp[k] = v elif ".py" in k: @@ -1382,6 +1401,13 @@ def remove_full_paths(d: dict) -> dict: else: general[k] = v + # Merge _defaults into topp so summary shows custom defaults + user overrides + defaults = params.get("_defaults", {}) + for tool_name, default_vals in defaults.items(): + if tool_name not in topp: + topp[tool_name] = {} + topp[tool_name] = {**default_vals, **topp.get(tool_name, {})} + markdown = [] def dict_to_markdown(d: dict): diff --git a/tests/test_parameter_defaults.py b/tests/test_parameter_defaults.py new file mode 100644 index 00000000..9ccd679e --- /dev/null +++ b/tests/test_parameter_defaults.py @@ -0,0 +1,363 @@ +""" +Tests for get_merged_params() and the refactored get_topp_parameters(). + +This module verifies the three-layer parameter merge: + ini defaults < _defaults < user overrides +""" +import os +import sys +import json +import pytest +import tempfile +from pathlib import Path +from unittest.mock import MagicMock + +# Add project root to path for imports +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.append(PROJECT_ROOT) + +# Mock streamlit before importing ParameterManager so that the imported module +# uses a controllable `st.session_state` (a plain dict) instead of the real one, +# which requires a running Streamlit app context. +mock_streamlit = MagicMock() +mock_streamlit.session_state = {} + +# Temporarily replace streamlit in sys.modules so that ParameterManager's +# `import streamlit as st` picks up the mock. Restore immediately after import +# so other test files (e.g., test_gui.py AppTest) get the real streamlit. +_original_streamlit = sys.modules.get('streamlit') +sys.modules['streamlit'] = mock_streamlit + +from src.workflow.ParameterManager import ParameterManager + +if _original_streamlit is not None: + sys.modules['streamlit'] = _original_streamlit +else: + sys.modules.pop('streamlit', None) + +# Remove cached src.workflow modules that were imported with mocked streamlit so +# that AppTest (in test_gui.py) re-imports them fresh with the real package. +for _key in list(sys.modules.keys()): + if _key.startswith('src.workflow'): + sys.modules.pop(_key, None) + + +@pytest.fixture(autouse=True) +def reset_streamlit_state(): + """Reset mock streamlit session state before each test.""" + mock_streamlit.session_state.clear() + yield + + +@pytest.fixture +def temp_workflow_dir(): + """Create a temporary workflow directory for testing.""" + with tempfile.TemporaryDirectory() as tmpdir: + workflow_dir = Path(tmpdir) / "test-workflow" + workflow_dir.mkdir() + ini_dir = workflow_dir / "ini" + ini_dir.mkdir() + yield workflow_dir + + +class TestGetMergedParams: + """Tests for ParameterManager.get_merged_params().""" + + def test_returns_ini_params_when_no_json(self, temp_workflow_dir): + """ini params returned when params.json doesn't exist.""" + pm = ParameterManager(temp_workflow_dir) + ini_params = {"algorithm:param": 1.0, "algorithm:other": "hello"} + + result = pm.get_merged_params("SomeTool", ini_params=ini_params) + + assert result == {"algorithm:param": 1.0, "algorithm:other": "hello"} + + def test_defaults_override_ini(self, temp_workflow_dir): + """_defaults layer overrides ini values.""" + pm = ParameterManager(temp_workflow_dir) + params_json = { + "_defaults": {"SomeTool": {"algorithm:param": 42.0}} + } + with open(pm.params_file, "w") as f: + json.dump(params_json, f) + + result = pm.get_merged_params("SomeTool", ini_params={"algorithm:param": 1.0}) + + assert result["algorithm:param"] == 42.0 + + def test_user_overrides_defaults(self, temp_workflow_dir): + """User overrides take priority over _defaults.""" + pm = ParameterManager(temp_workflow_dir) + params_json = { + "_defaults": {"SomeTool": {"algorithm:param": 42.0}}, + "SomeTool": {"algorithm:param": 99.0} + } + with open(pm.params_file, "w") as f: + json.dump(params_json, f) + + result = pm.get_merged_params("SomeTool", ini_params={"algorithm:param": 1.0}) + + assert result["algorithm:param"] == 99.0 + + def test_full_three_layer_merge(self, temp_workflow_dir): + """All three layers merge correctly: ini < _defaults < user.""" + pm = ParameterManager(temp_workflow_dir) + params_json = { + "_defaults": { + "SomeTool": { + "algorithm:param_a": 10.0, # overrides ini + "algorithm:param_b": 20.0, # overrides ini, NOT overridden by user + } + }, + "SomeTool": { + "algorithm:param_a": 99.0, # overrides _defaults + "algorithm:param_c": 55.0, # only in user + } + } + with open(pm.params_file, "w") as f: + json.dump(params_json, f) + + ini_params = { + "algorithm:param_a": 1.0, + "algorithm:param_b": 2.0, + "algorithm:param_d": 3.0, # only in ini + } + + result = pm.get_merged_params("SomeTool", ini_params=ini_params) + + assert result["algorithm:param_a"] == 99.0 # user wins + assert result["algorithm:param_b"] == 20.0 # _defaults wins over ini + assert result["algorithm:param_c"] == 55.0 # user-only key present + assert result["algorithm:param_d"] == 3.0 # ini-only key present + + def test_no_ini_params(self, temp_workflow_dir): + """Works when ini_params is None.""" + pm = ParameterManager(temp_workflow_dir) + params_json = { + "_defaults": {"SomeTool": {"algorithm:param": 42.0}}, + "SomeTool": {"algorithm:other": 7.0} + } + with open(pm.params_file, "w") as f: + json.dump(params_json, f) + + result = pm.get_merged_params("SomeTool") + + assert result["algorithm:param"] == 42.0 + assert result["algorithm:other"] == 7.0 + + def test_different_instances_same_tool(self, temp_workflow_dir): + """Different instance names get independent _defaults and user overrides.""" + pm = ParameterManager(temp_workflow_dir) + params_json = { + "_defaults": { + "IDFilter_step1": {"score:min": 0.05}, + "IDFilter_step2": {"score:min": 0.01}, + }, + "IDFilter_step1": {"score:min": 0.001}, + } + with open(pm.params_file, "w") as f: + json.dump(params_json, f) + + result1 = pm.get_merged_params("IDFilter_step1", ini_params={"score:min": 0.5}) + result2 = pm.get_merged_params("IDFilter_step2", ini_params={"score:min": 0.5}) + + assert result1["score:min"] == 0.001 # user override for step1 + assert result2["score:min"] == 0.01 # _defaults for step2 (no user override) + + def test_empty_params_json(self, temp_workflow_dir): + """Returns empty dict when params.json is empty and no ini_params.""" + pm = ParameterManager(temp_workflow_dir) + with open(pm.params_file, "w") as f: + json.dump({}, f) + + result = pm.get_merged_params("SomeTool") + + assert result == {} + + +class TestGetToppParametersWithDefaults: + + def test_get_topp_parameters_includes_defaults(self, temp_workflow_dir): + """get_topp_parameters merges _defaults between ini and user values.""" + pm = ParameterManager(temp_workflow_dir) + params_json = { + "_defaults": {"SomeTool": {"algorithm:param": 42.0}}, + "SomeTool": {"algorithm:other": 99.0} + } + with open(pm.params_file, "w") as f: + json.dump(params_json, f) + + result = pm.get_merged_params("SomeTool", ini_params={"algorithm:param": 1.0}) + assert result["algorithm:param"] == 42.0 + assert result["algorithm:other"] == 99.0 + + +class TestDefaultsSeeding: + + def test_seed_writes_defaults_to_params_json(self, temp_workflow_dir): + """Seeding creates _defaults entry in params.json.""" + pm = ParameterManager(temp_workflow_dir) + custom_defaults = {"param_a": 10.0, "param_b": "fast"} + + # Simulate what input_TOPP seeding does + params = pm.get_parameters_from_json() + if "_defaults" not in params: + params["_defaults"] = {} + params["_defaults"]["MyTool"] = custom_defaults + with open(pm.params_file, "w") as f: + json.dump(params, f) + + # Verify + loaded = pm.get_parameters_from_json() + assert loaded["_defaults"]["MyTool"] == {"param_a": 10.0, "param_b": "fast"} + + def test_seed_is_idempotent(self, temp_workflow_dir): + """Seeding the same tool twice overwrites cleanly.""" + pm = ParameterManager(temp_workflow_dir) + + # First seed + params = {"_defaults": {"Tool": {"p1": 1.0}}, "other_key": "keep"} + with open(pm.params_file, "w") as f: + json.dump(params, f) + + # Second seed with updated defaults + params = pm.get_parameters_from_json() + params["_defaults"]["Tool"] = {"p1": 2.0} + with open(pm.params_file, "w") as f: + json.dump(params, f) + + loaded = pm.get_parameters_from_json() + assert loaded["_defaults"]["Tool"]["p1"] == 2.0 + assert loaded["other_key"] == "keep" + + def test_seed_multiple_instances(self, temp_workflow_dir): + """Different instances of the same tool get independent _defaults.""" + pm = ParameterManager(temp_workflow_dir) + params = { + "_defaults": { + "IDFilter_strict": {"score:pep": 0.01}, + "IDFilter_lenient": {"score:pep": 0.05}, + } + } + with open(pm.params_file, "w") as f: + json.dump(params, f) + + loaded = pm.get_parameters_from_json() + assert loaded["_defaults"]["IDFilter_strict"]["score:pep"] == 0.01 + assert loaded["_defaults"]["IDFilter_lenient"]["score:pep"] == 0.05 + + +try: + import pyopenms as poms + HAS_PYOPENMS = True +except ImportError: + HAS_PYOPENMS = False + + +@pytest.mark.skipif(not HAS_PYOPENMS, reason="pyopenms not available") +class TestSaveParametersWithDefaults: + + def _create_fake_ini(self, pm, tool_name, params_dict): + """Create a fake .ini file with given parameters.""" + param = poms.Param() + for key, value in params_dict.items(): + param.setValue(f"{tool_name}:1:{key}".encode(), value) + poms.ParamXMLFile().store(str(Path(pm.ini_dir, f"{tool_name}.ini")), param) + + def test_value_matching_custom_default_not_saved(self, temp_workflow_dir): + """A value equal to the _defaults entry should not be saved as a user override.""" + pm = ParameterManager(temp_workflow_dir) + + # Create a fake ini with a default value + self._create_fake_ini(pm, "Tool", {"param_a": 10.0}) + + # Pre-seed _defaults with a different value than ini + params = {"_defaults": {"Tool": {"param_a": 42.0}}} + with open(pm.params_file, "w") as f: + json.dump(params, f) + + # Session state has value matching the custom default (42.0), not the ini default (10.0) + mock_streamlit.session_state[f"{pm.topp_param_prefix}Tool:1:param_a"] = 42.0 + mock_streamlit.session_state["_topp_tool_instance_map"] = {"Tool": "Tool"} + + pm.save_parameters() + + with open(pm.params_file, "r") as f: + saved = json.load(f) + + # param_a should NOT appear under Tool (it matches the _defaults value) + assert "param_a" not in saved.get("Tool", {}) + # _defaults should still be present + assert saved["_defaults"]["Tool"]["param_a"] == 42.0 + + def test_value_different_from_custom_default_saved(self, temp_workflow_dir): + """A value different from _defaults entry should be saved as user override.""" + pm = ParameterManager(temp_workflow_dir) + + # Create a fake ini with a default value + self._create_fake_ini(pm, "Tool", {"param_a": 10.0}) + + params = {"_defaults": {"Tool": {"param_a": 42.0}}} + with open(pm.params_file, "w") as f: + json.dump(params, f) + + mock_streamlit.session_state[f"{pm.topp_param_prefix}Tool:1:param_a"] = 99.0 + mock_streamlit.session_state["_topp_tool_instance_map"] = {"Tool": "Tool"} + + pm.save_parameters() + + with open(pm.params_file, "r") as f: + saved = json.load(f) + + assert saved["Tool"]["param_a"] == 99.0 + + +class TestNonDefaultParamsSummaryDefaults: + + def test_defaults_key_excluded_from_classification(self): + """_defaults dict should not appear as a TOPP tool in the summary.""" + params = { + "_defaults": {"Tool": {"p1": 10}}, + "Tool": {"p1": 20}, + "general_param": "value" + } + # Simulate the classification logic + topp = {} + general = {} + for k, v in params.items(): + if k == "_defaults": + continue + if isinstance(v, dict): + topp[k] = v + else: + general[k] = v + + assert "_defaults" not in topp + assert "Tool" in topp + assert "general_param" in general + + def test_defaults_merged_into_summary(self): + """_defaults values should appear in summary merged with user overrides.""" + params = { + "_defaults": { + "ToolA": {"p1": 10, "p2": 20}, + "ToolB": {"p3": 30} + }, + "ToolA": {"p1": 99} + } + # Simulate the merge logic for summary + topp = {} + for k, v in params.items(): + if k == "_defaults": + continue + if isinstance(v, dict): + topp[k] = v + + defaults = params.get("_defaults", {}) + for tool_name, default_vals in defaults.items(): + if tool_name not in topp: + topp[tool_name] = {} + topp[tool_name] = {**default_vals, **topp.get(tool_name, {})} + + assert topp["ToolA"] == {"p1": 99, "p2": 20} # user override wins for p1 + assert topp["ToolB"] == {"p3": 30} # defaults-only tool appears diff --git a/tests/test_tool_instance_name.py b/tests/test_tool_instance_name.py new file mode 100644 index 00000000..cd060cac --- /dev/null +++ b/tests/test_tool_instance_name.py @@ -0,0 +1,268 @@ +""" +Tests for the tool_instance_name functionality. + +This module verifies that save_parameters correctly resolves tool instance names +to real tool names when calling create_ini, and that parameters are stored and +retrieved using the instance name as the key. +""" +import os +import sys +import json +import pytest +import tempfile +from pathlib import Path +from unittest.mock import patch, MagicMock, call + +# Add project root to path for imports +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.append(PROJECT_ROOT) + +# Mock streamlit before importing ParameterManager +mock_streamlit = MagicMock() +mock_streamlit.session_state = {} + +_original_streamlit = sys.modules.get('streamlit') +sys.modules['streamlit'] = mock_streamlit + +from src.workflow.ParameterManager import ParameterManager + +if _original_streamlit is not None: + sys.modules['streamlit'] = _original_streamlit +else: + sys.modules.pop('streamlit', None) + +# Remove cached src.workflow modules +for _key in list(sys.modules.keys()): + if _key.startswith('src.workflow'): + sys.modules.pop(_key, None) + + +@pytest.fixture +def temp_workflow_dir(): + """Create a temporary workflow directory for testing.""" + with tempfile.TemporaryDirectory() as tmpdir: + workflow_dir = Path(tmpdir) / "test-workflow" + workflow_dir.mkdir() + ini_dir = workflow_dir / "ini" + ini_dir.mkdir() + yield workflow_dir + + +@pytest.fixture(autouse=True) +def reset_streamlit_state(): + """Reset mock streamlit session state before each test.""" + mock_streamlit.session_state.clear() + yield + + +class TestSaveParametersWithInstanceName: + """Tests for save_parameters correctly resolving tool instance names.""" + + def test_save_parameters_uses_real_tool_name_for_create_ini(self, temp_workflow_dir): + """Test that save_parameters resolves instance name to real tool name + before calling create_ini.""" + pm = ParameterManager(temp_workflow_dir) + + # Simulate session state with an instance name (IDFilter_step1) + # that differs from the real tool name (IDFilter) + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter_step1:1:score:pep"] = 0.05 + # Register instance mapping (as input_TOPP would) + mock_streamlit.session_state["_topp_tool_instance_map"] = { + "IDFilter_step1": "IDFilter" + } + + # Mock create_ini to track calls - return False (tool not found) + # to prevent further processing that requires actual ini files + with patch.object(pm, 'create_ini', return_value=False) as mock_create_ini: + pm.save_parameters() + + # Verify create_ini was called with the REAL tool name, not the instance name + mock_create_ini.assert_called_once_with("IDFilter") + + def test_save_parameters_without_instance_map_uses_tool_name_directly(self, temp_workflow_dir): + """Test that save_parameters works normally when no instance map exists + (backward compatibility).""" + pm = ParameterManager(temp_workflow_dir) + + # Simulate session state with a normal tool name (no instance mapping) + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter:1:score:pep"] = 0.05 + + with patch.object(pm, 'create_ini', return_value=False) as mock_create_ini: + pm.save_parameters() + + # Should use the tool name directly + mock_create_ini.assert_called_once_with("IDFilter") + + def test_save_parameters_stores_under_instance_name(self, temp_workflow_dir): + """Test that parameters are stored in JSON under the instance name, not + the real tool name.""" + pm = ParameterManager(temp_workflow_dir) + + # Create a mock ini file for IDFilter + ini_path = temp_workflow_dir / "ini" / "IDFilter.ini" + ini_path.touch() + + # Set up instance mapping + mock_streamlit.session_state["_topp_tool_instance_map"] = { + "IDFilter_step1": "IDFilter" + } + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter_step1:1:score:pep"] = 0.05 + + # Mock pyopenms Param and ParamXMLFile to avoid needing real ini files + mock_param = MagicMock() + mock_param.getValue.return_value = 0.01 # Different from session state value + + with patch.object(pm, 'create_ini', return_value=True), \ + patch('pyopenms.Param', return_value=mock_param), \ + patch('pyopenms.ParamXMLFile') as mock_xml: + pm.save_parameters() + + # Load saved parameters + with open(pm.params_file, "r") as f: + saved = json.load(f) + + # Parameters should be stored under the instance name + assert "IDFilter_step1" in saved + assert saved["IDFilter_step1"]["score:pep"] == 0.05 + + def test_save_parameters_multiple_instances_same_tool(self, temp_workflow_dir): + """Test that two instances of the same tool get separate parameter entries.""" + pm = ParameterManager(temp_workflow_dir) + + ini_path = temp_workflow_dir / "ini" / "IDFilter.ini" + ini_path.touch() + + # Set up two instances with different parameter values + mock_streamlit.session_state["_topp_tool_instance_map"] = { + "IDFilter_step1": "IDFilter", + "IDFilter_step2": "IDFilter", + } + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter_step1:1:score:pep"] = 0.01 + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter_step2:1:score:pep"] = 0.05 + + mock_param = MagicMock() + mock_param.getValue.return_value = 0.0 # Default differs from both + + with patch.object(pm, 'create_ini', return_value=True), \ + patch('pyopenms.Param', return_value=mock_param), \ + patch('pyopenms.ParamXMLFile'): + pm.save_parameters() + + with open(pm.params_file, "r") as f: + saved = json.load(f) + + # Both instances should have separate entries + assert "IDFilter_step1" in saved + assert "IDFilter_step2" in saved + assert saved["IDFilter_step1"]["score:pep"] == 0.01 + assert saved["IDFilter_step2"]["score:pep"] == 0.05 + + def test_save_parameters_ini_key_maps_instance_to_real_tool(self, temp_workflow_dir): + """Test that ini_key correctly maps instance name back to real tool name + for param.getValue lookup.""" + pm = ParameterManager(temp_workflow_dir) + + ini_path = temp_workflow_dir / "ini" / "IDFilter.ini" + ini_path.touch() + + mock_streamlit.session_state["_topp_tool_instance_map"] = { + "IDFilter_step1": "IDFilter" + } + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter_step1:1:score:pep"] = 0.05 + + mock_param = MagicMock() + mock_param.getValue.return_value = 0.01 + + with patch.object(pm, 'create_ini', return_value=True), \ + patch('pyopenms.Param', return_value=mock_param), \ + patch('pyopenms.ParamXMLFile'): + pm.save_parameters() + + # Verify that param.getValue was called with the REAL tool name key + # (IDFilter:1:score:pep), not the instance name key (IDFilter_step1:1:score:pep) + mock_param.getValue.assert_called_with(b"IDFilter:1:score:pep") + + def test_save_parameters_display_keys_skipped_with_instance_name(self, temp_workflow_dir): + """Test that _display keys are still skipped when using instance names.""" + pm = ParameterManager(temp_workflow_dir) + + ini_path = temp_workflow_dir / "ini" / "IDFilter.ini" + ini_path.touch() + + mock_streamlit.session_state["_topp_tool_instance_map"] = { + "IDFilter_step1": "IDFilter" + } + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter_step1:1:score:pep"] = 0.05 + mock_streamlit.session_state[f"{pm.topp_param_prefix}IDFilter_step1:1:score:pep_display"] = ["0.05"] + + mock_param = MagicMock() + mock_param.getValue.return_value = 0.01 + + with patch.object(pm, 'create_ini', return_value=True), \ + patch('pyopenms.Param', return_value=mock_param), \ + patch('pyopenms.ParamXMLFile'): + pm.save_parameters() + + with open(pm.params_file, "r") as f: + saved = json.load(f) + + # _display key should not be stored + assert "score:pep_display" not in saved.get("IDFilter_step1", {}) + assert "score:pep" in saved.get("IDFilter_step1", {}) + + +class TestGetToppParametersWithInstanceName: + """Tests for get_topp_parameters with tool_instance_name.""" + + def test_get_topp_parameters_with_instance_name(self, temp_workflow_dir): + """Test that get_topp_parameters uses instance name for JSON lookup.""" + pm = ParameterManager(temp_workflow_dir) + + # Create params.json with instance-keyed parameters + params = { + "IDFilter_step1": { + "score:pep": 0.05 + } + } + with open(pm.params_file, "w") as f: + json.dump(params, f) + + # Create a mock ini file + ini_path = temp_workflow_dir / "ini" / "IDFilter.ini" + ini_path.touch() + + mock_param = MagicMock() + mock_param.keys.return_value = [b"IDFilter:1:score:pep"] + mock_param.getValue.return_value = 0.01 # default + + with patch('pyopenms.Param', return_value=mock_param), \ + patch('pyopenms.ParamXMLFile'): + result = pm.get_topp_parameters("IDFilter", tool_instance_name="IDFilter_step1") + + # Should return the instance-specific value + assert result["score:pep"] == 0.05 + + def test_get_topp_parameters_without_instance_name_backward_compat(self, temp_workflow_dir): + """Test that get_topp_parameters works without instance name (backward compat).""" + pm = ParameterManager(temp_workflow_dir) + + params = { + "IDFilter": { + "score:pep": 0.05 + } + } + with open(pm.params_file, "w") as f: + json.dump(params, f) + + ini_path = temp_workflow_dir / "ini" / "IDFilter.ini" + ini_path.touch() + + mock_param = MagicMock() + mock_param.keys.return_value = [b"IDFilter:1:score:pep"] + mock_param.getValue.return_value = 0.01 + + with patch('pyopenms.Param', return_value=mock_param), \ + patch('pyopenms.ParamXMLFile'): + result = pm.get_topp_parameters("IDFilter") + + assert result["score:pep"] == 0.05