Test user config objects before creating them.

Provide detailed error message before proceeding with creation.

client/src/api/configTemplates.ts

@@ -14,6 +14,9 @@ export type VariableValueType = (string | boolean | number) | undefined;
 export type VariableData = { [key: string]: VariableValueType };
 export type SecretData = { [key: string]: string };
 
+export type PluginAspectStatus = components["schemas"]["PluginAspectStatus"];
+export type PluginStatus = components["schemas"]["PluginStatus"];
+
 export interface TemplateSummary {
     description: string | null;
     hidden?: boolean;

client/src/api/schema/schema.ts

@@ -324,6 +324,10 @@ export interface paths {
         /** Create a user-bound file source. */
         post: operations["file_sources__create_instance"];
     };
+    "/api/file_source_instances/test": {
+        /** Test payload for creating user-bound file source. */
+        post: operations["file_sources__test_new_instance_configuration"];
+    };
     "/api/file_source_instances/{user_file_source_id}": {
         /** Get a list of persisted file source instances defined by the requesting user. */
         get: operations["file_sources__instances_get"];
@@ -1269,6 +1273,10 @@ export interface paths {
         /** Create a user-bound object store. */
         post: operations["object_stores__create_instance"];
     };
+    "/api/object_store_instances/test": {
+        /** Test payload for creating user-bound object store. */
+        post: operations["object_stores__test_new_instance_configuration"];
+    };
     "/api/object_store_instances/{user_object_store_id}": {
         /** Get a persisted object store instances owned by the requesting user. */
         get: operations["object_stores__instances_get"];
@@ -10455,12 +10463,28 @@ export interface components {
          * @enum {string}
          */
         PersonalNotificationCategory: "message" | "new_shared_item";
+        /** PluginAspectStatus */
+        PluginAspectStatus: {
+            /** Message */
+            message: string;
+            /**
+             * State
+             * @enum {string}
+             */
+            state: "ok" | "not_ok" | "unknown";
+        };
         /**
          * PluginKind
          * @description Enum to distinguish between different kinds or categories of plugins.
          * @enum {string}
          */
         PluginKind: "rfs" | "drs" | "rdm" | "stock";
+        /** PluginStatus */
+        PluginStatus: {
+            connection?: components["schemas"]["PluginAspectStatus"] | null;
+            template_definition: components["schemas"]["PluginAspectStatus"];
+            template_settings?: components["schemas"]["PluginAspectStatus"] | null;
+        };
         /** Position */
         Position: {
             /** Left */
@@ -15003,6 +15027,34 @@ export interface operations {
             };
         };
     };
+    file_sources__test_new_instance_configuration: {
+        /** Test payload for creating user-bound file source. */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string | null;
+            };
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["CreateInstancePayload"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["PluginStatus"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     file_sources__instances_get: {
         /** Get a list of persisted file source instances defined by the requesting user. */
         parameters: {
@@ -20880,6 +20932,34 @@ export interface operations {
             };
         };
     };
+    object_stores__test_new_instance_configuration: {
+        /** Test payload for creating user-bound object store. */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string | null;
+            };
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["CreateInstancePayload"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["PluginStatus"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     object_stores__instances_get: {
         /** Get a persisted object store instances owned by the requesting user. */
         parameters: {

client/src/components/ConfigTemplates/formUtil.ts

@@ -1,5 +1,6 @@
 import type {
     Instance,
+    PluginStatus,
     SecretData,
     TemplateSecret,
     TemplateSummary,
@@ -243,3 +244,14 @@ export function upgradeFormDataToPayload(template: TemplateSummary, formData: an
     };
     return payload;
 }
+
+export function pluginStatusToErrorMessage(pluginStatus: PluginStatus): string | null {
+    if (pluginStatus.template_definition.state == "not_ok") {
+        return pluginStatus.template_definition.message;
+    } else if (pluginStatus.template_settings?.state == "not_ok") {
+        return pluginStatus.template_settings.message;
+    } else if (pluginStatus.connection?.state == "not_ok") {
+        return pluginStatus.connection.message;
+    }
+    return null;
+}

client/src/components/FileSources/Instances/CreateForm.vue

@@ -3,10 +3,14 @@ import { BAlert } from "bootstrap-vue";
 import { computed, ref } from "vue";
 
 import type { FileSourceTemplateSummary, UserFileSourceModel } from "@/api/fileSources";
-import { createFormDataToPayload, createTemplateForm } from "@/components/ConfigTemplates/formUtil";
+import {
+    createFormDataToPayload,
+    createTemplateForm,
+    pluginStatusToErrorMessage,
+} from "@/components/ConfigTemplates/formUtil";
 import { errorMessageAsString } from "@/utils/simple-error";
 
-import { create } from "./services";
+import { create, test } from "./services";
 
 import InstanceForm from "@/components/ConfigTemplates/InstanceForm.vue";
 
@@ -25,6 +29,12 @@ const inputs = computed(() => {
 
 async function onSubmit(formData: any) {
     const payload = createFormDataToPayload(props.template, formData);
+    const { data: pluginStatus } = await test(payload);
+    const testError = pluginStatusToErrorMessage(pluginStatus);
+    if (testError) {
+        error.value = testError;
+        return;
+    }
     try {
         const { data: fileSource } = await create(payload);
         emit("created", fileSource);

client/src/components/FileSources/Instances/services.ts

@@ -2,6 +2,7 @@ import type { UserFileSourceModel } from "@/api/fileSources";
 import { fetcher } from "@/api/schema/fetcher";
 
 export const create = fetcher.path("/api/file_source_instances").method("post").create();
+export const test = fetcher.path("/api/file_source_instances/test").method("post").create();
 export const update = fetcher.path("/api/file_source_instances/{user_file_source_id}").method("put").create();
 
 export async function hide(instance: UserFileSourceModel) {

client/src/components/ObjectStore/Instances/CreateForm.test.ts

@@ -2,6 +2,7 @@ import { mount } from "@vue/test-utils";
 import flushPromises from "flush-promises";
 import { getLocalVue } from "tests/jest/helpers";
 
+import { PluginStatus } from "@/api/configTemplates";
 import { mockFetcher } from "@/api/schema/__mocks__";
 import type { ObjectStoreTemplateSummary } from "@/components/ObjectStore/Templates/types";
 
@@ -57,6 +58,21 @@ const SIMPLE_TEMPLATE: ObjectStoreTemplateSummary = {
     badges: [],
 };
 
+const FAKE_PLUGIN_STATUS: PluginStatus = {
+    template_definition: {
+        state: "ok",
+        message: "ok",
+    },
+    template_settings: {
+        state: "ok",
+        message: "ok",
+    },
+    connection: {
+        state: "ok",
+        message: "ok",
+    },
+};
+
 describe("CreateForm", () => {
     it("should render a form with admin markdown converted to HTML in help", async () => {
         const wrapper = mount(CreateForm, {
@@ -83,6 +99,7 @@ describe("CreateForm", () => {
             },
             localVue,
         });
+        mockFetcher.path("/api/object_store_instances/test").method("post").mock({ data: FAKE_PLUGIN_STATUS });
         mockFetcher.path("/api/object_store_instances").method("post").mock({ data: FAKE_OBJECT_STORE });
         await flushPromises();
         const nameForElement = wrapper.find("#form-element-_meta_name");
@@ -102,6 +119,7 @@ describe("CreateForm", () => {
             },
             localVue,
         });
+        mockFetcher.path("/api/object_store_instances/test").method("post").mock({ data: FAKE_PLUGIN_STATUS });
         mockFetcher
             .path("/api/object_store_instances")
             .method("post")

client/src/components/ObjectStore/Instances/CreateForm.vue

@@ -2,12 +2,17 @@
 import { BAlert } from "bootstrap-vue";
 import { computed, ref } from "vue";
 
-import { createFormDataToPayload, createTemplateForm, type FormEntry } from "@/components/ConfigTemplates/formUtil";
+import {
+    createFormDataToPayload,
+    createTemplateForm,
+    type FormEntry,
+    pluginStatusToErrorMessage,
+} from "@/components/ConfigTemplates/formUtil";
 import type { UserConcreteObjectStore } from "@/components/ObjectStore/Instances/types";
 import type { ObjectStoreTemplateSummary } from "@/components/ObjectStore/Templates/types";
 import { errorMessageAsString } from "@/utils/simple-error";
 
-import { create } from "./services";
+import { create, test } from "./services";
 
 import InstanceForm from "@/components/ConfigTemplates/InstanceForm.vue";
 
@@ -26,6 +31,12 @@ const inputs = computed<Array<FormEntry>>(() => {
 
 async function onSubmit(formData: any) {
     const payload = createFormDataToPayload(props.template, formData);
+    const { data: pluginStatus } = await test(payload);
+    const testError = pluginStatusToErrorMessage(pluginStatus);
+    if (testError) {
+        error.value = testError;
+        return;
+    }
     try {
         const { data: objectStore } = await create(payload);
         emit("created", objectStore);

client/src/components/ObjectStore/Instances/services.ts

@@ -3,6 +3,7 @@ import { fetcher } from "@/api/schema/fetcher";
 import type { UserConcreteObjectStore } from "./types";
 
 export const create = fetcher.path("/api/object_store_instances").method("post").create();
+export const test = fetcher.path("/api/object_store_instances/test").method("post").create();
 export const update = fetcher.path("/api/object_store_instances/{user_object_store_id}").method("put").create();
 
 export async function hide(instance: UserConcreteObjectStore) {

lib/galaxy/managers/_config_templates.py

@@ -36,6 +36,7 @@ from galaxy.util.config_templates import (
     secrets_as_dict,
     SecretsDict,
     Template,
+    TemplateEnvironmentEntry,
     TemplateEnvironmentSecret,
     TemplateEnvironmentVariable,
     TemplateVariableValueType,
@@ -102,8 +103,14 @@ def recover_secrets(
 def prepare_environment(
     configuration_template: HasConfigEnvironment, vault: Vault, app_config: UsesTemplatesAppConfig
 ) -> EnvironmentDict:
+    return prepare_environment_from_root(configuration_template.template_environment.root, vault, app_config)
+
+
+def prepare_environment_from_root(
+    root: Optional[List[TemplateEnvironmentEntry]], vault: Vault, app_config: UsesTemplatesAppConfig
+):
     environment: EnvironmentDict = {}
-    for environment_entry in configuration_template.template_environment.root:
+    for environment_entry in root or []:
         e_type = environment_entry.type
         e_name = environment_entry.name
         if e_type == "secret":

lib/galaxy/managers/file_source_instances.py

@@ -7,6 +7,7 @@ from typing import (
     Literal,
     Optional,
     Set,
+    Tuple,
     Union,
 )
 from uuid import uuid4
@@ -23,6 +24,7 @@ from galaxy.exceptions import (
 from galaxy.files import (
     FileSourceScore,
     FileSourcesUserContext,
+    ProvidesFileSourcesUserContext,
     UserDefinedFileSources,
 )
 from galaxy.files.plugins import (
@@ -34,6 +36,7 @@ from galaxy.files.sources import (
     file_source_type_is_browsable,
     FilesSourceProperties,
     PluginKind,
+    SupportsBrowsing,
 )
 from galaxy.files.templates import (
     ConfiguredFileSourceTemplates,
@@ -41,6 +44,7 @@ from galaxy.files.templates import (
     FileSourceTemplate,
     FileSourceTemplateSummaries,
     FileSourceTemplateType,
+    template_to_configuration,
 )
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import (
@@ -50,6 +54,11 @@ from galaxy.model import (
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.security.vault import Vault
 from galaxy.util.config_templates import (
+    connection_exception_to_status,
+    PluginAspectStatus,
+    PluginStatus,
+    settings_exception_to_status,
+    status_template_definition,
     TemplateVariableValueType,
     validate_no_extra_secrets_defined,
     validate_no_extra_variables_defined,
@@ -59,6 +68,7 @@ from ._config_templates import (
     CreateInstancePayload,
     ModifyInstancePayload,
     prepare_environment,
+    prepare_environment_from_root,
     purge_template_instance,
     recover_secrets,
     save_template_instance,
@@ -114,6 +124,7 @@ class FileSourceInstancesManager:
     _sa_session: galaxy_scoped_session
     _app_vault: Vault
     _app_config: UserDefinedFileSourcesConfig
+    _resolver: "UserDefinedFileSourcesImpl"
 
     def __init__(
         self,
@@ -121,11 +132,13 @@ class FileSourceInstancesManager:
         sa_session: galaxy_scoped_session,
         vault: Vault,
         app_config: UserDefinedFileSourcesConfig,
+        resolver: "UserDefinedFileSourcesImpl",
     ):
         self._catalog = catalog
         self._sa_session = sa_session
         self._app_vault = vault
         self._app_config = app_config
+        self._resolver = resolver
 
     @property
     def summaries(self) -> FileSourceTemplateSummaries:
@@ -222,6 +235,77 @@ class FileSourceInstancesManager:
         self._save(persisted_file_source)
         return self._to_model(trans, persisted_file_source)
 
+    def plugin_status(self, trans: ProvidesUserContext, payload: CreateInstancePayload) -> PluginStatus:
+        template = self._catalog.find_template(payload)
+        template_definition_status = status_template_definition(template)
+        status_kwds = {"template_definition": template_definition_status}
+        if template_definition_status.is_not_ok:
+            return PluginStatus(**status_kwds)
+        assert template
+        configuration, template_settings_status = self._template_settings_status(trans, payload, template)
+        status_kwds["template_settings"] = template_settings_status
+        if template_settings_status.is_not_ok:
+            return PluginStatus(**status_kwds)
+        assert configuration
+        file_source, connection_status = self._connection_status(trans, payload, configuration)
+        status_kwds["connection"] = connection_status
+        if connection_status.is_not_ok:
+            return PluginStatus(**status_kwds)
+        assert file_source
+        # Lets circle back to this - we need to add an entry point to the file source plugins
+        # to test if things are writable. We could ping remote APIs or do something like os.access('/path/to/folder', os.W_OK)
+        # locally.
+        return PluginStatus(**status_kwds)
+
+    def _template_settings_status(
+        self,
+        trans: ProvidesUserContext,
+        payload: CreateInstancePayload,
+        template: FileSourceTemplate,
+    ) -> Tuple[Optional[FileSourceConfiguration], PluginAspectStatus]:
+        secrets = payload.secrets
+        variables = payload.variables
+        environment = prepare_environment_from_root(template.environment, self._app_vault, self._app_config)
+        user_details = trans.user.config_template_details()
+
+        configuration = None
+        exception = None
+        try:
+            configuration = template_to_configuration(
+                template,
+                variables=variables,
+                secrets=secrets,
+                user_details=user_details,
+                environment=environment,
+            )
+        except Exception as e:
+            exception = e
+        return configuration, settings_exception_to_status(exception)
+
+    def _connection_status(
+        self, trans: ProvidesUserContext, payload: CreateInstancePayload, configuration: FileSourceConfiguration
+    ) -> Tuple[Optional[BaseFilesSource], PluginAspectStatus]:
+        file_source = None
+        exception = None
+        try:
+            file_source_properties = configuration_to_file_source_properties(
+                configuration,
+                label=payload.name,
+                doc=payload.description,
+                id=uuid4().hex,
+            )
+            file_source = self._resolver._file_source(file_source_properties)
+            if hasattr(file_source, "list"):
+                assert file_source
+                # if we can list the root, do that and assume there is
+                # a connection problem if we cannot
+                browsable_file_source = cast(SupportsBrowsing, file_source)
+                user_context = ProvidesFileSourcesUserContext(trans)
+                browsable_file_source.list("/", recursive=False, user_context=user_context)
+        except Exception as e:
+            exception = e
+        return file_source, connection_exception_to_status("file source", exception)
+
     def _index_filter(self, id: Union[str, int]):
         index_by = self._app_config.user_config_templates_index_by
         index_filter: Any
@@ -344,21 +428,12 @@ class UserDefinedFileSourcesImpl(UserDefinedFileSources):
         file_source_configuration: FileSourceConfiguration = user_file_source.file_source_configuration(
             secrets=secrets, environment=environment, templates=templates
         )
-        file_source_properties = cast(FilesSourceProperties, file_source_configuration.model_dump())
-        file_source_properties["label"] = user_file_source.name
-        file_source_properties["doc"] = user_file_source.description
-        file_source_properties["id"] = f"{user_file_source.uuid}"
-        file_source_properties["scheme"] = USER_FILE_SOURCES_SCHEME
-        # Moved this into templates - plugins should just define this and decide what
-        # that looks like. aws public buckets are clearly not writable, private buckets
-        # maybe should give users the option, etc..
-        # file_source_properties["writable"] = True
-
-        # We did templating with Jinja - disable Galaxy's Cheetah templating for
-        # these plugins. I can't imagine a use case for that and I would hate to templating
-        # languages having odd interactions.
-        file_source_properties["disable_templating"] = True
-        return file_source_properties
+        return configuration_to_file_source_properties(
+            file_source_configuration,
+            label=user_file_source.name,
+            doc=user_file_source.description,
+            id=f"{user_file_source.uuid}",
+        )
 
     def validate_uri_root(self, uri: str, user_context: FileSourcesUserContext) -> None:
         user_object_store = self._user_file_source(uri)
@@ -424,6 +499,31 @@ class UserDefinedFileSourcesImpl(UserDefinedFileSources):
         return as_dicts
 
 
+# Turn the validated Pydantic thing describe what is possible to configure to the
+# raw TypedDict consumed by the actual galaxy.files plugins.
+def configuration_to_file_source_properties(
+    file_source_configuration: FileSourceConfiguration,
+    label: str,
+    doc: Optional[str],
+    id: str,
+) -> FilesSourceProperties:
+    file_source_properties = cast(FilesSourceProperties, file_source_configuration.model_dump())
+    file_source_properties["label"] = label
+    file_source_properties["doc"] = doc
+    file_source_properties["id"] = id
+    file_source_properties["scheme"] = USER_FILE_SOURCES_SCHEME
+    # Moved this into templates - plugins should just define this and decide what
+    # that looks like. aws public buckets are clearly not writable, private buckets
+    # maybe should give users the option, etc..
+    # file_source_properties["writable"] = True
+
+    # We did templating with Jinja - disable Galaxy's Cheetah templating for
+    # these plugins. I can't imagine a use case for that and I would hate to templating
+    # languages having odd interactions.
+    file_source_properties["disable_templating"] = True
+    return file_source_properties
+
+
 __all__ = (
     "CreateInstancePayload",
     "FileSourceInstancesManager",

lib/galaxy/managers/object_store_instances.py

@@ -13,6 +13,7 @@ from typing import (
     Dict,
     List,
     Optional,
+    Tuple,
     Union,
 )
 from uuid import uuid4
@@ -25,7 +26,9 @@ from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import UserObjectStore
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.objectstore import (
+    BaseObjectStore,
     BaseUserObjectStoreResolver,
+    build_test_object_store_from_user_config,
     ConcreteObjectStoreModel,
     QuotaModel,
     UserObjectStoresAppConfig,
@@ -37,9 +40,15 @@ from galaxy.objectstore.templates import (
     ObjectStoreTemplate,
     ObjectStoreTemplateSummaries,
     ObjectStoreTemplateType,
+    template_to_configuration,
 )
 from galaxy.security.vault import Vault
 from galaxy.util.config_templates import (
+    connection_exception_to_status,
+    PluginAspectStatus,
+    PluginStatus,
+    settings_exception_to_status,
+    status_template_definition,
     TemplateVariableValueType,
     validate_no_extra_secrets_defined,
     validate_no_extra_variables_defined,
@@ -48,6 +57,7 @@ from ._config_templates import (
     CreateInstancePayload,
     ModifyInstancePayload,
     prepare_environment,
+    prepare_environment_from_root,
     purge_template_instance,
     recover_secrets,
     save_template_instance,
@@ -209,6 +219,64 @@ class ObjectStoreInstancesManager:
             raise ItemOwnershipException()
         return user_object_store
 
+    def plugin_status(self, trans: ProvidesUserContext, payload: CreateInstancePayload) -> PluginStatus:
+        template = self._catalog.find_template(payload)
+        template_definition_status = status_template_definition(template)
+        status_kwds = {"template_definition": template_definition_status}
+        if template_definition_status.is_not_ok:
+            return PluginStatus(**status_kwds)
+        assert template
+        configuration, template_settings_status = self._template_settings_status(trans, payload, template)
+        status_kwds["template_settings"] = template_settings_status
+        if template_settings_status.is_not_ok:
+            return PluginStatus(**status_kwds)
+        assert configuration
+        object_store, connection_status = self._connection_status(trans, payload, configuration)
+        status_kwds["connection"] = connection_status
+        if connection_status.is_not_ok:
+            return PluginStatus(**status_kwds)
+        assert object_store
+        # Lets circle back to this - we need to add an entry point to the file source plugins
+        # to test if things are writable. We could ping remote APIs or do something like os.access('/path/to/folder', os.W_OK)
+        # locally.
+        return PluginStatus(**status_kwds)
+
+    def _template_settings_status(
+        self,
+        trans: ProvidesUserContext,
+        payload: CreateInstancePayload,
+        template: ObjectStoreTemplate,
+    ) -> Tuple[Optional[ObjectStoreConfiguration], PluginAspectStatus]:
+        secrets = payload.secrets
+        variables = payload.variables
+        environment = prepare_environment_from_root(template.environment, self._app_vault, self._app_config)
+        user_details = trans.user.config_template_details()
+
+        configuration = None
+        exception = None
+        try:
+            configuration = template_to_configuration(
+                template,
+                variables=variables,
+                secrets=secrets,
+                user_details=user_details,
+                environment=environment,
+            )
+        except Exception as e:
+            exception = e
+        return configuration, settings_exception_to_status(exception)
+
+    def _connection_status(
+        self, trans: ProvidesUserContext, payload: CreateInstancePayload, configuration: ObjectStoreConfiguration
+    ) -> Tuple[Optional[BaseObjectStore], PluginAspectStatus]:
+        object_store = None
+        exception = None
+        try:
+            object_store = build_test_object_store_from_user_config(trans.app.config, configuration)
+        except Exception as e:
+            exception = e
+        return object_store, connection_exception_to_status("storage location", exception)
+
     def _index_filter(self, id: Union[str, int]):
         index_by = self._app_config.user_config_templates_index_by
         index_filter: Any

lib/galaxy/model/__init__.py

@@ -1206,6 +1206,16 @@ ON CONFLICT
         environment = User.user_template_environment(user)
         return Template(in_string).safe_substitute(environment)
 
+    # above templating is for Cheetah in tools where we discouraged user details from being exposed.
+    # the following templating if user details in Jinja for object stores and file sources where user
+    # details are critical and documented.
+    def config_template_details(self) -> Dict[str, Any]:
+        return {
+            "username": self.username,
+            "email": self.email,
+            "id": self.id,
+        }
+
     def is_active(self):
         return self.active
 
@@ -11060,12 +11070,7 @@ class UserObjectStore(Base, HasConfigTemplate):
     ) -> ObjectStoreConfiguration:
         if templates is None:
             templates = [self.template]
-        user = self.user
-        user_details = {
-            "username": user.username,
-            "email": user.email,
-            "id": user.id,
-        }
+        user_details = self.user.config_template_details()
         variables: CONFIGURATION_TEMPLATE_CONFIGURATION_VARIABLES_TYPE = self.template_variables or {}
         first_exception = None
         for template in templates:
@@ -11133,12 +11138,7 @@ class UserFileSource(Base, HasConfigTemplate):
     ) -> FileSourceConfiguration:
         if templates is None:
             templates = [self.template]
-        user = self.user
-        user_details = {
-            "username": user.username,
-            "email": user.email,
-            "id": user.id,
-        }
+        user_details = self.user.config_template_details()
         variables: CONFIGURATION_TEMPLATE_CONFIGURATION_VARIABLES_TYPE = self.template_variables or {}
         first_exception = None
         for template in templates:

lib/galaxy/objectstore/__init__.py

@@ -23,6 +23,7 @@ from typing import (
     Type,
     TYPE_CHECKING,
 )
+from uuid import uuid4
 
 import yaml
 from pydantic import BaseModel
@@ -1047,6 +1048,17 @@ class NestedObjectStore(BaseObjectStore):
             return default
 
 
+def user_object_store_configuration_to_config_dict(object_store_config: ObjectStoreConfiguration, id) -> Dict[str, Any]:
+    # convert a pydantic model describing a user object store into a config dict ready to be
+    # slotted into a distributed job runner or stand alone.
+    dynamic_object_store_as_dict = object_store_config.model_dump()
+    dynamic_object_store_as_dict["id"] = id
+    dynamic_object_store_as_dict["weight"] = 0
+    # these are all forward facing object stores...
+    dynamic_object_store_as_dict["store_by"] = "uuid"
+    return dynamic_object_store_as_dict
+
+
 class DistributedObjectStore(NestedObjectStore):
     """
     ObjectStore that defers to a list of backends.
@@ -1200,11 +1212,9 @@ class DistributedObjectStore(NestedObjectStore):
                 object_store_config = self.user_object_store_resolver.resolve_object_store_uri_config(
                     user_object_store_uri
                 )
-                dynamic_object_store_as_dict = object_store_config.model_dump()
-                dynamic_object_store_as_dict["id"] = user_object_store_uri
-                dynamic_object_store_as_dict["weight"] = 0
-                # these are all forward facing object stores...
-                dynamic_object_store_as_dict["store_by"] = "uuid"
+                dynamic_object_store_as_dict = user_object_store_configuration_to_config_dict(
+                    object_store_config, user_object_store_uri
+                )
                 backends.append(dynamic_object_store_as_dict)
 
         as_dict["backends"] = backends
@@ -1561,6 +1571,21 @@ def type_to_object_store_class(
     return objectstore_class, objectstore_constructor_kwds
 
 
+def build_test_object_store_from_user_config(
+    config,
+    object_store_config: ObjectStoreConfiguration,
+):
+    # check an object store configuration by building a standalone object store
+    # from a supplied user object store configuration.
+    config_dict = user_object_store_configuration_to_config_dict(object_store_config, uuid4().hex)
+    object_store = build_object_store_from_config(
+        config,
+        config_dict=config_dict,
+        disable_process_management=True,
+    )
+    return object_store
+
+
 def build_object_store_from_config(
     config,
     fsmon=False,

lib/galaxy/util/config_templates.py

@@ -20,6 +20,7 @@ from pydantic import (
     BaseModel,
     ConfigDict,
     RootModel,
+    ValidationError,
 )
 from typing_extensions import (
     Literal,
@@ -27,9 +28,15 @@ from typing_extensions import (
 )
 
 try:
+    from jinja2 import (
+        StrictUndefined,
+        UndefinedError,
+    )
     from jinja2.nativetypes import NativeEnvironment
 except ImportError:
     NativeEnvironment = None  # type:ignore[assignment, misc, unused-ignore]
+    StrictUndefined = None  # type:ignore[assignment, misc, unused-ignore]
+    UndefinedError = None  # type:ignore[assignment, misc, unused-ignore]
 
 from galaxy.exceptions import (
     ObjectNotFound,
@@ -122,6 +129,7 @@ def _environment(template_start: str, template_end: str) -> NativeEnvironment:
     env = NativeEnvironment(
         variable_start_string=template_start,
         variable_end_string=template_end,
+        undefined=StrictUndefined,
     )
     env.filters["ensure_path_component"] = _ensure_path_component
     env.filters["asbool"] = asbool
@@ -415,3 +423,58 @@ def acts_as_simple_path_component(value: str):
     if should_be_value != value:
         return False
     return True
+
+
+class PluginAspectStatus(StrictModel):
+    state: Literal["ok", "not_ok", "unknown"]
+    message: str
+
+    @property
+    def is_not_ok(self):
+        return self.state == "not_ok"
+
+
+class PluginStatus(StrictModel):
+    template_definition: PluginAspectStatus
+    template_settings: Optional[PluginAspectStatus] = None
+    connection: Optional[PluginAspectStatus] = None
+    # I would love to disambiguate connection vs auth errors but would
+    # attempting to do that cause confusion. Maybe not if the user interface
+    # skipped presenting the one that couldn't be disambiguated for that
+    # particular plugin?
+
+    # TODO: Fill in writable checks.
+    # writable: Optional[PluginAspectStatus] = None
+
+
+def status_template_definition(template: Optional[Template]) -> PluginAspectStatus:
+    # if we found a template in the catalog, it was validated at load time. Reflect
+    # this as a PluginAspectStatus
+    if template:
+        return PluginAspectStatus(state="ok", message="Template definition found and validates against schema")
+    else:
+        return PluginAspectStatus(state="not_ok", message="Template not found or not loaded")
+
+
+def settings_exception_to_status(exception: Optional[Exception]) -> PluginAspectStatus:
+    if exception is None:
+        status = PluginAspectStatus(state="ok", message="Valid configuration resulted from supplied settings")
+    elif isinstance(exception, UndefinedError):
+        message = f"Problem with template definition causing invalid settings resolution, please contact admin to correct template: {exception}"
+        status = PluginAspectStatus(state="not_ok", message=message)
+    elif isinstance(exception, ValidationError):
+        message = f"Problem with template definition causing invalid configuration, template expanded without error but resulting configuration is invalid. please contact admin to correct template: {exception}"
+        status = PluginAspectStatus(state="not_ok", message=message)
+    else:
+        message = f"Unknown problem with resolving configuration from supplied settings: {exception}"
+        status = PluginAspectStatus(state="not_ok", message=message)
+    return status
+
+
+def connection_exception_to_status(what: str, exception: Optional[Exception]) -> PluginAspectStatus:
+    if exception is None:
+        connection_status = PluginAspectStatus(state="ok", message="Valid connection resulted from supplied settings")
+    else:
+        message = f"Failed to connect to a {what} with supplied settings: {exception}"
+        connection_status = PluginAspectStatus(state="not_ok", message=message)
+    return connection_status

lib/galaxy/webapps/galaxy/api/file_sources.py

@@ -16,6 +16,7 @@ from galaxy.managers.file_source_instances import (
     ModifyInstancePayload,
     UserFileSourceModel,
 )
+from galaxy.util.config_templates import PluginStatus
 from . import (
     depends,
     DependsOnTrans,
@@ -60,6 +61,18 @@ class FastAPIFileSources:
     ) -> UserFileSourceModel:
         return self.file_source_instances_manager.create_instance(trans, payload)
 
+    @router.post(
+        "/api/file_source_instances/test",
+        summary="Test payload for creating user-bound file source.",
+        operation_id="file_sources__test_new_instance_configuration",
+    )
+    def test_instance_configuration(
+        self,
+        trans: ProvidesUserContext = DependsOnTrans,
+        payload: CreateInstancePayload = Body(...),
+    ) -> PluginStatus:
+        return self.file_source_instances_manager.plugin_status(trans, payload)
+
     @router.get(
         "/api/file_source_instances",
         summary="Get a list of persisted file source instances defined by the requesting user.",

lib/galaxy/webapps/galaxy/api/object_store.py

@@ -32,6 +32,7 @@ from galaxy.objectstore import (
     ConcreteObjectStoreModel,
 )
 from galaxy.objectstore.templates import ObjectStoreTemplateSummaries
+from galaxy.util.config_templates import PluginStatus
 from . import (
     depends,
     DependsOnTrans,
@@ -98,6 +99,18 @@ class FastAPIObjectStore:
     ) -> UserConcreteObjectStoreModel:
         return self.object_store_instance_manager.create_instance(trans, payload)
 
+    @router.post(
+        "/api/object_store_instances/test",
+        summary="Test payload for creating user-bound object store.",
+        operation_id="object_stores__test_new_instance_configuration",
+    )
+    def test_instance_configuration(
+        self,
+        trans: ProvidesUserContext = DependsOnTrans,
+        payload: CreateInstancePayload = Body(...),
+    ) -> PluginStatus:
+        return self.object_store_instance_manager.plugin_status(trans, payload)
+
     @router.get(
         "/api/object_store_instances",
         summary="Get a list of persisted object store instances defined by the requesting user.",

test/integration_selenium/test_user_object_store_generic_s3.py

@@ -1,40 +0,0 @@
-from galaxy.selenium.navigates_galaxy import (
-    ConfigTemplateParameter,
-    ObjectStoreInstance,
-)
-from ._base_user_object_stores import BaseUserObjectStoreSeleniumIntegration
-from .framework import (
-    managed_history,
-    selenium_test,
-)
-from .test_user_object_store_generic_s3_legacy import (
-    GALAXY_TEST_PLAY_MINIO_BUCKET,
-    GALAXY_TEST_PLAY_MINIO_KEY,
-    GALAXY_TEST_PLAY_MINIO_SECRET,
-    PLAY_HOST,
-    PLAY_PORT,
-)
-
-PLAY_ENDPOINT_URL = f"https://{PLAY_HOST}:{PLAY_PORT}/"
-
-
-class TestUserObjectStoreGenericS3(BaseUserObjectStoreSeleniumIntegration):
-    example_filename = "production_generic_s3.yml"
-
-    @managed_history
-    @selenium_test
-    def test_create_and_use(self):
-        random_name = self._get_random_name(prefix="generic s3 object store using play.min.io")
-        instance = ObjectStoreInstance(
-            template_id="generic_s3",
-            name=random_name,
-            description="automated test for legacy generic s3 object store against play.min.io",
-            parameters=[
-                ConfigTemplateParameter("string", "access_key", GALAXY_TEST_PLAY_MINIO_KEY),
-                ConfigTemplateParameter("string", "secret_key", GALAXY_TEST_PLAY_MINIO_SECRET),
-                ConfigTemplateParameter("string", "bucket", GALAXY_TEST_PLAY_MINIO_BUCKET),
-                ConfigTemplateParameter("string", "endpoint_url", PLAY_ENDPOINT_URL),
-            ],
-        )
-        object_store_id = self.create_object_store_template(instance)
-        assert object_store_id

test/integration_selenium/test_user_object_store_generic_s3_legacy.py

@@ -1,38 +0,0 @@
-from galaxy.selenium.navigates_galaxy import (
-    ConfigTemplateParameter,
-    ObjectStoreInstance,
-)
-from ._base_user_object_stores import BaseUserObjectStoreSeleniumIntegration
-from .framework import (
-    managed_history,
-    selenium_test,
-)
-
-PLAY_HOST = "play.min.io"
-PLAY_PORT = "9000"
-GALAXY_TEST_PLAY_MINIO_KEY = "LEHFJDNqSA4xcJmmezU7"
-GALAXY_TEST_PLAY_MINIO_SECRET = "E3ycZrp2nV8WscER8HqgsPPL2aFc2uuTbRchelcX"
-GALAXY_TEST_PLAY_MINIO_BUCKET = "gxtest1"
-
-
-class TestUserObjectStoreGenericS3Legacy(BaseUserObjectStoreSeleniumIntegration):
-    example_filename = "production_generic_s3_legacy.yml"
-
-    @managed_history
-    @selenium_test
-    def test_create_and_use(self):
-        random_name = self._get_random_name(prefix="generic s3 object store using play.min.io")
-        instance = ObjectStoreInstance(
-            template_id="generic_s3_legacy",
-            name=random_name,
-            description="automated test for legacy generic s3 object store against play.min.io",
-            parameters=[
-                ConfigTemplateParameter("string", "access_key", GALAXY_TEST_PLAY_MINIO_KEY),
-                ConfigTemplateParameter("string", "secret_key", GALAXY_TEST_PLAY_MINIO_SECRET),
-                ConfigTemplateParameter("string", "bucket", GALAXY_TEST_PLAY_MINIO_BUCKET),
-                ConfigTemplateParameter("string", "host", PLAY_HOST),
-                ConfigTemplateParameter("integer", "port", PLAY_PORT),
-            ],
-        )
-        object_store_id = self.create_object_store_template(instance)
-        assert object_store_id

test/unit/app/managers/test_user_file_sources.py

@@ -45,6 +45,35 @@ def home_directory_template(tmp_path):
     }
 
 
+def invalid_home_directory_template(tmp_path):
+    # create a jinja runtime problem - so the template loads but the expansion fails.
+    return {
+        "id": "invalid_home_directory",
+        "name": "Home Directory",
+        "description": "Your Home Directory on this System",
+        "configuration": {
+            "type": "posix",
+            "root": str(tmp_path / "{{ username }}"),  # should be user.username
+            "writable": True,
+        },
+    }
+
+
+def invalid_home_directory_template_type_error(tmp_path):
+    # create a runtime problem - so the template loads but the expansion fails
+    # because pydantic error related to typing
+    return {
+        "id": "invalid_home_directory",
+        "name": "Home Directory",
+        "description": "Your Home Directory on this System",
+        "configuration": {
+            "type": "posix",
+            "root": str(tmp_path / "{{ user.username }}"),
+            "writable": "{{ username }}",
+        },
+    }
+
+
 def secret_directory_template(tmp_path):
     return {
         "id": "admin_secret_directory",
@@ -329,6 +358,86 @@ class TestFileSourcesTestCase(BaseTestCase):
         self.manager.modify_instance(self.trans, user_file_source.uuid, update)
         assert user_vault.read_secret(config_secret_key) == "newvalue"
 
+    def test_status_valid(self, tmp_path):
+        self.init_user_in_database()
+        self._init_managers(tmp_path)
+        (tmp_path / self.trans.user.username).mkdir()
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="home_directory",
+            template_version=0,
+            variables={},
+            secrets={},
+        )
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert status.connection
+        assert not status.connection.is_not_ok
+        assert not status.template_definition.is_not_ok
+        assert status.template_settings
+        assert not status.template_settings.is_not_ok
+
+    def test_status_invalid_connection(self, tmp_path):
+        self.init_user_in_database()
+        self._init_managers(tmp_path)
+        # We don't make the directory like above so it doesn't exist
+        # (tmp_path / self.trans.user.username).mkdir()
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="home_directory",
+            template_version=0,
+            variables={},
+            secrets={},
+        )
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert not status.template_definition.is_not_ok
+        assert status.template_settings
+        assert not status.template_settings.is_not_ok
+        # Language is weird with the local disk stuff but the "connection"
+        # is invalid. Do we search for better language or loosen the framework
+        # structure in someway and push these specific checks into the plugins?
+        assert status.connection
+        assert status.connection.is_not_ok
+
+    def test_status_invalid_settings_undefined_variable(self, tmp_path):
+        self.init_user_in_database()
+        self._init_managers(tmp_path, config_dict=invalid_home_directory_template(tmp_path))
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="invalid_home_directory",
+            template_version=0,
+            variables={},
+            secrets={},
+        )
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert not status.template_definition.is_not_ok
+        assert status.template_settings
+        assert status.template_settings.is_not_ok
+        assert (
+            "Problem with template definition causing invalid settings resolution" in status.template_settings.message
+        )
+        assert status.connection is None
+
+    def test_status_invalid_settings_configuration_validation(self, tmp_path):
+        self.init_user_in_database()
+        self._init_managers(tmp_path, config_dict=invalid_home_directory_template_type_error(tmp_path))
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="invalid_home_directory",
+            template_version=0,
+            variables={},
+            secrets={},
+        )
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert not status.template_definition.is_not_ok
+        assert status.template_settings
+        assert status.template_settings.is_not_ok
+        assert "Input should be a valid boolean" in status.template_settings.message
+        assert status.connection is None
+
     def _init_and_create_simple(self, tmp_path) -> UserFileSourceModel:
         self._init_managers(tmp_path)
         user_file_source = self._create_user_file_source()

test/unit/app/managers/test_user_object_stores.py

@@ -55,6 +55,23 @@ def simple_vault_template(tmp_path):
     }
 
 
+def simple_vault_template_with_undefined_jinja_problem(tmp_path):
+    return {
+        "id": "simple_vault",
+        "name": "Simple Vault",
+        "description": "This is a simple description",
+        "configuration": {
+            "type": "disk",
+            "files_dir": str(tmp_path / "{{ username }}/{{ secrets.sec1 }}"),  # should be user.username
+        },
+        "secrets": {
+            "sec1": {
+                "help": "This is some simple help.",
+            },
+        },
+    }
+
+
 SIMPLE_VAULT_CREATE_PAYLOAD = CreateInstancePayload(
     name=SIMPLE_FILE_SOURCE_NAME,
     description=SIMPLE_FILE_SOURCE_DESCRIPTION,
@@ -294,6 +311,44 @@ class TestUserObjectStoreTestCase(BaseTestCase):
 
         self._assert_modify_throws_exception(user_object_store, upgrade_to_1, RequestParameterMissingException)
 
+    def test_status_valid(self, tmp_path):
+        self.init_user_in_database()
+        self._init_managers(tmp_path, simple_vault_template(tmp_path))
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="simple_vault",
+            template_version=0,
+            variables={},
+            secrets={"sec1": "foosec"},
+        )
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert status.connection
+        assert not status.connection.is_not_ok
+        assert not status.template_definition.is_not_ok
+        assert status.template_settings
+        assert not status.template_settings.is_not_ok
+
+    def test_status_invalid_settings_undefined_variable(self, tmp_path):
+        self.init_user_in_database()
+        self._init_managers(tmp_path, config_dict=simple_vault_template_with_undefined_jinja_problem(tmp_path))
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="simple_vault",
+            template_version=0,
+            variables={},
+            secrets={},
+        )
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert not status.template_definition.is_not_ok
+        assert status.template_settings
+        assert status.template_settings.is_not_ok
+        assert (
+            "Problem with template definition causing invalid settings resolution" in status.template_settings.message
+        )
+        assert status.connection is None
+
     def _init_upgrade_test_case(self, tmp_path) -> UserConcreteObjectStoreModel:
         example_yaml_str = UPGRADE_EXAMPLE
         example_yaml_str.replace("/data", str(tmp_path))