From 5949251d0e6843237ba8122df1de80ce26f5329d Mon Sep 17 00:00:00 2001 From: Tom David Mueller Date: Tue, 7 Apr 2026 13:31:34 +0200 Subject: [PATCH 1/2] Fix tool instance name handling and move custom defaults to params.json MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolves the issue where multiple instances of the same TOPP tool could not have independent custom defaults. The .ini file (keyed by tool name) only supported one set of defaults — the first instance's defaults were baked in and subsequent instances were silently ignored. Changes: - Add get_merged_params() to ParameterManager for centralized three-layer parameter resolution (ini defaults < _defaults < user overrides) - Replace .ini file mutation with _defaults seeding in params.json, keyed by tool instance name - Update save_parameters() to compare against _defaults when determining which values are non-default - Update run_topp() and non_default_params_summary() to use merged params - Resolve instance names to real tool names across save_parameters, input_TOPP, run_topp, and get_topp_parameters --- .streamlit/config.toml | 2 + src/workflow/CommandExecutor.py | 32 +-- src/workflow/ParameterManager.py | 72 ++++-- src/workflow/StreamlitUI.py | 78 ++++--- tests/test_parameter_defaults.py | 363 +++++++++++++++++++++++++++++++ tests/test_tool_instance_name.py | 268 +++++++++++++++++++++++ 6 files changed, 756 insertions(+), 59 deletions(-) create mode 100644 tests/test_parameter_defaults.py create mode 100644 tests/test_tool_instance_name.py 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..bfea0822 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}"] 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 From e3c3665a7e53814f1c057aa4b57a0bd01f7a21fb Mon Sep 17 00:00:00 2001 From: Tom David Mueller Date: Tue, 7 Apr 2026 13:40:39 +0200 Subject: [PATCH 2/2] Remove redundant -ini flag from run_topp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The .ini file is no longer mutated with custom defaults — it stays pristine. All parameters (_defaults + user overrides) are already passed as CLI flags via get_merged_params(), making -ini redundant. --- src/workflow/CommandExecutor.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/workflow/CommandExecutor.py b/src/workflow/CommandExecutor.py index bfea0822..14d28a32 100644 --- a/src/workflow/CommandExecutor.py +++ b/src/workflow/CommandExecutor.py @@ -310,11 +310,6 @@ def run_topp(self, tool: str, input_output: dict, custom_params: dict = {}, tool 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])