Merge pull request #21643 from arash77/add-credentials-tests

Add support for test credentials in tools

lib/galaxy/tool_util/parser/interface.py

@@ -117,6 +117,25 @@ class ToolSourceTestInput(TypedDict):
     attributes: ToolSourceTestInputAttributes
 
 
+class DirectCredentialValue(TypedDict):
+    """Represents a credential value (variable or secret) provided directly."""
+
+    name: str
+    value: str
+
+
+class _DirectCredentialRequired(TypedDict):
+    name: str  # Name of the credentials group
+    variables: List[DirectCredentialValue]
+    secrets: List[DirectCredentialValue]
+
+
+class DirectCredential(_DirectCredentialRequired, total=False):
+    """Represents a credential group with variables and secrets provided directly."""
+
+    version: str  # Version of the credential definition (defaults to "1.0")
+
+
 ToolSourceTestInputs = List[ToolSourceTestInput]
 ToolSourceTestOutputs = List[ToolSourceTestOutput]
 TestSourceTestOutputColllection = Any
@@ -136,6 +155,7 @@ class ToolSourceTest(TypedDict):
     command: AssertionList
     command_version: AssertionList
     value_state_representation: Literal["test_case_xml", "test_case_json"]
+    credentials: Optional[List[DirectCredential]]
 
 
 class ToolSourceTests(TypedDict):

lib/galaxy/tool_util/parser/xml.py

@@ -18,6 +18,7 @@ from typing import (
 )
 
 from packaging.version import Version
+from pydantic import TypeAdapter
 
 from galaxy.tool_util.deps import requirements
 from galaxy.tool_util.parser.util import (
@@ -53,6 +54,7 @@ from galaxy.util import (
 )
 from .interface import (
     AssertionList,
+    DirectCredential,
     DrillDownDynamicOptions,
     DynamicOptions,
     InputSource,
@@ -819,6 +821,7 @@ def _test_elem_to_dict(test_elem, i, profile=None) -> ToolSourceTest:
         expect_test_failure=string_as_bool(test_elem.get("expect_test_failure", False)),
         maxseconds=test_elem.get("maxseconds", None),
         value_state_representation="test_case_xml",
+        credentials=__parse_credentials_elems(test_elem),
     )
     _copy_to_dict_if_present(test_elem, rval, ["num_outputs"])
     return rval
@@ -1093,6 +1096,29 @@ def __parse_inputs_elems(test_elem, i) -> ToolSourceTestInputs:
     return raw_inputs
 
 
+_direct_credential_adapter: TypeAdapter = TypeAdapter(List[DirectCredential])
+
+
+def __parse_credentials_elems(test_elem):
+    """
+    Parse credential definitions from test element.
+    Returns a list of DirectCredential dictionaries or None if no credentials are defined.
+    """
+    raw_list = []
+    for cred_elem in test_elem.findall("credentials"):
+        variables = [{"name": v.get("name"), "value": v.get("value")} for v in cred_elem.findall("variable")]
+        secrets = [{"name": s.get("name"), "value": s.get("value")} for s in cred_elem.findall("secret")]
+        raw: dict = {"name": cred_elem.get("name"), "variables": variables, "secrets": secrets}
+        version = cred_elem.get("version")
+        if version is not None:
+            raw["version"] = version
+        raw_list.append(raw)
+
+    if not raw_list:
+        return None
+    return _direct_credential_adapter.validate_python(raw_list)
+
+
 def _test_collection_def_dict(elem: Element) -> XmlTestCollectionDefDict:
     elements: List[TestCollectionDefElementDict] = []
     attrib: Dict[str, Any] = _element_to_dict(elem)

lib/galaxy/tool_util/parser/yaml.py

@@ -12,6 +12,7 @@ from typing import (
 )
 
 import packaging.version
+from pydantic import TypeAdapter
 
 from galaxy.tool_util.deps import requirements
 from galaxy.tool_util.parameters.convert import _select_which_when
@@ -41,6 +42,7 @@ from galaxy.util import listify
 from .interface import (
     AssertionDict,
     AssertionList,
+    DirectCredential,
     InputSource,
     PageSource,
     PagesSource,
@@ -424,9 +426,34 @@ def _parse_test(i: int, test_dict: dict) -> ToolSourceTest:
     test_dict["expect_failure"] = test_dict.get("expect_failure", False)
     test_dict["expect_test_failure"] = test_dict.get("expect_test_failure", False)
     test_dict["value_state_representation"] = "test_case_json"
+    test_dict["credentials"] = __parse_credentials_yaml(test_dict.get("credentials", None))
     return cast(ToolSourceTest, test_dict)
 
 
+_direct_credential_adapter: TypeAdapter = TypeAdapter(List[DirectCredential])
+
+
+def __parse_credentials_yaml(credentials_data) -> Optional[List[DirectCredential]]:
+    """
+    Parse credentials from YAML test definition.
+
+    Supports both list and dict formats:
+    - List: [{name: "cred1", variables: [...], secrets: [...]}]
+    - Dict: {cred1: {variables: [...], secrets: []}}
+    """
+    if not credentials_data:
+        return None
+
+    # Normalise both dict and list formats into a flat list of raw dicts.
+    if is_dict(credentials_data):
+        # {name: {variables: [], secrets: []}} → [{name: ..., variables: [], secrets: []}]
+        raw_list = [{"name": name, **cred_data} for name, cred_data in credentials_data.items()]
+    else:
+        raw_list = list(credentials_data)
+
+    return _direct_credential_adapter.validate_python(raw_list)
+
+
 def to_test_assert_list(assertions) -> AssertionList:
     assertions = assertions or []
 

lib/galaxy/tool_util/verify/_types.py

@@ -16,6 +16,7 @@ from typing_extensions import (
 
 from galaxy.tool_util.parser.interface import (
     AssertionList,
+    DirectCredential,
     TestSourceTestOutputColllection,
     ToolSourceTestOutputs,
 )
@@ -63,3 +64,4 @@ class ToolTestDescriptionDict(TypedDict):
     request_unavailable_reason: NotRequired[Optional[str]]
     maxseconds: NotRequired[Optional[int]]
     value_state_representation: NotRequired[ValueStateRepresentationT]
+    credentials: NotRequired[Optional[List[DirectCredential]]]

lib/galaxy/tool_util/verify/interactor.py

@@ -21,6 +21,7 @@ from typing import (
     List,
     NamedTuple,
     Optional,
+    Tuple,
     Union,
 )
 
@@ -46,6 +47,7 @@ from galaxy.tool_util.parameters import (
 )
 from galaxy.tool_util.parser.interface import (
     AssertionList,
+    DirectCredential,
     TestCollectionDef,
     TestCollectionOutputDef,
     TestSourceTestOutputColllection,
@@ -141,6 +143,7 @@ class ValidToolTestDict(TypedDict):
     required_files: NotRequired[RequiredFilesT]
     required_data_tables: NotRequired[RequiredDataTablesT]
     required_loc_files: NotRequired[RequiredLocFileT]
+    credentials: NotRequired[Optional[List[DirectCredential]]]
     error: Literal[False]
     tool_id: str
     tool_version: str
@@ -209,6 +212,14 @@ class RunToolResponse(NamedTuple):
     jobs: List[Dict[str, Any]]
 
 
+class ToolSubmissionResponse(NamedTuple):
+    inputs: Dict[str, Any]
+    tool_request_id: Optional[str]  # None for legacy submissions
+    submit_response_object: Dict[str, Any]  # raw validated response
+    is_legacy: bool
+    cleanup: Optional[Callable[[], None]] = None
+
+
 class InteractorStagingInterface(StagingInterface):
 
     def __init__(self, galaxy_interactor: "GalaxyInteractorApi", maxseconds: Optional[int], upload_async: bool) -> None:
@@ -241,6 +252,17 @@ class InteractorStagingInterface(StagingInterface):
         return True
 
 
+def raise_for_status(response: Response) -> None:
+    try:
+        response.raise_for_status()
+    except requests.exceptions.HTTPError as e:
+        try:
+            body = response.json()
+        except Exception:
+            body = response.text
+        raise requests.exceptions.HTTPError(f"{e} - Response body: {body}", response=response) from e
+
+
 class GalaxyInteractorApi:
     # api_key and cookies can also be manually set by UsesApiTestCaseMixin._different_user()
     api_key: Optional[str]
@@ -692,13 +714,87 @@ class GalaxyInteractorApi:
             raise ValueError(f"Invalid `location` URL: `{location}`")
         return location
 
+    def _credential_api_call(self, method: str, path: str, data: Optional[Dict[str, Any]] = None) -> Any:
+        """Low-level helper: call a credential API endpoint, raise on error, return JSON."""
+        if method == "post":
+            response = self._post(path, data=data or {}, json=True)
+        elif method == "get":
+            response = self._get(path)
+        elif method == "delete":
+            response = self._delete(path)
+        else:
+            raise ValueError(f"Unsupported method: {method}")
+        raise_for_status(response)
+        return response.json()
+
+    def _create_test_credentials(
+        self, testdef: "ToolTestDescription"
+    ) -> Tuple[List[Dict[str, Any]], Optional[List[Dict[str, Any]]]]:
+        """Create vault credentials for a test and return (created_credentials, credentials_context)."""
+        if not testdef.credentials:
+            return [], None
+
+        user_id = self._credential_api_call("get", "whoami")["id"]
+        created_credentials = []
+        credentials_context_list = []
+
+        for cred in testdef.credentials:
+            credential_payload = {
+                "source_type": "tool",
+                "source_id": testdef.tool_id,
+                "source_version": testdef.tool_version or "1.0.0",
+                "service_credential": {
+                    "name": cred["name"],
+                    "version": cred.get("version", "1.0"),
+                    "group": {
+                        "name": f"test_group_{cred['name']}",
+                        "variables": cred.get("variables", []),
+                        "secrets": cred.get("secrets", []),
+                    },
+                },
+            }
+            created_cred = self._credential_api_call("post", f"users/{user_id}/credentials", data=credential_payload)
+            all_credentials = self._credential_api_call("get", f"users/{user_id}/credentials")
+
+            # Find user_credentials_id by matching the newly-created group id.
+            user_credentials_id = None
+            for user_cred in all_credentials:
+                if (
+                    user_cred["source_type"] == "tool"
+                    and user_cred["source_id"] == testdef.tool_id
+                    and user_cred.get("source_version") == (testdef.tool_version or "1.0.0")
+                ):
+                    for group in user_cred["groups"]:
+                        if group["id"] == created_cred["id"]:
+                            user_credentials_id = user_cred["id"]
+                            break
+                    if user_credentials_id:
+                        break
+
+            if not user_credentials_id:
+                raise RuntimeError(
+                    f"Failed to find user_credentials_id for created credential group {created_cred['id']}"
+                )
+
+            created_credentials.append({"user_credentials_id": user_credentials_id, "user_id": user_id})
+            credentials_context_list.append(
+                {
+                    "user_credentials_id": user_credentials_id,
+                    "name": cred["name"],
+                    "version": cred.get("version", "1.0"),
+                    "selected_group": {"id": created_cred["id"], "name": created_cred["name"]},
+                }
+            )
+
+        return created_credentials, credentials_context_list
+
     def run_tool(
         self,
         testdef: "ToolTestDescription",
         history_id: str,
         resource_parameters: Optional[Dict[str, Any]] = None,
         use_legacy_api: UseLegacyApiT = DEFAULT_USE_LEGACY_API,
-    ) -> RunToolResponse:
+    ) -> "ToolSubmissionResponse":
         # We need to handle the case where we've uploaded a valid compressed file since the upload
         # tool will have uncompressed it on the fly.
         resource_parameters = resource_parameters or {}
@@ -708,7 +804,9 @@ class GalaxyInteractorApi:
         if testdef.value_state_representation == "test_case_json":
             # Don't submit user / YAML tools to the old endpoint.
             submit_with_legacy_api = False
-
+        if testdef.credentials:
+            # Force legacy API for credential-bearing tests since /api/tools already supports credentials_context.
+            submit_with_legacy_api = True
         if submit_with_legacy_api:
             inputs_tree = testdef.inputs.copy()
             for key, value in inputs_tree.items():
@@ -766,6 +864,12 @@ class GalaxyInteractorApi:
                 inputs_tree[f"__job_resource|{key}"] = value
 
         submit_response = None
+
+        extra_data: Dict[str, Any] = {}
+        created_credentials, credentials_context = self._create_test_credentials(testdef)
+        if credentials_context is not None:
+            extra_data["credentials_context"] = dumps(credentials_context)
+
         for _ in range(DEFAULT_TOOL_TEST_WAIT):
             submit_response = self.__submit_tool(
                 history_id,
@@ -773,6 +877,7 @@ class GalaxyInteractorApi:
                 tool_input=inputs_tree,
                 tool_version=testdef.tool_version,
                 use_legacy_api=submit_with_legacy_api,
+                extra_data=extra_data,
             )
             if _are_tool_inputs_not_ready(submit_response):
                 print("Tool inputs not ready yet")
@@ -781,8 +886,36 @@ class GalaxyInteractorApi:
             else:
                 break
         submit_response_object = ensure_tool_run_response_okay(submit_response, "execute tool", inputs_tree)
-        if not submit_with_legacy_api:
-            tool_request_id = submit_response_object["tool_request_id"]
+        tool_request_id = None if submit_with_legacy_api else submit_response_object.get("tool_request_id")
+
+        cleanup: Optional[Callable[[], None]] = None
+        if created_credentials:
+
+            def _cleanup_credentials():
+                for cred_info in created_credentials:
+                    try:
+                        self._credential_api_call(
+                            "delete", f"users/{cred_info['user_id']}/credentials/{cred_info['user_credentials_id']}"
+                        )
+                    except Exception as e:
+                        print(f"Warning: Failed to delete test credentials: {e}")
+
+            cleanup = _cleanup_credentials
+
+        return ToolSubmissionResponse(
+            inputs=inputs_tree,
+            tool_request_id=tool_request_id,
+            submit_response_object=submit_response_object,
+            is_legacy=submit_with_legacy_api,
+            cleanup=cleanup,
+        )
+
+    def resolve_tool_submission(self, submission: "ToolSubmissionResponse") -> RunToolResponse:
+        inputs_tree = submission.inputs
+        submit_response_object = submission.submit_response_object
+        if not submission.is_legacy:
+            tool_request_id = submission.tool_request_id
+            assert tool_request_id is not None
             successful = self.wait_on_tool_request(tool_request_id)
             if not successful:
                 request = self.get_tool_request(tool_request_id) or {}
@@ -791,17 +924,15 @@ class GalaxyInteractorApi:
                     inputs_tree,
                 )
             job_refs = self.jobs_for_tool_request(tool_request_id)
-            outputs = OutputsDict()
-            output_collections = {}
             if len(job_refs) != 1:
                 raise Exception(
                     f"Found incorrect number of jobs for tool request - was expecting a single job {job_refs}"
                 )
-            assert len(job_refs) == 1, job_refs
             job_id = job_refs[0]["id"]
             jobs = [self.__get_job(job_id).json()]
-            job_outputs = self.job_outputs(job_id)
-            for job_output in job_outputs:
+            outputs = OutputsDict()
+            output_collections: Dict[str, Any] = {}
+            for job_output in self.job_outputs(job_id):
                 if "dataset" in job_output:
                     outputs[job_output["name"]] = job_output["dataset"]
                 else:
@@ -1631,6 +1762,7 @@ def verify_tool(
     tool_execution_exception: Optional[Exception] = None
     input_staging_exc_info = None
     expected_failure_occurred = False
+    credential_cleanup: Optional[Callable[[], None]] = None
     begin_time = time.time()
     try:
         try:
@@ -1648,10 +1780,13 @@ def verify_tool(
             input_staging_exc_info = sys.exc_info()
             raise
         try:
-            tool_response = galaxy_interactor.run_tool(
+            submission = galaxy_interactor.run_tool(
                 testdef, test_history, resource_parameters=resource_parameters, use_legacy_api=use_legacy_api
             )
-            data_list, jobs, tool_inputs = tool_response.outputs, tool_response.jobs, tool_response.inputs
+            tool_inputs = submission.inputs
+            credential_cleanup = submission.cleanup
+            tool_response = galaxy_interactor.resolve_tool_submission(submission)
+            data_list, jobs = tool_response.outputs, tool_response.jobs
             data_collection_list = tool_response.output_collections
         except RunToolException as e:
             tool_inputs = e.inputs
@@ -1677,6 +1812,8 @@ def verify_tool(
                 job_output_exceptions = [e]
                 raise e
     finally:
+        if credential_cleanup:
+            credential_cleanup()
         if register_job_data is not None:
             end_time = time.time()
             job_data["time_seconds"] = end_time - begin_time
@@ -1909,6 +2046,7 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD
     request: Optional[Dict[str, Any]] = None
     request_schema: Optional[Dict[str, Any]] = None
     request_unavailable_reason: Optional[str] = None
+    credentials: Optional[List[DirectCredential]] = None
 
     if not error_in_test_definition:
         processed_test_dict = cast(ValidToolTestDict, processed_dict)
@@ -1937,6 +2075,7 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD
         request = processed_test_dict.get("request", None)
         request_schema = processed_test_dict.get("request_schema", None)
         request_unavailable_reason = processed_test_dict.get("request_unavailable_reason", None)
+        credentials = processed_test_dict.get("credentials", None)
     else:
         invalid_test_dict = cast(InvalidToolTestDict, processed_dict)
         maxseconds = DEFAULT_TOOL_TEST_WAIT
@@ -1970,6 +2109,7 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD
         request_schema=request_schema,
         request_unavailable_reason=request_unavailable_reason,
         value_state_representation=value_state_representation,
+        credentials=credentials,
     )
 
 
@@ -2036,6 +2176,7 @@ class ToolTestDescription:
     output_collections: List[TestCollectionOutputDef]
     maxseconds: Optional[int]
     value_state_representation: ValueStateRepresentationT
+    credentials: Optional[List[DirectCredential]]
 
     @staticmethod
     def from_tool_source_dict(processed_test_dict: ToolTestDict) -> "ToolTestDescription":
@@ -2068,6 +2209,7 @@ class ToolTestDescription:
         self.tool_version = json_dict.get("tool_version")
         self.maxseconds = json_dict.get("maxseconds")
         self.value_state_representation = json_dict.get("value_state_representation", "test_case_xml")
+        self.credentials = json_dict.get("credentials")
 
     def test_data(self):
         """
@@ -2105,6 +2247,8 @@ class ToolTestDescription:
         }
         if self.maxseconds is not None:
             test_description_def["maxseconds"] = self.maxseconds
+        if self.credentials is not None:
+            test_description_def["credentials"] = self.credentials
         return ToolTestDescriptionDict(**test_description_def)
 
 

lib/galaxy/tool_util/verify/parse.py

@@ -185,6 +185,7 @@ def _description_from_tool_source(
                 "maxseconds": maxseconds,
                 "error": False,
                 "value_state_representation": value_state_representation,
+                "credentials": raw_test_dict.get("credentials", None),
             }
         )
     except Exception:

lib/galaxy/tool_util/xsd/galaxy.xsd

@@ -1399,6 +1399,7 @@ that at least one of the assumptions of the test is not met. This is most useful
       <xs:element name="repeat" type="TestRepeat"/>
       <xs:element name="conditional" type="TestConditional"/>
       <xs:element name="section" type="TestSection"/>
+      <xs:element name="credentials" type="TestCredentials"/>
       <xs:element name="output" type="TestOutput" minOccurs="0" maxOccurs="unbounded"/>
       <xs:element name="output_collection" type="TestOutputCollection"/>
       <xs:element name="assert_command" type="TestAssertions">
@@ -1439,6 +1440,72 @@ $assertions
       </xs:element>
     </xs:choice>
   </xs:group>
+  <xs:complexType name="TestCredentials">
+    <xs:annotation>
+      <xs:documentation xml:lang="en"><![CDATA[
+
+Specify test credentials for a service that requires authentication.
+This allows tool tests to provide credential values (variables and secrets)
+that will be injected as environment variables during test execution.
+
+This is intended for use with public test accounts/credentials only.
+
+### Example
+
+```xml
+<test>
+    <credentials name="ega_credentials">
+        <variable name="ega_username" value="test_user" />
+        <secret name="ega_password" value="test_password" />
+    </credentials>
+    <param name="input" value="test.txt" />
+    <output name="output" file="expected.txt" />
+</test>
+```
+
+]]></xs:documentation>
+    </xs:annotation>
+    <xs:sequence>
+      <xs:element name="variable" minOccurs="0" maxOccurs="unbounded">
+        <xs:complexType>
+          <xs:attribute name="name" type="xs:string" use="required">
+            <xs:annotation>
+              <xs:documentation xml:lang="en">Name of the credential variable (must match a variable defined in the tool's requirements/credentials)</xs:documentation>
+            </xs:annotation>
+          </xs:attribute>
+          <xs:attribute name="value" type="xs:string" use="required">
+            <xs:annotation>
+              <xs:documentation xml:lang="en">Test value for this variable</xs:documentation>
+            </xs:annotation>
+          </xs:attribute>
+        </xs:complexType>
+      </xs:element>
+      <xs:element name="secret" minOccurs="0" maxOccurs="unbounded">
+        <xs:complexType>
+          <xs:attribute name="name" type="xs:string" use="required">
+            <xs:annotation>
+              <xs:documentation xml:lang="en">Name of the credential secret (must match a secret defined in the tool's requirements/credentials)</xs:documentation>
+            </xs:annotation>
+          </xs:attribute>
+          <xs:attribute name="value" type="xs:string" use="required">
+            <xs:annotation>
+              <xs:documentation xml:lang="en">Test value for this secret (use public test credentials only)</xs:documentation>
+            </xs:annotation>
+          </xs:attribute>
+        </xs:complexType>
+      </xs:element>
+    </xs:sequence>
+    <xs:attribute name="name" type="xs:string" use="required">
+      <xs:annotation>
+        <xs:documentation xml:lang="en">Name of the credentials group (must match the name defined in tool's requirements/credentials)</xs:documentation>
+      </xs:annotation>
+    </xs:attribute>
+    <xs:attribute name="version" type="xs:string" default="1.0">
+      <xs:annotation>
+        <xs:documentation xml:lang="en">Version of the credentials definition (must match the version defined in tool's requirements/credentials). Defaults to "1.0".</xs:documentation>
+      </xs:annotation>
+    </xs:attribute>
+  </xs:complexType>
   <xs:complexType name="TestSection">
     <xs:annotation>
       <xs:documentation xml:lang="en"><![CDATA[

lib/galaxy/tools/actions/__init__.py

@@ -972,6 +972,7 @@ class DefaultToolAction(ToolAction):
         if credentials_context is None:
             return
 
+        # Create database associations for vault-based credentials
         for service_context in credentials_context.root:
             association = JobCredentialsContextAssociation(
                 job=job,

test/functional/test_toolbox_pytest.py

@@ -13,6 +13,7 @@ from galaxy.tool_util.verify.interactor import (
 )
 from galaxy_test.api._framework import ApiTestCase
 from galaxy_test.driver.driver_util import GalaxyTestDriver
+from galaxy_test.driver.integration_util import ConfiguresDatabaseVault
 
 SKIPTEST = os.path.join(os.path.dirname(__file__), "known_broken_tools.txt")
 
@@ -60,10 +61,16 @@ def idfn(val: ToolTest):
     return f"{val.tool_id}/{val.tool_version}-{val.test_index}"
 
 
-class TestFrameworkTools(ApiTestCase):
+class TestFrameworkTools(ApiTestCase, ConfiguresDatabaseVault):
     conda_auto_init = True
     conda_auto_install = True
 
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        """Configure vault for credential testing."""
+        super().handle_galaxy_config_kwds(config)
+        cls._configure_database_vault(config)
+
     @pytest.mark.parametrize("testcase", cases(), ids=idfn)
     def test_tool(self, testcase: ToolTest):
         use_legacy_api = os.environ.get("GALAXY_TEST_USE_LEGACY_TOOL_API", DEFAULT_USE_LEGACY_API)

test/functional/tools/credentials_test.xml

@@ -0,0 +1,52 @@
+<tool id="credentials_test" name="Credentials Test Tool" version="1.0.0" profile="25.1">
+    <description>Test tool for validating credential injection</description>
+    <requirements>
+        <credentials name="test_service" version="1.0">
+            <variable name="TEST_USERNAME" inject_as_env="TEST_USERNAME"/>
+            <secret name="TEST_PASSWORD" inject_as_env="TEST_PASSWORD"/>
+        </credentials>
+    </requirements>
+    <command><![CDATA[
+        echo "Username: \$TEST_USERNAME" > '$output' &&
+        echo "Password length: \${#TEST_PASSWORD}" >> '$output'
+    ]]></command>
+    <inputs>
+        <param name="input_text" type="text" value="default" label="Input text"/>
+    </inputs>
+    <outputs>
+        <data name="output" format="txt"/>
+    </outputs>
+    <tests>
+        <test>
+            <credentials name="test_service">
+                <variable name="TEST_USERNAME" value="test_user"/>
+                <secret name="TEST_PASSWORD" value="test_password_123"/>
+            </credentials>
+            <param name="input_text" value="test_value"/>
+            <output name="output">
+                <assert_contents>
+                    <has_text text="Username: test_user"/>
+                    <has_text text="Password length: 17"/>
+                </assert_contents>
+            </output>
+        </test>
+        <test>
+            <credentials name="test_service">
+                <variable name="TEST_USERNAME" value="another_user"/>
+                <secret name="TEST_PASSWORD" value="secret"/>
+            </credentials>
+            <param name="input_text" value="another_test"/>
+            <output name="output">
+                <assert_contents>
+                    <has_text text="Username: another_user"/>
+                    <has_text text="Password length: 6"/>
+                </assert_contents>
+            </output>
+        </test>
+    </tests>
+    <help><![CDATA[
+This is a test tool to validate that credentials are properly injected as environment variables during tool execution.
+
+The tool echoes the TEST_USERNAME environment variable and the length of the TEST_PASSWORD to verify they were set.
+    ]]></help>
+</tool>

test/functional/tools/sample_tool_conf.xml

@@ -335,6 +335,7 @@
   <tool file="configfile_user_defined.yml" />
   <tool file="from_work_dir_glob.xml" />
   <tool file="secret_tool.xml" />
+  <tool file="credentials_test.xml" />
 
   <!-- Tools without tool test but useful for hand-crafted, artisanal test cases. -->
   <tool file="cat_data_and_sleep.xml" />

test/unit/tool_util/test_credential_parsing.py

@@ -0,0 +1,43 @@
+"""Unit tests for credential parsing from tool test definitions."""
+
+from galaxy.tool_util.unittest_utils import functional_test_tool_path
+from galaxy.util.unittest import TestCase
+from .test_parsing import FunctionalTestToolTestCase
+
+
+class TestCredentialParsing(FunctionalTestToolTestCase):
+    test_path = "credentials_test.xml"
+
+    def test_credentials_parsing(self):
+        tests_dict = self._tool_source.parse_tests_to_dict()
+        tests = tests_dict["tests"]
+        assert len(tests) == 2
+
+        # First test: test_user / test_password_123
+        creds = tests[0]["credentials"]
+        assert creds is not None
+        assert len(creds) == 1
+        cred = creds[0]
+        assert cred["name"] == "test_service"
+        assert cred["variables"] == [{"name": "TEST_USERNAME", "value": "test_user"}]
+        assert cred["secrets"] == [{"name": "TEST_PASSWORD", "value": "test_password_123"}]
+
+        # Second test: another_user / secret
+        creds2 = tests[1]["credentials"]
+        assert creds2 is not None
+        assert len(creds2) == 1
+        assert creds2[0]["variables"] == [{"name": "TEST_USERNAME", "value": "another_user"}]
+        assert creds2[0]["secrets"] == [{"name": "TEST_PASSWORD", "value": "secret"}]
+
+
+class TestNoCredentials(TestCase):
+    """Verify tools without credentials return None."""
+
+    def test_no_credentials_field(self):
+        from galaxy.tool_util.parser.factory import get_tool_source
+
+        path = functional_test_tool_path("simple_constructs.xml")
+        tool_source = get_tool_source(path)
+        tests = tool_source.parse_tests_to_dict()["tests"]
+        for test in tests:
+            assert test.get("credentials") is None