Implement storage access endpoint list. see https://github.com/doe-ir…

app/demo_adapter.py

@@ -115,6 +115,7 @@ def __init__(self):
         self.user_allocations = []
         self.facility = {}
         self.locations = {}  # resource_id -> list[StorageInstance templates]
+        self.access_endpoints = {}  # resource_id -> list[AccessEndpoint]
         self.sites = []
         self._init_state()
 
@@ -399,6 +400,74 @@ def _init_state(self):
         # Login nodes: same filesystem layout as CFS — outside-of-job semantics for everything.
         self.locations[login.id] = self.locations[cfs.id]
 
+        globus_cfs_id = demo_uuid("endpoint", "globus-cfs")
+        globus_hpss_id = demo_uuid("endpoint", "globus-hpss")
+
+        self.access_endpoints[cfs.id] = [
+            storage_models.AccessEndpoint(
+                id="globus-cfs-demo",
+                resource_id=cfs.id,
+                protocol=storage_models.AccessProtocol.globus,
+                display_name="Demo CFS Globus",
+                endpoint_id=globus_cfs_id,
+                uri=f"globus://{globus_cfs_id}/",
+                root_path="/",
+                auth_type="globus",
+                capabilities=[
+                    storage_models.AccessCapability.list,
+                    storage_models.AccessCapability.read,
+                    storage_models.AccessCapability.write,
+                    storage_models.AccessCapability.transfer,
+                ],
+            ),
+            storage_models.AccessEndpoint(
+                id="xrootd-cfs-demo",
+                resource_id=cfs.id,
+                protocol=storage_models.AccessProtocol.xrootd,
+                display_name="Demo CFS XRootD",
+                endpoint="root://cfs.demo.example/",
+                auth_type="x509",
+                capabilities=[
+                    storage_models.AccessCapability.read,
+                    storage_models.AccessCapability.streaming,
+                ],
+            ),
+            storage_models.AccessEndpoint(
+                id="s3-cfs-demo",
+                resource_id=cfs.id,
+                protocol=storage_models.AccessProtocol.s3,
+                display_name="Demo CFS S3",
+                bucket="demo-cfs",
+                region="us-east-1",
+                endpoint_url="https://s3.demo.example",
+                auth_type="aws_s3",
+                capabilities=[
+                    storage_models.AccessCapability.list,
+                    storage_models.AccessCapability.read,
+                    storage_models.AccessCapability.write,
+                ],
+            ),
+        ]
+
+        self.access_endpoints[hpss.id] = [
+            storage_models.AccessEndpoint(
+                id="globus-hpss-demo",
+                resource_id=hpss.id,
+                protocol=storage_models.AccessProtocol.globus,
+                display_name="Demo HPSS Globus",
+                endpoint_id=globus_hpss_id,
+                uri=f"globus://{globus_hpss_id}/",
+                root_path="/home",
+                auth_type="globus",
+                capabilities=[
+                    storage_models.AccessCapability.list,
+                    storage_models.AccessCapability.read,
+                    storage_models.AccessCapability.write,
+                    storage_models.AccessCapability.transfer,
+                ],
+            ),
+        ]
+
         # Populate site resource_ids based on which resources are at each site
         site1.resource_ids = [r.id for r in self.resources if r.site_id == site1.id]
         site2.resource_ids = [r.id for r in self.resources if r.site_id == site2.id]
@@ -870,6 +939,20 @@ async def get_locations(
                 ))
         return result
 
+    async def get_access_endpoints(
+        self,
+        resource: status_models.Resource,
+        user: User,
+        protocol: storage_models.AccessProtocol | None,
+        endpoint_id: str | None,
+    ) -> list[storage_models.AccessEndpoint]:
+        endpoints = self.access_endpoints.get(resource.id, [])
+        if protocol:
+            endpoints = [e for e in endpoints if e.protocol == protocol]
+        if endpoint_id:
+            endpoints = [e for e in endpoints if e.id == endpoint_id]
+        return endpoints
+
     def validate_path(self, path: str, allow_symlinks: bool = True) -> str:
         """Validate that the given path is within the sandbox base directory and optionally check for symlinks."""
         basedir = PathSandbox.get_base_temp_dir()

app/routers/storage/facility_adapter.py

@@ -36,3 +36,19 @@ async def get_locations(
           - read: no filtering (all accessible paths)
         """
         pass
+
+    @abstractmethod
+    async def get_access_endpoints(
+        self,
+        resource: status_models.Resource,
+        user: User,
+        protocol: storage_models.AccessProtocol | None,
+        endpoint_id: str | None,
+    ) -> list[storage_models.AccessEndpoint]:
+        """
+        Return the list of data access endpoints for the given storage resource.
+        Each entry describes a protocol (Globus, XRootD, S3, ...) and the connection
+        details needed to use it.
+        Results are optionally filtered by protocol and/or endpoint ID.
+        """
+        pass

app/routers/storage/models.py

@@ -1,6 +1,7 @@
 """Models for storage location and mount API endpoints."""
 from enum import Enum
 from pydantic import Field, BaseModel
+from typing import Optional
 
 
 class LogicalName(str, Enum):
@@ -77,3 +78,42 @@ class StorageInstance(BaseModel):
         ...,
         description="Access permissions through the queried resource context",
     )
+
+
+class AccessProtocol(str, Enum):
+    """Supported data access protocols."""
+    globus = "globus"
+    xrootd = "xrootd"
+    s3 = "s3"
+
+
+class AccessCapability(str, Enum):
+    """Data operations supported by an access endpoint."""
+    list = "list"
+    read = "read"
+    write = "write"
+    transfer = "transfer"
+    streaming = "streaming"
+
+
+class AccessEndpoint(BaseModel):
+    """
+    A single data access endpoint for a storage resource.
+    Protocol-specific connection fields are present only for the relevant protocol.
+    """
+    id: str = Field(..., description="Unique identifier for this access endpoint", example="globus-cfs-demo")
+    resource_id: str = Field(..., description="ID of the storage resource this endpoint belongs to")
+    protocol: AccessProtocol = Field(..., description="Data access protocol")
+    display_name: Optional[str] = Field(default=None, description="Human-readable name for this endpoint", example="Demo CFS Globus")
+    auth_type: str = Field(..., description="Authentication mechanism required to use this endpoint", example="globus")
+    capabilities: list[AccessCapability] = Field(..., description="Supported data operations")
+    # Globus-specific
+    endpoint_id: Optional[str] = Field(default=None, description="Globus endpoint UUID (Globus only)", example="5e0cdbd2-3f1a-4e57-beed-b95scbb83b7c")
+    uri: Optional[str] = Field(default=None, description="Full Globus URI (Globus only)", example="globus://5e0cdbd2-3f1a-4e57-beed-b95scbb83b7c/")
+    root_path: Optional[str] = Field(default=None, description="Root path within the endpoint (Globus only)", example="/")
+    # XRootD-specific
+    endpoint: Optional[str] = Field(default=None, description="XRootD server address (XRootD only)", example="root://cfs.demo.example/")
+    # S3-specific
+    bucket: Optional[str] = Field(default=None, description="S3 bucket name (S3 only)", example="demo-cfs")
+    region: Optional[str] = Field(default=None, description="AWS region (S3 only)", example="us-east-1")
+    endpoint_url: Optional[str] = Field(default=None, description="S3-compatible endpoint URL for non-AWS providers (S3 only)", example="https://s3.demo.example")

app/routers/storage/storage.py

@@ -1,6 +1,6 @@
 from typing import Annotated
 
-from fastapi import Depends, HTTPException, Query, Request, status
+from fastapi import Depends, HTTPException, Query, Request, status as http_status
 
 from ...types.http import forbidExtraQueryParams
 from ...types.user import User
@@ -32,7 +32,7 @@
         "- `write`: excludes paths that are read-only in a compute-job context\n"
         "- `read`: no filtering\n"
     ),
-    status_code=status.HTTP_200_OK,
+    status_code=http_status.HTTP_200_OK,
     response_model=list[models.StorageInstance],
     responses=DEFAULT_RESPONSES,
     operation_id="getStorageLocations",
@@ -67,3 +67,42 @@ async def get_locations(
     if logicalpath and not locations:
         raise HTTPException(status_code=404, detail=f"No storage location found for logical name '{logicalpath}'")
     return locations
+
+
+@router.get(
+    "/access-endpoints/{resource_id}",
+    summary="Get data access endpoints for a storage resource",
+    description=(
+        "Return the list of data access endpoints for the given storage resource for the "
+        "authenticated user. Each entry describes a protocol (Globus, XRootD, S3, ...) and "
+        "the connection details needed to use it. Adapters may use the authenticated identity "
+        "to include user-specific paths (e.g. home directories, per-user Globus collections). "
+        "Protocol-specific fields (endpoint_id, uri, bucket, etc.) are present only for the "
+        "relevant protocol; unrelated fields are omitted.\n\n"
+        "Optionally filter by protocol and/or endpoint ID."
+    ),
+    status_code=http_status.HTTP_200_OK,
+    response_model=list[models.AccessEndpoint],
+    response_model_exclude_none=True,
+    responses=DEFAULT_RESPONSES,
+    operation_id="getStorageAccessEndpoints",
+    openapi_extra=iri_meta_dict("in_development", "required"),
+)
+async def get_access_endpoints(
+    resource_id: str,
+    request: Request,
+    protocol: Annotated[
+        models.AccessProtocol | None,
+        Query(description="Filter by access protocol"),
+    ] = None,
+    endpoint_id: Annotated[
+        str | None,
+        Query(description="Filter by endpoint ID"),
+    ] = None,
+    user: User = Depends(router.current_user),
+    _forbid=Depends(forbidExtraQueryParams("protocol", "endpoint_id")),
+) -> list[models.AccessEndpoint]:
+    resource = await status_router.adapter.get_resource(resource_id)
+    if not resource:
+        raise HTTPException(status_code=404, detail="Resource not found")
+    return await router.adapter.get_access_endpoints(resource, user, protocol, endpoint_id)

test/test_storage.py

@@ -2,13 +2,15 @@
 """Focused regression tests for the remaining storage endpoint contract and OpenAPI wiring."""
 
 import asyncio
+import datetime
 import os
 import unittest
 
 os.environ.setdefault("IRI_SHOW_MISSING_ROUTES", "true")
 
 from app.demo_adapter import DemoAdapter
 from app.main import APP
+from app import config
 from app.routers.storage import models as storage_models
 
 
@@ -86,15 +88,99 @@ def test_project_scoped_entries_expand_under_remaining_locations_endpoint(self):
         self.assertEqual(len(location_payload), len(self.adapter._user_project_codes(self.user)))
 
     def test_openapi_exposes_only_resource_scoped_storage_locations(self):
-        resolved_locations = self.openapi["paths"]["/api/v1/storage/locations/{resource_id}"]["get"]
-        self.assertNotIn("/api/v1/storage/locations", self.openapi["paths"])
-        self.assertNotIn("/api/v1/storage/mounts/{resource_id}", self.openapi["paths"])
+        prefix = f"/{config.API_URL}"
+        resolved_locations = self.openapi["paths"][f"{prefix}/storage/locations/{{resource_id}}"]["get"]
+        self.assertNotIn(f"{prefix}/storage/locations", self.openapi["paths"])
+        self.assertNotIn(f"{prefix}/storage/mounts/{{resource_id}}", self.openapi["paths"])
         self.assertTrue(
             resolved_locations["responses"]["200"]["content"]["application/json"]["schema"]["items"]["$ref"].endswith(
                 "/StorageInstance"
             )
         )
 
+    def test_access_endpoints_return_all_protocols_for_cfs(self):
+        cfs_resource = self._resource("cfs", "cfs")
+
+        endpoints = asyncio.run(
+            self.adapter.get_access_endpoints(cfs_resource, None, None)
+        )
+
+        self.assertGreater(len(endpoints), 0)
+        self.assertTrue(all(isinstance(e, storage_models.AccessEndpoint) for e in endpoints))
+        protocols = {e.protocol for e in endpoints}
+        self.assertEqual(
+            protocols,
+            {
+                storage_models.AccessProtocol.globus,
+                storage_models.AccessProtocol.xrootd,
+                storage_models.AccessProtocol.s3,
+            },
+        )
+
+    def test_access_endpoints_filter_by_protocol(self):
+        cfs_resource = self._resource("cfs", "cfs")
+
+        endpoints = asyncio.run(
+            self.adapter.get_access_endpoints(cfs_resource, storage_models.AccessProtocol.globus, None)
+        )
+
+        self.assertEqual(len(endpoints), 1)
+        self.assertEqual(endpoints[0].protocol, storage_models.AccessProtocol.globus)
+        self.assertIsNotNone(endpoints[0].endpoint_id)
+        self.assertIsNotNone(endpoints[0].uri)
+
+    def test_access_endpoints_filter_by_endpoint_id(self):
+        cfs_resource = self._resource("cfs", "cfs")
+
+        endpoints = asyncio.run(
+            self.adapter.get_access_endpoints(cfs_resource, None, "xrootd-cfs-demo")
+        )
+
+        self.assertEqual(len(endpoints), 1)
+        self.assertEqual(endpoints[0].id, "xrootd-cfs-demo")
+        self.assertEqual(endpoints[0].protocol, storage_models.AccessProtocol.xrootd)
+        self.assertIsNotNone(endpoints[0].endpoint)
+
+    def test_access_endpoints_hpss_has_only_globus(self):
+        hpss_resource = self._resource("hpss", "hpss")
+
+        endpoints = asyncio.run(
+            self.adapter.get_access_endpoints(hpss_resource, None, None)
+        )
+
+        self.assertEqual(len(endpoints), 1)
+        self.assertEqual(endpoints[0].protocol, storage_models.AccessProtocol.globus)
+
+    def test_access_endpoints_unknown_resource_returns_empty(self):
+        from app.routers.status import models as status_models
+
+        fake_resource = status_models.Resource(
+            id="does-not-exist",
+            site_id="x",
+            group="x",
+            name="x",
+            description="x",
+            capability_ids=[],
+            current_status=status_models.Status.up,
+            resource_type=status_models.ResourceType.storage,
+            last_modified=datetime.datetime.now(datetime.timezone.utc),
+        )
+
+        endpoints = asyncio.run(
+            self.adapter.get_access_endpoints(fake_resource, None, None)
+        )
+
+        self.assertEqual(endpoints, [])
+
+    def test_openapi_exposes_access_endpoints_path(self):
+        prefix = f"/{config.API_URL}"
+        path = self.openapi["paths"][f"{prefix}/storage/{{resource_id}}/access-endpoints"]["get"]
+        self.assertTrue(
+            path["responses"]["200"]["content"]["application/json"]["schema"]["items"]["$ref"].endswith(
+                "/AccessEndpoint"
+            )
+        )
+
 
 if __name__ == "__main__":
     unittest.main()