Simplify gxformat2 contract - drop ImporterGalaxyInterface, convert_and_import_workflow

Inline format2 conversion logic directly in upload_yaml_workflow,
remove ImporterGalaxyInterface from populator class hierarchies,
remove Format2ConverterGalaxyInterface (pass None), remove import_tool
method and embedded GalaxyTool test support, deduplicate selenium
upload_yaml_workflow override into BaseWorkflowPopulator.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

lib/galaxy/managers/workflows.py

@@ -16,7 +16,6 @@ from typing import (
 import yaml
 from gxformat2 import (
     from_galaxy_native,
-    ImporterGalaxyInterface,
     ImportOptions,
     python_to_workflow,
 )
@@ -618,7 +617,7 @@ class WorkflowContentsManager(UsesAnnotations):
         workflow_class, as_dict, object_id = artifact_class(trans, as_dict, allow_in_directory=allow_in_directory)
         assert workflow_class == "GalaxyWorkflow"
         # Format 2 Galaxy workflow.
-        galaxy_interface = Format2ConverterGalaxyInterface()
+        galaxy_interface = None
         import_options = ImportOptions()
         import_options.deduplicate_subworkflows = True
         as_dict = python_to_workflow(as_dict, galaxy_interface, workflow_directory=None, import_options=import_options)
@@ -648,7 +647,7 @@ class WorkflowContentsManager(UsesAnnotations):
         workflow_class, as_dict, object_id = artifact_class(trans, as_dict)
         if workflow_class == "GalaxyWorkflow" or "yaml_content" in as_dict:
             # Format 2 Galaxy workflow.
-            galaxy_interface = Format2ConverterGalaxyInterface()
+            galaxy_interface = None
             import_options = ImportOptions()
             import_options.deduplicate_subworkflows = True
             try:
@@ -2470,13 +2469,6 @@ class RawWorkflowDescription:
         self.workflow_path = workflow_path
 
 
-class Format2ConverterGalaxyInterface(ImporterGalaxyInterface):
-    def import_workflow(self, workflow, **kwds):
-        raise NotImplementedError(
-            "Direct format 2 import of nested workflows is not yet implemented, use bioblend client."
-        )
-
-
 def _get_stored_workflow(session, workflow_uuid, workflow_id, by_stored_id):
     stmt = select(StoredWorkflow)
     if workflow_uuid is not None:

lib/galaxy_test/api/embed_test_1.gxwf.yml

@@ -1,20 +0,0 @@
-class: GalaxyWorkflow
-inputs:
-  - label: input1
-
-steps:
-  - tool_id: cat1
-    label: first_cat
-    state:
-      input1:
-        $link: input1
-  - label: embed1
-    run: 
-      "@import": "embed_test_1_tool.gxtool.yml"
-  - tool_id: cat1
-    state:
-      input1:
-        $link: first_cat/out_file1
-      queries:
-        - input2:
-            $link: embed1/output1

lib/galaxy_test/api/embed_test_1_tool.gxtool.yml

@@ -1,8 +0,0 @@
-class: GalaxyTool
-command: echo 'hello world 2' > $output1
-name: embed_test_1_tool
-version: "0.1"
-outputs:
-  output1:
-    format: txt
-    type: data

lib/galaxy_test/api/test_workflows.py

@@ -1399,44 +1399,6 @@ steps:
             self._assert_status_code_is(other_import_response, 200)
             self._assert_user_has_workflow_with_name("imported: test_import_published_deprecated")
 
-    def test_import_export_dynamic(self):
-        workflow_id = self._upload_yaml_workflow("""
-class: GalaxyWorkflow
-steps:
-  - type: input
-    label: input1
-  - tool_id: cat1
-    label: first_cat
-    state:
-      input1:
-        $link: 0
-  - label: embed1
-    run:
-      class: GalaxyTool
-      version: "0.1"
-      command: echo 'hello world 2' > $output1
-      outputs:
-        output1:
-          format: txt
-          type: data
-  - tool_id: cat1
-    state:
-      input1:
-        $link: first_cat/out_file1
-      queries:
-        input2:
-          $link: embed1/output1
-test_data:
-  input1: "hello world"
-""")
-        downloaded_workflow = self._download_workflow(workflow_id)
-        # The _upload_yaml_workflow entry point uses an admin key, but if we try to
-        # do the raw re-import as a regular user we expect a 403 error.
-        response = self.workflow_populator.create_workflow_response(downloaded_workflow)
-        self._assert_status_code_is(response, 403)
-        response_dict = response.json()
-        assert response_dict["err_msg"] == "Only admin users can create tools dynamically."
-
     def test_import_annotations(self):
         workflow_id = self.workflow_populator.simple_workflow("test_import_annotations", publish=True)
         with self._different_user():
@@ -9830,47 +9792,6 @@ steps:
 class TestAdminWorkflowsApi(BaseWorkflowsApiTestCase):
     require_admin_user = True
 
-    def test_import_export_dynamic_tools(self, history_id):
-        workflow_id = self._upload_yaml_workflow("""
-class: GalaxyWorkflow
-steps:
-  - type: input
-    label: input1
-  - tool_id: cat1
-    label: first_cat
-    state:
-      input1:
-        $link: 0
-  - label: embed1
-    run:
-      class: GalaxyTool
-      version: "0.1"
-      command: echo 'hello world 2' > $output1
-      outputs:
-        output1:
-          format: txt
-          type: data
-  - tool_id: cat1
-    state:
-      input1:
-        $link: first_cat/out_file1
-      queries:
-      - input2:
-          $link: embed1/output1
-test_data:
-  input1: "hello world"
-""")
-        downloaded_workflow = self._download_workflow(workflow_id)
-        response = self.workflow_populator.create_workflow_response(downloaded_workflow)
-        workflow_id = response.json()["id"]
-        hda1 = self.dataset_populator.new_dataset(history_id, content="Hello World Second!")
-        workflow_request = dict(
-            inputs_by="name",
-            inputs=json.dumps({"input1": self._ds_entry(hda1)}),
-        )
-        self.workflow_populator.invoke_workflow_and_wait(workflow_id, history_id=history_id, request=workflow_request)
-        assert self.dataset_populator.get_history_dataset_content(history_id) == "Hello World Second!\nhello world 2\n"
-
 
 class TestCachedWorkflowsApi(BaseWorkflowsApiTestCase, ChangeDatatypeTests):
     dataset_populator: DatasetPopulator

lib/galaxy_test/api/test_workflows_from_yaml.py

@@ -299,53 +299,6 @@ steps:
         assert "no file specified" in content
         assert "7 7 5" in content
 
-    def test_workflow_embed_tool(self):
-        history_id = self.dataset_populator.new_history()
-        self._run_jobs(
-            """
-class: GalaxyWorkflow
-steps:
-  - type: input
-    label: input1
-  - tool_id: cat1
-    label: first_cat
-    state:
-      input1:
-        $link: 0
-  - label: embed1
-    run:
-      class: GalaxyTool
-      name: embed1
-      version: "0.1"
-      command: echo 'hello world 2' > $output1
-      outputs:
-        output1:
-          format: txt
-          type: data
-  - tool_id: cat1
-    state:
-      input1:
-        $link: first_cat/out_file1
-      queries:
-        - input2:
-            $link: embed1/output1
-test_data:
-  input1: "hello world"
-""",
-            history_id=history_id,
-        )
-
-        content = self.dataset_populator.get_history_dataset_content(history_id)
-        assert content == "hello world\nhello world 2\n"
-
-    def test_workflow_import_tool(self):
-        history_id = self.dataset_populator.new_history()
-        workflow_path = os.path.join(WORKFLOWS_DIRECTORY, "embed_test_1.gxwf.yml")
-        jobs_descriptions = {"test_data": {"input1": "hello world"}}
-        self._run_jobs(workflow_path, source_type="path", jobs_descriptions=jobs_descriptions, history_id=history_id)
-        content = self.dataset_populator.get_history_dataset_content(history_id)
-        assert content == "hello world\nhello world 2\n"
-
     def test_parameter_default_rep(self):
         workflow = self._upload_and_download(WORKFLOW_PARAMETER_INPUT_INTEGER_DEFAULT)
         int_input = self._steps_by_label(workflow)["int_input"]

lib/galaxy_test/base/populators.py

@@ -73,10 +73,7 @@ import cwltest.compare
 import requests
 import yaml
 from bioblend.galaxyclient import GalaxyClient
-from gxformat2 import (
-    convert_and_import_workflow,
-    ImporterGalaxyInterface,
-)
+from gxformat2 import python_to_workflow
 from gxformat2.yaml import ordered_load
 from pydantic import (
     BaseModel,
@@ -2316,6 +2313,9 @@ class BaseWorkflowPopulator(BasePopulator):
     dataset_populator: BaseDatasetPopulator
     dataset_collection_populator: "BaseDatasetCollectionPopulator"
 
+    @abstractmethod
+    def import_workflow(self, workflow, **kwds) -> dict[str, Any]: ...
+
     def load_workflow(self, name: str, content: str = workflow_str, add_pja=False) -> dict:
         workflow = json.loads(content)
         workflow["name"] = name
@@ -2368,9 +2368,33 @@ class BaseWorkflowPopulator(BasePopulator):
     def upload_yaml_workflow(self, yaml_content: YamlContentT, **kwds) -> str:
         round_trip_conversion = kwds.get("round_trip_format_conversion", False)
         client_convert = kwds.pop("client_convert", not round_trip_conversion)
-        kwds["convert"] = client_convert
-        workflow = convert_and_import_workflow(yaml_content, galaxy_interface=self, **kwds)
-        workflow_id = workflow["id"]
+        source_type = kwds.get("source_type", None)
+        workflow_directory = None
+
+        if source_type == "path":
+            assert isinstance(yaml_content, (str, os.PathLike))
+            workflow_directory = os.path.abspath(os.path.dirname(yaml_content))
+            with open(yaml_content) as f:
+                yaml_content = ordered_load(f)
+
+        if client_convert:
+            as_python = ordered_load(yaml_content) if not isinstance(yaml_content, dict) else yaml_content
+            workflow = python_to_workflow(as_python, galaxy_interface=None, workflow_directory=workflow_directory)
+        else:
+            workflow = {"yaml_content": yaml_content} if not isinstance(yaml_content, dict) else yaml_content
+
+        name = kwds.get("name")
+        if name is not None:
+            workflow["name"] = name
+        import_kwds = {"fill_defaults": kwds.get("fill_defaults", True)}
+        if kwds.get("publish"):
+            import_kwds["publish"] = True
+        if kwds.get("exact_tools"):
+            import_kwds["exact_tools"] = True
+
+        result = self.import_workflow(workflow, **import_kwds)
+        workflow_id = result["id"]
+
         if round_trip_conversion:
             workflow_yaml_wrapped = self.download_workflow(workflow_id, style="format2_wrapped_yaml")
             assert "yaml_content" in workflow_yaml_wrapped, workflow_yaml_wrapped
@@ -2679,7 +2703,6 @@ class BaseWorkflowPopulator(BasePopulator):
         extra_invocation_kwds: Optional[dict[str, Any]] = None,
         round_trip_format_conversion: bool = False,
         invocations: int = 1,
-        raw_yaml: bool = False,
         use_cached_job: bool = False,
         copy_inputs_to_history: bool = False,
         job_dir: Optional[str] = None,
@@ -2694,7 +2717,6 @@ class BaseWorkflowPopulator(BasePopulator):
             source_type=source_type,
             client_convert=client_convert,
             round_trip_format_conversion=round_trip_format_conversion,
-            raw_yaml=raw_yaml,
         )
 
         if test_data is None:
@@ -3053,14 +3075,12 @@ class RunJobsSummary(NamedTuple):
         return [j for j in self.jobs if j["tool_id"] == tool_id]
 
 
-class WorkflowPopulator(GalaxyInteractorHttpMixin, BaseWorkflowPopulator, ImporterGalaxyInterface):
+class WorkflowPopulator(GalaxyInteractorHttpMixin, BaseWorkflowPopulator):
     def __init__(self, galaxy_interactor):
         self.galaxy_interactor = galaxy_interactor
         self.dataset_populator = DatasetPopulator(galaxy_interactor)
         self.dataset_collection_populator = DatasetCollectionPopulator(galaxy_interactor)
 
-    # Required for ImporterGalaxyInterface interface - so we can recursively import
-    # nested workflows.
     def import_workflow(self, workflow, **kwds) -> dict[str, Any]:
         workflow_str = json.dumps(workflow, indent=4)
         data = {
@@ -3071,13 +3091,6 @@ class WorkflowPopulator(GalaxyInteractorHttpMixin, BaseWorkflowPopulator, Import
         assert upload_response.status_code == 200, upload_response.text
         return upload_response.json()
 
-    def import_tool(self, tool) -> dict[str, Any]:
-        """Import a new dynamically defined tool
-
-        Required to implement ImporterGalaxyInterface.
-        """
-        return self.dataset_populator.create_tool(tool)
-
     def build_module(self, step_type: str, content_id: Optional[str] = None, inputs: Optional[dict[str, Any]] = None):
         payload = {"inputs": inputs or {}, "type": step_type, "content_id": content_id}
         response = self._post("workflows/build_module", data=payload, json=True)

lib/galaxy_test/selenium/framework.py

@@ -19,10 +19,6 @@ from typing import (
 
 import requests
 import yaml
-from gxformat2 import (
-    convert_and_import_workflow,
-    ImporterGalaxyInterface,
-)
 from requests.models import Response
 from selenium.common.exceptions import NoSuchElementException
 from selenium.webdriver.common.by import By
@@ -72,7 +68,6 @@ from galaxy_test.base.env import (
 from galaxy_test.base.populators import (
     load_data_dict,
     stage_inputs,
-    YamlContentT,
 )
 from galaxy_test.base.testcase import FunctionalTestCase
 
@@ -1501,9 +1496,7 @@ class SeleniumSessionDatasetCollectionPopulator(SeleniumSessionGetPostMixin, pop
         return create_response
 
 
-class SeleniumSessionWorkflowPopulator(
-    SeleniumSessionGetPostMixin, populators.BaseWorkflowPopulator, ImporterGalaxyInterface
-):
+class SeleniumSessionWorkflowPopulator(SeleniumSessionGetPostMixin, populators.BaseWorkflowPopulator):
     """Implementation of BaseWorkflowPopulator backed by bioblend."""
 
     def __init__(self, selenium_context: GalaxySeleniumContext):
@@ -1522,9 +1515,5 @@ class SeleniumSessionWorkflowPopulator(
         upload_response.raise_for_status()
         return upload_response.json()
 
-    def upload_yaml_workflow(self, yaml_content: YamlContentT, **kwds) -> str:
-        workflow = convert_and_import_workflow(yaml_content, galaxy_interface=self, **kwds)
-        return workflow["id"]
-
 
 __all__ = ("retry_during_transitions",)

test/integration/test_fail_job_tool_unavailable.py

@@ -30,25 +30,26 @@ class TestFailJobWhenToolUnavailable(integration_util.IntegrationTestCase):
         self.workflow_populator.run_workflow(
             """
 class: GalaxyWorkflow
+inputs:
+  input1:
+    type: data
 steps:
   sleep:
-    run:
-      class: GalaxyTool
-      name: sleep hello
-      version: "0.1"
-      command: sleep 10s && echo 'hello world 2' > '$output1'
-      outputs:
-        output1:
-          type: data
-          format: txt
+    tool_id: cat_data_and_sleep
+    state:
+      sleep_time: 10
+      input1:
+        $link: input1
   cat:
     tool_id: cat1
     state:
       input1:
-        $link: sleep/output1
+        $link: sleep/out_file1
       queries:
         input2:
-          $link: sleep/output1
+          $link: sleep/out_file1
+test_data:
+  input1: "hello world"
 """,
             history_id=history_id,
             assert_ok=False,
@@ -60,7 +61,7 @@ steps:
         self.dataset_populator.wait_for_history(history_id, assert_ok=False)
         state_details = self.galaxy_interactor.get(f"histories/{history_id}").json()["state_details"]
         assert state_details["running"] == 0
-        assert state_details["ok"] == 1
+        assert state_details["ok"] == 2
         assert state_details["error"] == 1
         failed_hda = self.dataset_populator.get_history_dataset_details(
             history_id=history_id, assert_ok=False, details=True