diff --git a/client/src/components/Markdown/Sections/MarkdownGalaxy.vue b/client/src/components/Markdown/Sections/MarkdownGalaxy.vue
index c91359d851a..21dded1c5be 100644
--- a/client/src/components/Markdown/Sections/MarkdownGalaxy.vue
+++ b/client/src/components/Markdown/Sections/MarkdownGalaxy.vue
@@ -158,11 +158,11 @@ watch(
:content="`Galaxy Version ${config.version_major}`" />
diff --git a/client/src/components/Tool/ToolForm.vue b/client/src/components/Tool/ToolForm.vue
index 60c8086acb8..3999709f3cd 100644
--- a/client/src/components/Tool/ToolForm.vue
+++ b/client/src/components/Tool/ToolForm.vue
@@ -275,10 +275,11 @@ export default {
},
hasCredentialsErrors() {
if (this.formConfig.credentials?.length) {
- const { hasUserProvidedAllRequiredServiceCredentials } = useUserToolCredentials(
- this.formConfig.id,
- this.formConfig.version,
- );
+ const { hasUserProvidedAllRequiredServiceCredentials, toolHasRequiredServiceCredentials } =
+ useUserToolCredentials(this.formConfig.id, this.formConfig.version);
+ if (!toolHasRequiredServiceCredentials.value) {
+ return false;
+ }
return !hasUserProvidedAllRequiredServiceCredentials.value;
}
return false;
diff --git a/doc/source/admin/galaxy_options.rst b/doc/source/admin/galaxy_options.rst
index 91bca499024..f42536e9d83 100644
--- a/doc/source/admin/galaxy_options.rst
+++ b/doc/source/admin/galaxy_options.rst
@@ -5927,4 +5927,15 @@
:Type: int
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+``enable_beta_tool_formats``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+:Description:
+ Enable beta tool formats (yaml, cwl, ...) which is a prerequisite
+ for user defined tools.
+:Default: ``false``
+:Type: bool
+
+
diff --git a/lib/galaxy/config/sample/galaxy.yml.sample b/lib/galaxy/config/sample/galaxy.yml.sample
index a6b611ca089..c2db28f3406 100644
--- a/lib/galaxy/config/sample/galaxy.yml.sample
+++ b/lib/galaxy/config/sample/galaxy.yml.sample
@@ -3180,3 +3180,7 @@ galaxy:
# ``true``. Runs in a Celery task.
#failed_jobs_working_directory_cleanup_interval: 86400
+ # Enable beta tool formats (yaml, cwl, ...) which is a prerequisite
+ # for user defined tools.
+ #enable_beta_tool_formats: false
+
diff --git a/lib/galaxy/config/schemas/config_schema.yml b/lib/galaxy/config/schemas/config_schema.yml
index cbbd52acdd8..5c4788cfc8f 100644
--- a/lib/galaxy/config/schemas/config_schema.yml
+++ b/lib/galaxy/config/schemas/config_schema.yml
@@ -4374,3 +4374,10 @@ mapping:
default: 86400
desc: |
The interval in seconds between attempts to delete all failed Galaxy job's working directories from the filesystem (every 24 hours by default) if enable_failed_jobs_working_directory_cleanup is ``true``. Runs in a Celery task.
+
+ enable_beta_tool_formats:
+ type: bool
+ default: false
+ required: false
+ desc: |
+ Enable beta tool formats (yaml, cwl, ...) which is a prerequisite for user defined tools.
\ No newline at end of file
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index ad20f3b389a..fd7cf44d9ba 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -1496,6 +1496,17 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle):
self.javascript_requirements = javasscript_requirements
self.credentials = credentials
+ # Add credential inject_as_env names to docker_env_pass_through
+ # so they are passed into containerized environments (Docker -e, Singularity SINGULARITYENV_)
+ if self.credentials:
+ if not self.docker_env_pass_through:
+ self.docker_env_pass_through = []
+ for credential in self.credentials:
+ for secret in credential.secrets:
+ self.docker_env_pass_through.append(secret.inject_as_env)
+ for variable in credential.variables:
+ self.docker_env_pass_through.append(variable.inject_as_env)
+
required_files = tool_source.parse_required_files()
if required_files is None:
old_id = self.old_id
diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py
index 1fd55b911d9..903717b2d89 100644
--- a/lib/galaxy/workflow/modules.py
+++ b/lib/galaxy/workflow/modules.py
@@ -29,6 +29,7 @@ from galaxy.exceptions import (
)
from galaxy.job_execution.actions.post import ActionBox
from galaxy.job_execution.compute_environment import ComputeEnvironment
+from galaxy.managers.credentials import _build_user_credentials_query
from galaxy.model import (
Job,
PostJobAction,
@@ -42,6 +43,11 @@ from galaxy.model.dataset_collections import matching
from galaxy.model.dataset_collections.query import HistoryQuery
from galaxy.model.dataset_collections.type_description import COLLECTION_TYPE_DESCRIPTION_FACTORY
from galaxy.model.dataset_collections.types.sample_sheet_util import validate_column_definitions
+from galaxy.schema.credentials import (
+ CredentialsContext,
+ SelectedGroup,
+ ServiceCredentialsContext,
+)
from galaxy.schema.invocation import (
CancelReason,
FailureReason,
@@ -2502,6 +2508,7 @@ class ToolModule(WorkflowModule):
if pja.action_type == "ValidateOutputsAction":
validate_outputs = True
+ credentials_context = self._resolve_credentials_context(tool)
execution_tracker = execute(
trans=self.trans,
tool=tool,
@@ -2517,6 +2524,7 @@ class ToolModule(WorkflowModule):
),
completed_jobs=completed_jobs,
workflow_resource_parameters=resource_parameters,
+ credentials_context=credentials_context,
)
complete = True
except PartialJobExecution as pje:
@@ -2580,6 +2588,38 @@ class ToolModule(WorkflowModule):
self.trans, self.trans.sa_session, pja, step_inputs, step_outputs, replacement_dict
)
+ def _resolve_credentials_context(self, tool: "Tool") -> Optional[CredentialsContext]:
+ """Auto-resolve the user's current credentials for a tool in workflow execution."""
+ if not tool.credentials:
+ return None
+ trans = self.trans
+ if not trans.user:
+ return None
+ stmt = _build_user_credentials_query(
+ user_id=trans.user.id,
+ source_type="tool",
+ source_id=tool.id,
+ current_group_only=True,
+ )
+ results = trans.sa_session.execute(stmt).tuples().all()
+ if not results:
+ return None
+ encode = trans.security.encode_id
+ seen = {}
+ for user_cred, group, _cred in results:
+ key = (user_cred.id, user_cred.name, user_cred.version)
+ if key not in seen:
+ seen[key] = ServiceCredentialsContext(
+ user_credentials_id=encode(user_cred.id),
+ name=user_cred.name,
+ version=user_cred.version,
+ selected_group=SelectedGroup(
+ id=encode(group.id),
+ name=group.name,
+ ),
+ )
+ return CredentialsContext(root=list(seen.values()))
+
def _handle_post_job_actions(self, step, job, replacement_dict):
# Create new PJA associations with the created job, to be run on completion.
# PJA Parameter Replacement (only applies to immediate actions-- rename specifically, for now)
diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py
index fe315426c2f..f70372b5556 100644
--- a/lib/galaxy_test/base/populators.py
+++ b/lib/galaxy_test/base/populators.py
@@ -1214,6 +1214,8 @@ class BaseDatasetPopulator(BasePopulator):
kwds["__files"][key] = value
del inputs[key]
+ if "credentials_context" in kwds and not isinstance(kwds["credentials_context"], str):
+ kwds["credentials_context"] = json.dumps(kwds["credentials_context"])
return dict(tool_id=tool_id, inputs=json.dumps(inputs), history_id=history_id, **kwds)
def build_tool_state(self, tool_id: str, history_id: str):
@@ -2086,6 +2088,204 @@ class DatasetPopulator(GalaxyInteractorHttpMixin, BaseDatasetPopulator):
yield history_id
+class BaseCredentialsPopulator(BasePopulator):
+ """Abstract base class for credential operations in Galaxy tests."""
+
+ DEFAULT_SOURCE_TYPE = "tool"
+ DEFAULT_SOURCE_VERSION = "test"
+ DEFAULT_SERVICE_NAME = "service1"
+ DEFAULT_SERVICE_VERSION = "v1"
+
+ def build_credentials_payload(
+ self,
+ tool_id: str,
+ variables: list,
+ secrets: list,
+ source_type: str = DEFAULT_SOURCE_TYPE,
+ source_version: str = DEFAULT_SOURCE_VERSION,
+ service_name: str = DEFAULT_SERVICE_NAME,
+ service_version: str = DEFAULT_SERVICE_VERSION,
+ group_name: Optional[str] = None,
+ ) -> dict:
+ """Build and return a credentials payload dict without posting it."""
+ if group_name is None:
+ group_name = random_name()
+ return {
+ "source_type": source_type,
+ "source_id": tool_id,
+ "source_version": source_version,
+ "service_credential": {
+ "name": service_name,
+ "version": service_version,
+ "group": {
+ "name": group_name,
+ "variables": [dict(v) for v in variables],
+ "secrets": [dict(s) for s in secrets],
+ },
+ },
+ }
+
+ def post_credentials(self, payload: dict, expected_status: int = 200, anon: bool = False) -> dict:
+ """Post credentials payload, assert expected status, and return the response JSON."""
+ response = self._post("/api/users/current/credentials", data=payload, json=True, anon=anon)
+ api_asserts.assert_status_code_is(response, expected_status)
+ return response.json()
+
+ def create_credentials(
+ self,
+ tool_id: str,
+ variables: list,
+ secrets: list,
+ expected_status: int = 200,
+ anon: bool = False,
+ **kwargs,
+ ) -> dict:
+ """Build and post credentials in one step. Returns response JSON."""
+ payload = self.build_credentials_payload(tool_id=tool_id, variables=variables, secrets=secrets, **kwargs)
+ return self.post_credentials(payload, expected_status=expected_status, anon=anon)
+
+ def list_credentials(
+ self,
+ source_type: Optional[str] = None,
+ source_id: Optional[str] = None,
+ include_definition: bool = False,
+ expected_status: int = 200,
+ ) -> list:
+ """Return credentials for the current user, optionally filtered by source or definition."""
+ params = []
+ if source_type is not None:
+ params.append(f"source_type={source_type}")
+ if source_id is not None:
+ params.append(f"source_id={source_id}")
+ if include_definition:
+ params.append("include_definition=true")
+ url = "/api/users/current/credentials"
+ if params:
+ url += "?" + "&".join(params)
+ response = self._get(url)
+ api_asserts.assert_status_code_is(response, expected_status)
+ return response.json()
+
+ def get_credentials(self, source_type: str, source_id: str) -> list:
+ """Return credentials list for the given source."""
+ return self.list_credentials(source_type=source_type, source_id=source_id)
+
+ def update_credentials_group(
+ self, user_credentials_id: str, group_id: str, payload: dict, expected_status: int = 200
+ ) -> dict:
+ """PUT /api/users/current/credentials/{id}/groups/{gid} and return response JSON."""
+ response = self._put(
+ f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}", data=payload, json=True
+ )
+ api_asserts.assert_status_code_is(response, expected_status)
+ return response.json()
+
+ def select_current_group(
+ self,
+ source_type: str,
+ source_id: str,
+ source_version: str,
+ user_credentials_id: str,
+ current_group_id: Optional[str],
+ expected_status: int = 204,
+ ) -> None:
+ """PUT /api/users/current/credentials to select (or unset) the current group."""
+ payload = {
+ "source_type": source_type,
+ "source_id": source_id,
+ "source_version": source_version,
+ "service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": current_group_id}],
+ }
+ response = self._put("/api/users/current/credentials", data=payload, json=True)
+ api_asserts.assert_status_code_is(response, expected_status)
+
+ def delete_service_credentials(self, user_credentials_id: str, expected_status: int = 204) -> None:
+ """DELETE /api/users/current/credentials/{id}."""
+ response = self._delete(f"/api/users/current/credentials/{user_credentials_id}")
+ api_asserts.assert_status_code_is(response, expected_status)
+
+ def delete_credentials_group(self, user_credentials_id: str, group_id: str, expected_status: int = 204) -> None:
+ """DELETE /api/users/current/credentials/{id}/groups/{gid}."""
+ response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}")
+ api_asserts.assert_status_code_is(response, expected_status)
+
+ def setup_credentials_context(
+ self,
+ tool_id: str,
+ variables: list,
+ secrets: list,
+ service_name: str = DEFAULT_SERVICE_NAME,
+ service_version: str = DEFAULT_SERVICE_VERSION,
+ group_name: str = "default",
+ source_version: str = DEFAULT_SOURCE_VERSION,
+ ) -> list:
+ """Create credentials and return a ready-to-use credentials_context list for tool execution.
+
+ Idempotent: if the service credentials and group already exist (e.g. from a prior test
+ in the same server session), reuses them instead of failing with a 409 conflict.
+ Also selects the group as the current group so server-side resolution
+ (e.g. during workflow execution) can find it via current_group_id.
+ """
+ # Check whether credentials already exist for this service.
+ existing = self.get_credentials("tool", tool_id)
+ user_cred_entry = next(
+ (c for c in existing if c.get("name") == service_name and c.get("version") == service_version),
+ None,
+ )
+ if user_cred_entry:
+ user_credentials_id = user_cred_entry["id"]
+ created_group = next(
+ (g for g in user_cred_entry.get("groups", []) if g["name"] == group_name),
+ None,
+ )
+ if created_group is None:
+ # User credentials exist but this group doesn't — create only the group.
+ created_group = self.create_credentials(
+ tool_id=tool_id,
+ variables=variables,
+ secrets=secrets,
+ service_name=service_name,
+ service_version=service_version,
+ group_name=group_name,
+ source_version=source_version,
+ )
+ else:
+ created_group = self.create_credentials(
+ tool_id=tool_id,
+ variables=variables,
+ secrets=secrets,
+ service_name=service_name,
+ service_version=service_version,
+ group_name=group_name,
+ source_version=source_version,
+ )
+ credentials_list = self.get_credentials("tool", tool_id)
+ user_credentials_id = credentials_list[0]["id"]
+ self.select_current_group(
+ source_type="tool",
+ source_id=tool_id,
+ source_version=source_version,
+ user_credentials_id=user_credentials_id,
+ current_group_id=created_group["id"],
+ )
+ return [
+ {
+ "user_credentials_id": user_credentials_id,
+ "name": service_name,
+ "version": service_version,
+ "selected_group": {
+ "id": created_group["id"],
+ "name": created_group["name"],
+ },
+ }
+ ]
+
+
+class CredentialsPopulator(GalaxyInteractorHttpMixin, BaseCredentialsPopulator):
+ def __init__(self, galaxy_interactor: ApiTestInteractor) -> None:
+ self.galaxy_interactor = galaxy_interactor
+
+
# Things gxformat2 knows how to upload as workflows
YamlContentT = Union[StrPath, dict]
diff --git a/test/functional/tools/secret_tool.xml b/test/functional/tools/secret_tool.xml
index a5d46bac93c..7fd4d4040b7 100644
--- a/test/functional/tools/secret_tool.xml
+++ b/test/functional/tools/secret_tool.xml
@@ -1,5 +1,6 @@
+ busybox:1.36.1-glibc
diff --git a/test/integration/test_containerized_jobs.py b/test/integration/test_containerized_jobs.py
index 913821bcba5..c00b8a6ee90 100644
--- a/test/integration/test_containerized_jobs.py
+++ b/test/integration/test_containerized_jobs.py
@@ -3,13 +3,19 @@
import json
import os
import unittest
-from typing import (
- Any,
-)
+from typing import Any
from galaxy.util.commands import which
-from galaxy_test.base.populators import DatasetPopulator
-from galaxy_test.driver.integration_util import IntegrationTestCase
+from galaxy_test.base.populators import (
+ CredentialsPopulator,
+ DatasetPopulator,
+ skip_without_tool,
+ WorkflowPopulator,
+)
+from galaxy_test.driver.integration_util import (
+ ConfiguresDatabaseVault,
+ IntegrationTestCase,
+)
from .test_job_environments import BaseJobEnvironmentIntegrationTestCase
SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
@@ -22,6 +28,10 @@ SINGULARITY_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "singularity_job_co
EXTENDED_TIMEOUT = 120
+CREDENTIALS_TEST_TOOL = "secret_tool"
+CONTAINER_TEST_VARIABLES = [{"name": "server", "value": "http://test-server:8080"}]
+CONTAINER_TEST_SECRETS = [{"name": "username", "value": "test_user"}, {"name": "password", "value": "test_pass"}]
+
class MulledJobTestCases:
"""
@@ -29,6 +39,7 @@ class MulledJobTestCases:
"""
dataset_populator: DatasetPopulator
+ credentials_populator: CredentialsPopulator
container_type: str
def _run_and_get_contents(self, tool_id: str, history_id: str):
@@ -98,7 +109,7 @@ def skip_if_container_type_unavailable(cls) -> None:
raise unittest.SkipTest(f"Executable '{cls.container_type}' not found on PATH")
-class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, MulledJobTestCases):
+class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, MulledJobTestCases, ConfiguresDatabaseVault):
dataset_populator: DatasetPopulator
jobs_directory: str
job_config_file = DOCKERIZED_JOB_CONFIG_FILE
@@ -112,6 +123,7 @@ class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, Mulle
config["jobs_directory"] = cls.jobs_directory
config["job_config_file"] = cls.job_config_file
disable_dependency_resolution(config)
+ cls._configure_database_vault(config)
@classmethod
def setUpClass(cls) -> None:
@@ -120,6 +132,14 @@ class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, Mulle
def setUp(self) -> None:
super().setUp()
+ self.credentials_populator = CredentialsPopulator(self.galaxy_interactor)
+ self.workflow_populator = WorkflowPopulator(self.galaxy_interactor)
+
+ def _setup_credentials_context(self, **kwargs):
+ kwargs.setdefault("tool_id", CREDENTIALS_TEST_TOOL)
+ kwargs.setdefault("variables", CONTAINER_TEST_VARIABLES)
+ kwargs.setdefault("secrets", CONTAINER_TEST_SECRETS)
+ return self.credentials_populator.setup_credentials_context(**kwargs)
def test_container_job_environment(self) -> None:
"""
@@ -224,6 +244,75 @@ class TestDockerizedJobsIntegration(BaseJobEnvironmentIntegrationTestCase, Mulle
"""
assert identifier == f"quay.io/local/{expected_hash}"
+ @skip_without_tool("secret_tool")
+ def test_credentials_passed_to_container(self) -> None:
+ """
+ Test that tool credentials are passed as environment variables into containerized environments.
+ """
+ credentials_context = self._setup_credentials_context()
+
+ # Run the containerized tool that outputs credential environment variables
+ with self.dataset_populator.test_history() as history_id:
+ run_response = self.dataset_populator.run_tool(
+ "secret_tool", {}, history_id, credentials_context=credentials_context
+ )
+ job_id = run_response["jobs"][0]["id"]
+ self.dataset_populator.wait_for_job(job_id=job_id, assert_ok=True, timeout=EXTENDED_TIMEOUT)
+
+ # Get the output - should contain the credential environment variables
+ output = self.dataset_populator.get_history_dataset_content(
+ history_id, content_id=run_response["outputs"][0]["id"]
+ )
+
+ # Verify that credential environment variables were available in the container
+ lines = output.strip().split("\n")
+ assert len(lines) == 3, f"Expected 3 lines in output, got {len(lines)}: {lines}"
+ assert lines[0] == "http://test-server:8080", f"Expected server URL in first line, got: {lines[0]}"
+ assert lines[1] == "test_user", f"Expected username in second line, got: {lines[1]}"
+ assert lines[2] == "test_pass", f"Expected password in third line, got: {lines[2]}"
+
+ @skip_without_tool("secret_tool")
+ def test_credentials_passed_to_container_in_workflow(self) -> None:
+ """
+ Test that tool credentials are passed into containerized workflow steps.
+
+ This is a regression test for issue #21715: when a tool using credentials
+ is run as a workflow step, the credentials are silently dropped because
+ ToolModule.execute() does not forward credentials_context.
+ """
+ self._setup_credentials_context()
+
+ workflow_yaml = """
+class: GalaxyWorkflow
+steps:
+ secret_step:
+ tool_id: secret_tool
+"""
+ workflow_id = self.workflow_populator.upload_yaml_workflow(workflow_yaml)
+
+ with self.dataset_populator.test_history() as history_id:
+ self.workflow_populator.invoke_workflow_and_wait(
+ workflow_id,
+ history_id=history_id,
+ assert_ok=True,
+ )
+
+ # Get the single output dataset from the history
+ history_contents = self.dataset_populator.get_history_contents(history_id)
+ datasets = [item for item in history_contents if item["history_content_type"] == "dataset"]
+ assert len(datasets) == 1, f"Expected 1 output dataset, got {len(datasets)}"
+ output = self.dataset_populator.get_history_dataset_content(
+ history_id, content_id=datasets[0]["id"], timeout=EXTENDED_TIMEOUT
+ )
+
+ # Verify that credential environment variables were available in the container.
+ # This will fail because credentials are not forwarded through workflow execution.
+ lines = output.strip().split("\n")
+ assert len(lines) == 3, f"Expected 3 lines in output, got {len(lines)}: {lines}"
+ assert lines[0] == "http://test-server:8080", f"Expected server URL in first line, got: {lines[0]}"
+ assert lines[1] == "test_user", f"Expected username in second line, got: {lines[1]}"
+ assert lines[2] == "test_pass", f"Expected password in third line, got: {lines[2]}"
+
class TestMappingContainerResolver(IntegrationTestCase):
"""
diff --git a/test/integration/test_credentials.py b/test/integration/test_credentials.py
index d3aa13485cc..e7cfb039b63 100644
--- a/test/integration/test_credentials.py
+++ b/test/integration/test_credentials.py
@@ -3,10 +3,15 @@ from typing import Optional
from galaxy.model.db.user import get_user_by_email
from galaxy.security.vault import UserVaultWrapper
from galaxy_test.base.api_util import random_name
-from galaxy_test.base.populators import skip_without_tool
+from galaxy_test.base.populators import (
+ CredentialsPopulator,
+ skip_without_tool,
+)
from galaxy_test.driver import integration_util
CREDENTIALS_TEST_TOOL = "secret_tool"
+DEFAULT_TOOL_VARIABLES = [{"name": "server", "value": "http://localhost:8080"}]
+DEFAULT_TOOL_SECRETS = [{"name": "username", "value": "user"}, {"name": "password", "value": "pass"}]
class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.ConfiguresDatabaseVault):
@@ -15,28 +20,27 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
super().handle_galaxy_config_kwds(config)
cls._configure_database_vault(config)
+ def setUp(self):
+ super().setUp()
+ self.credentials_populator = CredentialsPopulator(self.galaxy_interactor)
+
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_provide_credential(self):
- payload = self._build_credentials_payload(group_name="default")
- created_credential_group = self._provide_user_credentials(payload=payload)
+ created_credential_group = self._create_credentials(group_name="default")
assert created_credential_group["name"] == "default"
assert len(created_credential_group["variables"]) == 1
assert len(created_credential_group["secrets"]) == 2
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_anon_users_cannot_provide_credentials(self):
- payload = self._build_credentials_payload()
- response = self._post("/api/users/current/credentials", data=payload, json=True, anon=True)
- self._assert_status_code_is(response, 403)
+ self._create_credentials(expected_status=403, anon=True)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_list_user_credentials(self):
- self._provide_user_credentials()
+ self._create_credentials()
# Check there is at least one credential
- response = self._get("/api/users/current/credentials")
- self._assert_status_code_is(response, 200)
- list_user_credentials = response.json()
+ list_user_credentials = self.credentials_populator.list_credentials()
assert len(list_user_credentials) > 0
# Check the specific credential exists
@@ -44,7 +48,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_other_users_cannot_list_credentials(self):
- self._provide_user_credentials()
+ self._create_credentials()
self._check_credentials_exist()
@@ -52,12 +56,10 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
self._check_credentials_exist(num_credentials=0)
def test_list_by_source_id_requires_source_type(self):
- response = self._get("/api/users/current/credentials?source_id={CREDENTIALS_TEST_TOOL}")
- self._assert_status_code_is(response, 400)
+ self.credentials_populator.list_credentials(source_id=CREDENTIALS_TEST_TOOL, expected_status=400)
def test_list_unsupported_source_type(self):
- response = self._get("/api/users/current/credentials?source_type=invalid")
- self._assert_status_code_is(response, 400)
+ self.credentials_populator.list_credentials(source_type="invalid", expected_status=400)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_add_group_to_credentials(self):
@@ -65,8 +67,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
initial_group = random_name()
# First, create initial credentials with a unique group
- payload = self._build_credentials_payload(group_name=initial_group)
- self._provide_user_credentials(payload)
+ self._create_credentials(group_name=initial_group)
initial_credentials = self._check_credentials_exist()
assert len(initial_credentials) == 1
@@ -76,8 +77,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
# Create a new group with the same service credentials
second_group = random_name()
- new_payload = self._build_credentials_payload(group_name=second_group)
- self._provide_user_credentials(new_payload)
+ self._create_credentials(group_name=second_group)
# Check that both our groups exist
updated_credentials = self._check_credentials_exist()
@@ -90,8 +90,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_update_credentials_update_time(self):
- payload = self._build_credentials_payload()
- created_group = self._provide_user_credentials(payload)
+ created_group = self._create_credentials()
created_group_id = created_group["id"]
list_user_credentials = self._check_credentials_exist()
@@ -117,7 +116,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_update_credentials(self):
# Create initial credentials
- initial_group = self._provide_user_credentials()
+ initial_group = self._create_credentials()
group_id = initial_group["id"]
list_user_credentials = self._check_credentials_exist()
@@ -162,7 +161,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_update_credentials_error_cases(self):
"""Test update error scenarios."""
- group = self._provide_user_credentials()
+ group = self._create_credentials()
group_id = group["id"]
list_user_credentials = self._check_credentials_exist()
@@ -227,15 +226,14 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_delete_service_credentials(self):
# Create credentials
- self._provide_user_credentials()
+ self._create_credentials()
# Check credentials exist and get the service credentials ID
credentials_list = self._check_credentials_exist()
service_credentials_id = credentials_list[0]["id"]
# Delete the entire service credentials
- response = self._delete(f"/api/users/current/credentials/{service_credentials_id}")
- self._assert_status_code_is(response, 204)
+ self.credentials_populator.delete_service_credentials(service_credentials_id)
# Check credentials are deleted
self._check_credentials_exist(num_credentials=0)
@@ -247,12 +245,10 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
target_group_name = random_name()
# Create initial credentials with unique group
- payload1 = self._build_credentials_payload(group_name=initial_group)
- self._provide_user_credentials(payload1)
+ self._create_credentials(group_name=initial_group)
# Add a new group
- new_payload = self._build_credentials_payload(group_name=target_group_name)
- self._provide_user_credentials(new_payload)
+ self._create_credentials(group_name=target_group_name)
# Check credentials exist with both our groups
list_user_credentials = self._check_credentials_exist()
@@ -267,22 +263,16 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
target_group_id = groups_before[target_group_name]["id"]
# Set the new group as current
- select_payload = {
- "source_type": "tool",
- "source_id": CREDENTIALS_TEST_TOOL,
- "source_version": "test",
- "service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": target_group_id}],
- }
- response = self._put("/api/users/current/credentials", data=select_payload, json=True)
- self._assert_status_code_is(response, 204)
+ self.credentials_populator.select_current_group(
+ "tool", CREDENTIALS_TEST_TOOL, "test", user_credentials_id, target_group_id
+ )
# Verify it's set as current
list_user_credentials = self._check_credentials_exist()
assert list_user_credentials[0]["current_group_id"] == target_group_id
# Delete the group
- response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{target_group_id}")
- self._assert_status_code_is(response, 204)
+ self.credentials_populator.delete_credentials_group(user_credentials_id, target_group_id)
# Check group is deleted - should only have our initial group left
list_user_credentials = self._check_credentials_exist()
@@ -293,49 +283,41 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_provide_credential_invalid_group(self):
- payload = self._build_credentials_payload(group_name="")
- self._provide_user_credentials(payload, status_code=400)
+ self._create_credentials(group_name="", expected_status=400)
def test_invalid_source_type(self):
- payload = self._build_credentials_payload(source_type="invalid_source_type")
- self._provide_user_credentials(payload, status_code=400)
+ self._create_credentials(source_type="invalid_source_type", expected_status=400)
def test_not_existing_tool(self):
- payload = self._build_credentials_payload(source_id="nonexistent_tool")
- self._provide_user_credentials(payload, status_code=404)
+ self._create_credentials(tool_id="nonexistent_tool", expected_status=404)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_not_existing_tool_version(self):
- payload = self._build_credentials_payload(source_version="nonexistent_tool_version")
- self._provide_user_credentials(payload, status_code=404)
+ self._create_credentials(source_version="nonexistent_tool_version", expected_status=404)
def test_not_existing_service_name(self):
- payload = self._build_credentials_payload(service_name="nonexistent_service")
- self._provide_user_credentials(payload, status_code=404)
+ self._create_credentials(service_name="nonexistent_service", expected_status=404)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_not_existing_service_version(self):
- payload = self._build_credentials_payload(service_version="nonexistent_service_version")
- self._provide_user_credentials(payload, status_code=404)
+ self._create_credentials(service_version="nonexistent_service_version", expected_status=404)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_invalid_credential_name(self):
for key in ["variables", "secrets"]:
payload = self._build_credentials_payload()
payload["service_credential"]["group"][key][0]["name"] = "invalid_name"
- self._provide_user_credentials(payload, status_code=400)
+ self.credentials_populator.post_credentials(payload, expected_status=400)
def test_delete_nonexistent_service_credentials(self):
- response = self._delete("/api/users/current/credentials/f2db41e1fa331b3e")
- self._assert_status_code_is(response, 400)
+ self.credentials_populator.delete_service_credentials("f2db41e1fa331b3e", expected_status=400)
def test_delete_nonexistent_credentials_group(self):
- response = self._delete("/api/users/current/credentials/f2db41e1fa331b3e/groups/f2db41e1fa331b3e")
- self._assert_status_code_is(response, 400)
+ self.credentials_populator.delete_credentials_group("f2db41e1fa331b3e", "f2db41e1fa331b3e", expected_status=400)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_delete_default_credential_group(self):
- created_user_credentials = self._provide_user_credentials()
+ created_user_credentials = self._create_credentials()
# The new API returns a single ServiceCredentialGroupResponse, not a list
group_id = created_user_credentials["id"]
@@ -343,15 +325,13 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
user_credentials_list = self._check_credentials_exist()
user_credentials_id = user_credentials_list[0]["id"]
- response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}")
- self._assert_status_code_is(response, 204)
+ self.credentials_populator.delete_credentials_group(user_credentials_id, group_id)
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_unset_current_group(self):
# First create credentials with a unique group
group_name = random_name()
- payload = self._build_credentials_payload(group_name=group_name)
- self._provide_user_credentials(payload)
+ self._create_credentials(group_name=group_name)
# Set this group as current
user_credentials_list = self._check_credentials_exist()
@@ -362,14 +342,9 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
default_group_id = group["id"]
break
- select_payload = {
- "source_type": "tool",
- "source_id": CREDENTIALS_TEST_TOOL,
- "source_version": "test",
- "service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": default_group_id}],
- }
- response = self._put("/api/users/current/credentials", data=select_payload, json=True)
- self._assert_status_code_is(response, 204)
+ self.credentials_populator.select_current_group(
+ "tool", CREDENTIALS_TEST_TOOL, "test", user_credentials_id, default_group_id
+ )
# Verify it's set as current
list_user_credentials = self._check_credentials_exist()
@@ -382,14 +357,9 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
assert current_group_name == group_name
# Now unset the current group (set to None)
- unset_payload = {
- "source_type": "tool",
- "source_id": CREDENTIALS_TEST_TOOL,
- "source_version": "test",
- "service_credentials": [{"user_credentials_id": user_credentials_id, "current_group_id": None}],
- }
- response = self._put("/api/users/current/credentials", data=unset_payload, json=True)
- self._assert_status_code_is(response, 204)
+ self.credentials_populator.select_current_group(
+ "tool", CREDENTIALS_TEST_TOOL, "test", user_credentials_id, None
+ )
# Verify current group is unset
list_user_credentials = self._check_credentials_exist()
@@ -399,50 +369,42 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
def test_required_credentials_validation(self):
"""Test that required (non-optional) credentials are properly validated."""
# Test missing required variable
- payload = self._build_credentials_payload()
- payload["service_credential"]["group"]["variables"] = [] # Remove required 'server' variable
- self._provide_user_credentials(payload, status_code=400)
+ self._create_credentials(variables=[], expected_status=400)
# Test missing required secret
- payload = self._build_credentials_payload()
- payload["service_credential"]["group"]["secrets"] = [
- {"name": "password", "value": "pass"} # Remove required 'username' secret
- ]
- self._provide_user_credentials(payload, status_code=400)
+ self._create_credentials(secrets=[{"name": "password", "value": "pass"}], expected_status=400)
# Test empty required variable
- payload = self._build_credentials_payload()
- payload["service_credential"]["group"]["variables"] = [{"name": "server", "value": ""}]
- self._provide_user_credentials(payload, status_code=400)
+ self._create_credentials(variables=[{"name": "server", "value": ""}], expected_status=400)
# Test empty required secret
- payload = self._build_credentials_payload()
- payload["service_credential"]["group"]["secrets"] = [
- {"name": "username", "value": ""}, # Empty required secret
- {"name": "password", "value": "pass"},
- ]
- self._provide_user_credentials(payload, status_code=400)
+ self._create_credentials(
+ secrets=[
+ {"name": "username", "value": ""}, # Empty required secret
+ {"name": "password", "value": "pass"},
+ ],
+ expected_status=400,
+ )
# Test that optional credentials can be omitted (password is optional)
- payload = self._build_credentials_payload()
- payload["service_credential"]["group"]["secrets"] = [
- {"name": "username", "value": "user"} # Only required secret, optional 'password' omitted
- ]
- self._provide_user_credentials(payload, status_code=200)
+ self._create_credentials(
+ secrets=[{"name": "username", "value": "user"}], # Only required secret, optional 'password' omitted
+ expected_status=200,
+ )
@skip_without_tool(CREDENTIALS_TEST_TOOL)
def test_vault_integration(self):
test_user_email = "user@vault.test"
with self._different_user(test_user_email):
payload = self._build_credentials_payload()
- self._provide_user_credentials(payload)
+ self.credentials_populator.post_credentials(payload)
credentials_list = self._check_credentials_exist()
assert len(credentials_list) == 1
group = credentials_list[0]["groups"][0]
# Check that secrets are stored in the vault
- for secret in payload["service_credential"]["group"]["secrets"]:
+ for secret in DEFAULT_TOOL_SECRETS:
vault_ref = self._get_vault_ref(payload, group["id"], secret["name"])
expected_value = secret["value"]
self._check_vault_entry_exists(test_user_email, vault_ref, expected_value)
@@ -450,11 +412,10 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
# Delete the credentials group
user_credentials_id = credentials_list[0]["id"]
group_id = group["id"]
- response = self._delete(f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}")
- self._assert_status_code_is(response, 204)
+ self.credentials_populator.delete_credentials_group(user_credentials_id, group_id)
# Check that secrets are removed from the vault
- for secret in payload["service_credential"]["group"]["secrets"]:
+ for secret in DEFAULT_TOOL_SECRETS:
vault_ref = self._get_vault_ref(payload, group["id"], secret["name"])
self._check_vault_entry_exists(test_user_email, vault_ref, should_exist=False)
@@ -462,7 +423,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
def test_list_credentials_with_missing_tool(self):
# Create credentials for the test tool
payload = self._build_credentials_payload()
- self._provide_user_credentials(payload)
+ self.credentials_populator.post_credentials(payload)
# Verify credentials exist normally
credentials_list = self._check_credentials_exist()
@@ -482,9 +443,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
assert self._app.toolbox.get_tool(CREDENTIALS_TEST_TOOL) is None
# Test 1: List credentials with include_definition=True
- response = self._get("/api/users/current/credentials?include_definition=true")
- self._assert_status_code_is(response, 200)
- credentials_with_definition = response.json()
+ credentials_with_definition = self.credentials_populator.list_credentials(include_definition=True)
assert len(credentials_with_definition) == 1
credential = credentials_with_definition[0]
@@ -509,9 +468,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
assert len(credential["groups"]) > 0
# Test 2: List credentials without include_definition
- response = self._get("/api/users/current/credentials")
- self._assert_status_code_is(response, 200)
- credentials_without_definition = response.json()
+ credentials_without_definition = self.credentials_populator.list_credentials()
assert len(credentials_without_definition) == 1
credential_no_def = credentials_without_definition[0]
@@ -525,49 +482,23 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
if tool is not None:
self._app.toolbox.register_tool(tool)
- def _provide_user_credentials(self, payload=None, status_code=200):
- payload = payload or self._build_credentials_payload()
- response = self._post("/api/users/current/credentials", data=payload, json=True)
- self._assert_status_code_is(response, status_code)
- return response.json()
-
- def _build_credentials_payload(
- self,
- source_type: str = "tool",
- source_id: str = CREDENTIALS_TEST_TOOL,
- source_version: str = "test",
- service_name: str = "service1",
- service_version: str = "v1",
- group_name=None,
- ):
- if group_name is None:
- group_name = random_name()
-
- return {
- "source_type": source_type,
- "source_id": source_id,
- "source_version": source_version,
- "service_credential": {
- "name": service_name,
- "version": service_version,
- "group": {
- "name": group_name,
- "variables": [{"name": "server", "value": "http://localhost:8080"}],
- "secrets": [
- {"name": "username", "value": "user"},
- {"name": "password", "value": "pass"},
- ],
- },
- },
- }
-
def _update_credentials(self, user_credentials_id, group_id, payload=None, status_code=200):
payload = payload or self._build_update_credentials_payload()
- response = self._put(
- f"/api/users/current/credentials/{user_credentials_id}/groups/{group_id}", data=payload, json=True
+ return self.credentials_populator.update_credentials_group(
+ user_credentials_id, group_id, payload, expected_status=status_code
)
- self._assert_status_code_is(response, status_code)
- return response.json()
+
+ def _build_credentials_payload(self, **kwargs):
+ kwargs.setdefault("tool_id", CREDENTIALS_TEST_TOOL)
+ kwargs.setdefault("variables", DEFAULT_TOOL_VARIABLES)
+ kwargs.setdefault("secrets", DEFAULT_TOOL_SECRETS)
+ return self.credentials_populator.build_credentials_payload(**kwargs)
+
+ def _create_credentials(self, **kwargs):
+ kwargs.setdefault("tool_id", CREDENTIALS_TEST_TOOL)
+ kwargs.setdefault("variables", DEFAULT_TOOL_VARIABLES)
+ kwargs.setdefault("secrets", DEFAULT_TOOL_SECRETS)
+ return self.credentials_populator.create_credentials(**kwargs)
def _build_update_credentials_payload(
self,
@@ -585,9 +516,7 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
return update_payload
def _check_credentials_exist(self, source_id: str = CREDENTIALS_TEST_TOOL, num_credentials: int = 1):
- response = self._get(f"/api/users/current/credentials?source_type=tool&source_id={source_id}")
- self._assert_status_code_is(response, 200)
- list_user_credentials = response.json()
+ list_user_credentials = self.credentials_populator.list_credentials(source_type="tool", source_id=source_id)
assert len(list_user_credentials) == num_credentials
if num_credentials > 0:
assert list_user_credentials[0]["source_id"] == source_id
@@ -612,6 +541,6 @@ class TestCredentialsApi(integration_util.IntegrationTestCase, integration_util.
or stored_value == ""
), f"Expected vault entry '{vault_ref}' to not exist, but found value '{stored_value}'"
- def _get_vault_ref(self, payload: dict, group_id: str, secret_name: str):
+ def _get_vault_ref(self, payload: dict, group_id: str, secret_name: str) -> str:
decoded_group_id = self._app.security.decode_id(group_id)
return f"{payload['source_type']}|{payload['source_id']}|{payload['service_credential']['name']}|{payload['service_credential']['version']}|{decoded_group_id}|{secret_name}"