diff --git a/app/demo_adapter.py b/app/demo_adapter.py index 81fffbb7..1c55e878 100644 --- a/app/demo_adapter.py +++ b/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() diff --git a/app/routers/storage/facility_adapter.py b/app/routers/storage/facility_adapter.py index d650dc58..60af007f 100644 --- a/app/routers/storage/facility_adapter.py +++ b/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 diff --git a/app/routers/storage/models.py b/app/routers/storage/models.py index 4aaa7c81..c62b5281 100644 --- a/app/routers/storage/models.py +++ b/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") diff --git a/app/routers/storage/storage.py b/app/routers/storage/storage.py index cdfc89ac..f726c8cf 100644 --- a/app/routers/storage/storage.py +++ b/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) diff --git a/test/test_storage.py b/test/test_storage.py index cad737e6..25d21f1e 100644 --- a/test/test_storage.py +++ b/test/test_storage.py @@ -2,6 +2,7 @@ """Focused regression tests for the remaining storage endpoint contract and OpenAPI wiring.""" import asyncio +import datetime import os import unittest @@ -9,6 +10,7 @@ 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()