diff --git a/.gitignore b/.gitignore index a15a10c..0dc00fb 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ +# OhMyOpenCode / Sisyphus work artifacts +.sisyphus/ + # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] diff --git a/pyproject.toml b/pyproject.toml index 5c76d8d..8780edd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -89,6 +89,15 @@ noop = "rompy.postprocess:NoopPostprocessor" [project.entry-points."rompy.pipeline"] local = "rompy.pipeline:LocalPipelineBackend" +[project.entry-points."rompy.transfer"] +file = "rompy.transfer.file:FileTransfer" +http = "rompy.transfer.http:HttpTransfer" +https = "rompy.transfer.http:HttpTransfer" +oceanum = "rompy.transfer.oceanum:OceanumTransfer" +s3 = "rompy.transfer.cloud:CloudTransfer" +gs = "rompy.transfer.cloud:CloudTransfer" +az = "rompy.transfer.cloud:CloudTransfer" + [project.optional-dependencies] test = ["pytest", "envyaml", "coverage"] extra = ["gcsfs", "zarr", "cloudpathlib[s3,gs,azure]"] diff --git a/src/rompy/core/data.py b/src/rompy/core/data.py index aef5627..e6cf98d 100644 --- a/src/rompy/core/data.py +++ b/src/rompy/core/data.py @@ -11,13 +11,13 @@ import cartopy.feature as cfeature import matplotlib.pyplot as plt from cloudpathlib import AnyPath -from pydantic import Field, HttpUrl, PrivateAttr, field_validator, model_validator +from pydantic import Field, PrivateAttr, field_validator, model_validator from rompy.core.filters import Filter from rompy.core.grid import BaseGrid, RegularGrid -from rompy.core.http_handler import download_http_file from rompy.core.time import TimeRange from rompy.core.types import DatasetCoords, RompyBaseModel, Slice +from rompy.transfer import get_transfer from rompy.utils import load_entry_points logger = logging.getLogger(__name__) @@ -44,17 +44,17 @@ class DataBlob(DataBase): """Data source for model ingestion. Generic data source for files that either need to be copied to the model directory - or linked if `link` is set to True. Supports both local/cloud paths and remote - HTTP/HTTPS URLs. + or linked if `link` is set to True. Supports local/cloud paths, remote HTTP/HTTPS URLs, + and oceanum:// storage URIs via the transfer registry. Parameters ---------- - source : Path | HttpUrl - URI of the data source, either a local file path, cloud storage URI (s3://, gs://), - or a remote HTTP/HTTPS URL. + source : str | Path | AnyPath + URI of the data source: local file path, cloud storage URI (s3://, gs://), + remote HTTP/HTTPS URL, or oceanum:// storage URI. link : bool Whether to create a symbolic link instead of copying the file. - Note: Cannot be used with HTTP URLs (link=True with HTTP will raise ValueError). + Note: Only works with local file paths (file:// scheme). Examples -------- @@ -85,48 +85,47 @@ class DataBlob(DataBase): default="data_blob", description="Model type discriminator", ) - source: Union[AnyPath, HttpUrl] = Field( + source: Union[str, Path, AnyPath] = Field( description=( "URI of the data source: local file path, cloud storage URI (s3://, gs://), " - "or remote HTTP/HTTPS URL. HTTP/HTTPS URLs are automatically downloaded." + "remote HTTP/HTTPS URL, or oceanum:// URL. Sources are handled via the transfer registry." ), ) link: bool = Field( default=False, description="Whether to create a symbolic link instead of copying the file", ) - _copied: str = PrivateAttr(default=None) - - @field_validator("source", mode="before") - @classmethod - def validate_source(cls, v): - if isinstance(v, str) and (v.startswith("http://") or v.startswith("https://")): - return HttpUrl(v) - return v + _copied: Optional[str] = PrivateAttr(default=None) @model_validator(mode="after") - def validate_http_link_mode(self): - """Validate that HTTP URLs cannot be used with link=True.""" - if isinstance(self.source, HttpUrl) and self.link: - raise ValueError( - "Cannot use link=True with HTTP URLs. " - "HTTP sources must be downloaded (link=False)." - ) + def validate_link_scheme_compat(self): + """Validate that link=True only works with file:// scheme.""" + if self.link: + from rompy.transfer import parse_scheme + + scheme = parse_scheme(str(self.source)) + if scheme != "file": + raise ValueError( + f"Cannot use link=True with {scheme}:// URIs. " + f"Only local file paths support symbolic links." + ) return self - def get(self, destdir: Union[str, Path], name: str = None, *args, **kwargs) -> Path: + def get( + self, destdir: Union[str, Path], name: Optional[str] = None, *args, **kwargs + ) -> Path: """Copy, download, or link the data source to a new directory. - For HTTP/HTTPS URLs, the file is automatically downloaded with retry logic. - For local/cloud paths, the file is either copied or symlinked based on the `link` attribute. + Uses the transfer registry to dispatch based on URI scheme (file://, http://, https://, oceanum://). + For local files, respects the `link` attribute (symlink vs copy). + For remote sources (HTTP, oceanum), always downloads (ignores link flag). Parameters ---------- destdir : str | Path The destination directory to copy/download/link the data source to. name : str, optional - Override the output filename. For HTTP downloads, this overrides the filename - extracted from the URL. + Override the output filename. Returns ------- @@ -144,55 +143,19 @@ def get(self, destdir: Union[str, Path], name: str = None, *args, **kwargs) -> P Raises ------ ValueError - If link=True is used with an HTTP URL. + If link=True is used with a non-file:// URI. + UnsupportedOperation + If the transfer scheme does not support required operations. """ destdir = Path(destdir).resolve() - # Handle HTTP URLs - if isinstance(self.source, HttpUrl): - outfile = download_http_file( - url=str(self.source), dest_dir=destdir, name=name - ) - self._copied = str(outfile) - return outfile - - # Handle local/cloud paths - if self.link: - # Create a symbolic link - if name: - symlink_path = destdir / name - else: - symlink_path = destdir / self.source.name - - # Ensure the destination directory exists - destdir.mkdir(parents=True, exist_ok=True) - - # Remove existing symlink/file if it exists - if symlink_path.exists(): - symlink_path.unlink() - - # Compute the relative path from destdir to self.source - relative_source_path = os.path.relpath(self.source.resolve(), destdir) - - # Create symlink - os.symlink(relative_source_path, symlink_path) - self._copied = symlink_path + transfer = get_transfer(self.source) + outfile = transfer.get( + uri=str(self.source), destdir=destdir, name=name, link=self.link + ) - return symlink_path - else: - # Copy the data source - if self.source.is_dir(): - # Copy directory - outfile = copytree(self.source, destdir) - else: - if name: - outfile = destdir / name - else: - outfile = destdir / self.source.name - if outfile.resolve() != self.source.resolve(): - outfile.write_bytes(self.source.read_bytes()) - self._copied = outfile - return outfile + self._copied = str(outfile) + return outfile GRID_TYPES = Union[BaseGrid, RegularGrid] diff --git a/src/rompy/transfer/__init__.py b/src/rompy/transfer/__init__.py new file mode 100644 index 0000000..a357b77 --- /dev/null +++ b/src/rompy/transfer/__init__.py @@ -0,0 +1,22 @@ +from .base import TransferBase +from .exceptions import UnsupportedOperation +from .manager import ( + TransferManager, + TransferFailurePolicy, + TransferItemResult, + TransferBatchResult, +) +from .registry import get_transfer +from .utils import parse_scheme, join_prefix + +__all__ = [ + "TransferBase", + "UnsupportedOperation", + "TransferManager", + "TransferFailurePolicy", + "TransferItemResult", + "TransferBatchResult", + "get_transfer", + "parse_scheme", + "join_prefix", +] diff --git a/src/rompy/transfer/base.py b/src/rompy/transfer/base.py new file mode 100644 index 0000000..9af5db4 --- /dev/null +++ b/src/rompy/transfer/base.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from pathlib import Path +from typing import List, Optional, Dict, Any + + +class TransferBase(ABC): + """ + Abstract base class defining the transfer interface for ROMPy. + Concrete schemes (e.g., local file, s3, http) will implement these methods. + """ + + @abstractmethod + def get( + self, uri: str, destdir: Path, name: Optional[str] = None, link: bool = False + ) -> Path: + """ + Retrieve data from the given URI and place it under destdir. + Returns the final Path to the retrieved object. + """ + raise NotImplementedError + + @abstractmethod + def exists(self, uri: str) -> bool: + """ + Return True if the given URI exists in the transfer store. + """ + raise NotImplementedError + + @abstractmethod + def list(self, uri: str) -> List[str]: + """ + List items under the given URI and return a list of their names/paths. + """ + raise NotImplementedError + + @abstractmethod + def put(self, local_path: Path, uri: str) -> str: + """ + Upload or copy a local file to the given URI. + Returns the URI that was created. + """ + raise NotImplementedError + + @abstractmethod + def delete(self, uri: str, recursive: bool = False) -> None: + """ + Delete the resource at the given URI. + If recursive is True, delete recursively. + """ + raise NotImplementedError + + @abstractmethod + def stat(self, uri: str) -> Dict[str, Any]: + """ + Return a dictionary with metadata about the given URI. + """ + raise NotImplementedError diff --git a/src/rompy/transfer/cloud.py b/src/rompy/transfer/cloud.py new file mode 100644 index 0000000..7207f17 --- /dev/null +++ b/src/rompy/transfer/cloud.py @@ -0,0 +1,214 @@ +"""Cloud storage transfer implementation using cloudpathlib. + +Supports S3 (s3://), Google Cloud Storage (gs://), and Azure Blob (az://) +via cloudpathlib's unified interface. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Dict, Any, List, Optional + +from cloudpathlib import CloudPath, S3Path, GSPath, AzureBlobPath + +from .base import TransferBase + + +class CloudTransfer(TransferBase): + """ + Transfer implementation for cloud storage via cloudpathlib. + + Supports: + - S3: s3://bucket/key + - Google Cloud Storage: gs://bucket/key + - Azure Blob Storage: az://container/blob + + Credentials are managed via environment variables: + - S3: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION + - GCS: GOOGLE_APPLICATION_CREDENTIALS + - Azure: AZURE_STORAGE_CONNECTION_STRING + """ + + def _parse_cloud_uri(self, uri: str) -> CloudPath: + """ + Parse URI string into appropriate CloudPath object. + + Args: + uri: Cloud storage URI (s3://, gs://, or az://) + + Returns: + CloudPath object (S3Path, GSPath, or AzureBlobPath) + + Raises: + ValueError: If URI scheme is not supported + """ + if uri.startswith("s3://"): + return S3Path(uri) + elif uri.startswith("gs://"): + return GSPath(uri) + elif uri.startswith("az://"): + return AzureBlobPath(uri) + else: + raise ValueError( + f"Unsupported cloud URI scheme: {uri}. " + f"Supported schemes: s3://, gs://, az://" + ) + + def get( + self, uri: str, destdir: Path, name: Optional[str] = None, link: bool = False + ) -> Path: + """ + Download file from cloud storage to local directory. + + Args: + uri: Cloud storage URI + destdir: Local destination directory + name: Optional name for downloaded file (default: use cloud object name) + link: Ignored for cloud storage (always downloads) + + Returns: + Path to downloaded file + + Raises: + FileNotFoundError: If cloud object does not exist + """ + cloud_path = self._parse_cloud_uri(uri) + + if not cloud_path.exists(): + raise FileNotFoundError(f"Cloud object does not exist: {uri}") + + # Determine destination name + dest_name = name or cloud_path.name + dest_path = Path(destdir) / dest_name + + # Ensure destination directory exists + Path(destdir).mkdir(parents=True, exist_ok=True) + + # Download file + cloud_path.download_to(dest_path) + + return dest_path + + def exists(self, uri: str) -> bool: + """Check if cloud object exists.""" + try: + cloud_path = self._parse_cloud_uri(uri) + return cloud_path.exists() + except Exception: + return False + + def list(self, uri: str) -> List[str]: + """ + List objects under cloud prefix. + + Args: + uri: Cloud storage URI (can be directory or file) + + Returns: + List of object names (not full URIs) + + Raises: + FileNotFoundError: If cloud prefix does not exist + """ + cloud_path = self._parse_cloud_uri(uri) + + if not cloud_path.exists(): + raise FileNotFoundError(f"Cloud prefix does not exist: {uri}") + + # If it's a file, return single item + if cloud_path.is_file(): + return [cloud_path.name] + + # If it's a directory, list contents + if cloud_path.is_dir(): + return [p.name for p in cloud_path.iterdir()] + + return [] + + def put(self, local_path: Path, uri: str) -> str: + """ + Upload local file to cloud storage. + + Args: + local_path: Local file path + uri: Cloud destination URI + + Returns: + Cloud URI of uploaded object + + Raises: + FileNotFoundError: If local file does not exist + ValueError: If local_path is a directory (not supported yet) + """ + if not local_path.exists(): + raise FileNotFoundError(f"Local file does not exist: {local_path}") + + if not local_path.is_file(): + raise ValueError( + f"Directory uploads not yet supported: {local_path}. " + f"Please upload files individually." + ) + + cloud_path = self._parse_cloud_uri(uri) + + # Upload file + cloud_path.upload_from(local_path) + + return str(cloud_path) + + def delete(self, uri: str, recursive: bool = False) -> None: + """ + Delete cloud object. + + Args: + uri: Cloud storage URI + recursive: If True, delete directory recursively + + Raises: + ValueError: If trying to delete directory without recursive=True + """ + cloud_path = self._parse_cloud_uri(uri) + + if not cloud_path.exists(): + return # Already deleted + + if cloud_path.is_dir(): + if not recursive: + raise ValueError( + f"Cannot delete cloud directory without recursive=True: {uri}" + ) + # Delete directory recursively + cloud_path.rmtree() + else: + # Delete single file + cloud_path.unlink() + + def stat(self, uri: str) -> Dict[str, Any]: + """ + Get metadata for cloud object. + + Args: + uri: Cloud storage URI + + Returns: + Dictionary with metadata (size, type, etag, etc.) + + Raises: + FileNotFoundError: If cloud object does not exist + """ + cloud_path = self._parse_cloud_uri(uri) + + if not cloud_path.exists(): + raise FileNotFoundError(f"Cloud object does not exist: {uri}") + + stat_result = cloud_path.stat() + + return { + "size": stat_result.st_size if hasattr(stat_result, "st_size") else None, + "mtime": stat_result.st_mtime if hasattr(stat_result, "st_mtime") else None, + "is_file": cloud_path.is_file(), + "is_dir": cloud_path.is_dir(), + "type": "file" if cloud_path.is_file() else "directory", + "uri": str(cloud_path), + "etag": getattr(stat_result, "etag", None), + } diff --git a/src/rompy/transfer/exceptions.py b/src/rompy/transfer/exceptions.py new file mode 100644 index 0000000..0bafddb --- /dev/null +++ b/src/rompy/transfer/exceptions.py @@ -0,0 +1,7 @@ +class UnsupportedOperation(Exception): + """Raised when a transfer implementation doesn't support an operation.""" + + def __init__(self, scheme: str, operation: str): + self.scheme = scheme + self.operation = operation + super().__init__(f"Operation '{operation}' not supported for scheme '{scheme}'") diff --git a/src/rompy/transfer/file.py b/src/rompy/transfer/file.py new file mode 100644 index 0000000..b6a5770 --- /dev/null +++ b/src/rompy/transfer/file.py @@ -0,0 +1,149 @@ +"""File transfer implementation for local filesystem operations.""" + +from __future__ import annotations + +import os +import shutil +from pathlib import Path +from typing import Dict, Any, List, Optional +from urllib.parse import urlparse + +from .base import TransferBase + + +class FileTransfer(TransferBase): + """Transfer implementation for local file:// scheme and plain file paths.""" + + def _parse_file_uri(self, uri: str) -> Path: + """Convert file:// URI or plain path to Path object.""" + if uri.startswith("file://"): + # Remove file:// prefix and convert to path + parsed = urlparse(uri) + return Path(parsed.path) + return Path(uri) + + def get( + self, uri: str, destdir: Path, name: Optional[str] = None, link: bool = False + ) -> Path: + """ + Retrieve local file/directory and place it in destdir. + + Args: + uri: File URI or path string + destdir: Destination directory + name: Optional name for the destination (default: use source name) + link: If True, create symlink instead of copy (local only) + + Returns: + Path to the retrieved file/directory in destdir + """ + source = self._parse_file_uri(uri) + + if not source.exists(): + raise FileNotFoundError(f"Source path does not exist: {source}") + + # Determine destination name + dest_name = name or source.name + dest_path = Path(destdir) / dest_name + + # Ensure destination directory exists + Path(destdir).mkdir(parents=True, exist_ok=True) + + if link: + # Create symlink (relative path from destdir to source) + if dest_path.exists() or dest_path.is_symlink(): + dest_path.unlink() + + # Create relative symlink if possible + try: + rel_source = os.path.relpath(source, destdir) + os.symlink(rel_source, dest_path) + except (ValueError, OSError): + # Fall back to absolute path if relative fails + os.symlink(source.absolute(), dest_path) + else: + # Copy file or directory + if source.is_file(): + dest_path.write_bytes(source.read_bytes()) + elif source.is_dir(): + if dest_path.exists(): + shutil.rmtree(dest_path) + shutil.copytree(source, dest_path) + else: + raise ValueError(f"Source is neither file nor directory: {source}") + + return dest_path + + def exists(self, uri: str) -> bool: + """Check if local path exists.""" + path = self._parse_file_uri(uri) + return path.exists() + + def list(self, uri: str) -> List[str]: + """List directory contents, or return single item for file.""" + path = self._parse_file_uri(uri) + + if not path.exists(): + raise FileNotFoundError(f"Path does not exist: {path}") + + if path.is_file(): + return [path.name] + + return [p.name for p in path.iterdir()] + + def put(self, local_path: Path, uri: str) -> str: + """Copy local file to destination path.""" + dest = self._parse_file_uri(uri) + + # Ensure destination parent directory exists + dest.parent.mkdir(parents=True, exist_ok=True) + + if local_path.is_file(): + shutil.copy2(local_path, dest) + elif local_path.is_dir(): + if dest.exists(): + shutil.rmtree(dest) + shutil.copytree(local_path, dest) + else: + raise ValueError(f"Local path is neither file nor directory: {local_path}") + + return str(dest) + + def delete(self, uri: str, recursive: bool = False) -> None: + """Delete file or directory.""" + path = self._parse_file_uri(uri) + + if not path.exists(): + return # Already deleted + + if path.is_file() or path.is_symlink(): + path.unlink() + elif path.is_dir(): + if recursive: + shutil.rmtree(path) + else: + path.rmdir() # Fails if directory not empty + + def stat(self, uri: str) -> Dict[str, Any]: + """Return file metadata.""" + path = self._parse_file_uri(uri) + + if not path.exists(): + raise FileNotFoundError(f"Path does not exist: {path}") + + stat_result = path.stat() + + return { + "size": stat_result.st_size, + "mtime": stat_result.st_mtime, + "ctime": stat_result.st_ctime, + "is_file": path.is_file(), + "is_dir": path.is_dir(), + "is_symlink": path.is_symlink(), + "type": ( + "file" + if path.is_file() + else ("directory" if path.is_dir() else "symlink") + ), + "path": str(path.absolute()), + } diff --git a/src/rompy/transfer/http.py b/src/rompy/transfer/http.py new file mode 100644 index 0000000..a7ac951 --- /dev/null +++ b/src/rompy/transfer/http.py @@ -0,0 +1,54 @@ +"""HTTP/HTTPS transfer implementation (read-only).""" + +from __future__ import annotations + +from pathlib import Path +from typing import Dict, Any, List, Optional + +from .base import TransferBase +from .exceptions import UnsupportedOperation + + +class HttpTransfer(TransferBase): + """Transfer implementation for http:// and https:// schemes (read-only).""" + + def get( + self, uri: str, destdir: Path, name: Optional[str] = None, link: bool = False + ) -> Path: + """ + Download HTTP/HTTPS resource using existing http_handler. + + Args: + uri: HTTP/HTTPS URL + destdir: Destination directory + name: Optional name for downloaded file + link: Ignored for HTTP (always downloads) + + Returns: + Path to downloaded file + """ + # Import here to avoid circular dependency + from rompy.core.http_handler import download_http_file + + # Use existing HTTP download handler + return download_http_file(url=uri, dest_dir=destdir, name=name) + + def exists(self, uri: str) -> bool: + """Not supported for HTTP - would require HEAD request.""" + raise UnsupportedOperation("http", "exists") + + def list(self, uri: str) -> List[str]: + """Not supported for HTTP - no directory listing.""" + raise UnsupportedOperation("http", "list") + + def put(self, local_path: Path, uri: str) -> str: + """Not supported for HTTP - read-only.""" + raise UnsupportedOperation("http", "put") + + def delete(self, uri: str, recursive: bool = False) -> None: + """Not supported for HTTP - read-only.""" + raise UnsupportedOperation("http", "delete") + + def stat(self, uri: str) -> Dict[str, Any]: + """Not supported for HTTP - would require HEAD request.""" + raise UnsupportedOperation("http", "stat") diff --git a/src/rompy/transfer/manager.py b/src/rompy/transfer/manager.py new file mode 100644 index 0000000..a8a71e3 --- /dev/null +++ b/src/rompy/transfer/manager.py @@ -0,0 +1,168 @@ +"""Multi-destination transfer orchestration for ROMPy.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Optional + +from .registry import get_transfer +from .utils import join_prefix + + +class TransferFailurePolicy(Enum): + """Policy for handling transfer failures when uploading to multiple destinations. + + Attributes: + CONTINUE: Record failures but continue with remaining transfers + FAIL_FAST: Stop and raise exception on first failure + """ + + CONTINUE = "continue" + FAIL_FAST = "fail_fast" + + +@dataclass +class TransferItemResult: + """Result of transferring a single file to a single destination. + + Attributes: + local_path: Source file path on local filesystem + dest_prefix: Destination prefix (treated as folder-like) + target_name: Target filename to append to prefix + dest_uri: Final constructed URI (prefix + target_name) + ok: True if transfer succeeded, False otherwise + error: Error message if transfer failed, None if succeeded + """ + + local_path: Path + dest_prefix: str + target_name: str + dest_uri: str + ok: bool + error: Optional[str] = None + + +@dataclass +class TransferBatchResult: + """Aggregated results from transferring files to multiple destinations. + + Attributes: + total: Total number of transfer attempts + succeeded: Number of successful transfers + failed: Number of failed transfers + items: Detailed per-transfer results + """ + + total: int + succeeded: int + failed: int + items: list[TransferItemResult] = field(default_factory=list) + + def all_succeeded(self) -> bool: + """Check if all transfers succeeded. + + Returns: + True if all transfers succeeded, False otherwise + """ + return self.failed == 0 + + +class TransferManager: + """Orchestrates file transfers to multiple destination prefixes. + + Handles fan-out of local files to multiple destinations using rompy.transfer + backends, with configurable failure policies and aggregated result tracking. + """ + + def transfer_files( + self, + files: list[Path], + destinations: list[str], + name_map: dict[Path, str], + policy: TransferFailurePolicy = TransferFailurePolicy.CONTINUE, + ) -> TransferBatchResult: + """Transfer multiple files to multiple destination prefixes. + + For each file, computes the target name via name_map, then transfers + to all destination prefixes. Destinations are treated as **prefixes** + (folder-like), not complete object URIs. + + Final destination URI construction: + dest_uri = join_prefix(dest_prefix, target_name) + + Example: + files = [Path("restart.ww3"), Path("output.nc")] + destinations = ["s3://bucket/outputs/", "file:///local/backup/"] + name_map = { + Path("restart.ww3"): "20230101_000000_restart.ww3", + Path("output.nc"): "20230101_000000_output.nc" + } + + Results in transfers to: + - s3://bucket/outputs/20230101_000000_restart.ww3 + - s3://bucket/outputs/20230101_000000_output.nc + - file:///local/backup/20230101_000000_restart.ww3 + - file:///local/backup/20230101_000000_output.nc + + Args: + files: List of local file paths to transfer + destinations: List of destination prefixes (e.g., "s3://bucket/path/") + name_map: Mapping from local path to target filename + policy: Failure handling policy (CONTINUE or FAIL_FAST) + + Returns: + TransferBatchResult with aggregated success/failure counts and details + + Raises: + Exception: On first failure if policy is FAIL_FAST + """ + items: list[TransferItemResult] = [] + succeeded = 0 + failed = 0 + + for local_path in files: + target_name = name_map[local_path] + + for dest_prefix in destinations: + dest_uri = join_prefix(dest_prefix, target_name) + + try: + transfer = get_transfer(dest_prefix) + transfer.put(local_path, dest_uri) + + items.append( + TransferItemResult( + local_path=local_path, + dest_prefix=dest_prefix, + target_name=target_name, + dest_uri=dest_uri, + ok=True, + error=None, + ) + ) + succeeded += 1 + + except Exception as e: + error_msg = f"{type(e).__name__}: {str(e)}" + + items.append( + TransferItemResult( + local_path=local_path, + dest_prefix=dest_prefix, + target_name=target_name, + dest_uri=dest_uri, + ok=False, + error=error_msg, + ) + ) + failed += 1 + + if policy == TransferFailurePolicy.FAIL_FAST: + raise + + total = succeeded + failed + return TransferBatchResult( + total=total, succeeded=succeeded, failed=failed, items=items + ) diff --git a/src/rompy/transfer/oceanum.py b/src/rompy/transfer/oceanum.py new file mode 100644 index 0000000..1173741 --- /dev/null +++ b/src/rompy/transfer/oceanum.py @@ -0,0 +1,111 @@ +"""Oceanum storage transfer implementation using fsspec.""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Dict, Any, List, Optional + +from .base import TransferBase + + +class OceanumTransfer(TransferBase): + """Transfer implementation for oceanum:// scheme using fsspec.""" + + def __init__(self): + """Initialize Oceanum transfer with token from environment.""" + self._token = os.environ.get("DATAMESH_TOKEN") + if not self._token: + raise ValueError( + "DATAMESH_TOKEN environment variable not set. " + "Required for oceanum:// storage access." + ) + + # Lazy import fsspec to avoid hard dependency + try: + import fsspec + except ImportError as e: + raise ImportError( + "fsspec is required for oceanum:// storage. " + "Install with: pip install 'fsspec[oceanum]'" + ) from e + + # Initialize oceanum filesystem + self._fs = fsspec.filesystem("oceanum", token=self._token) + + def _strip_scheme(self, uri: str) -> str: + """Remove oceanum:// scheme prefix if present.""" + if uri.startswith("oceanum://"): + return uri[len("oceanum://") :] + return uri + + def get( + self, uri: str, destdir: Path, name: Optional[str] = None, link: bool = False + ) -> Path: + """ + Download from Oceanum storage to local destdir. + + Args: + uri: oceanum:// URI + destdir: Local destination directory + name: Optional name for downloaded file + link: Ignored for remote storage (always downloads) + + Returns: + Path to downloaded file + """ + remote_path = self._strip_scheme(uri) + + # Determine local destination + if name is None: + name = Path(remote_path).name + + dest_path = Path(destdir) / name + Path(destdir).mkdir(parents=True, exist_ok=True) + + # Download using fsspec + self._fs.get(remote_path, str(dest_path), recursive=False) + + return dest_path + + def exists(self, uri: str) -> bool: + """Check if path exists in Oceanum storage.""" + remote_path = self._strip_scheme(uri) + return self._fs.exists(remote_path) + + def list(self, uri: str) -> List[str]: + """List directory contents in Oceanum storage.""" + remote_path = self._strip_scheme(uri) + + # ls returns full paths, extract just names + entries = self._fs.ls(remote_path, detail=False) + return [Path(e).name for e in entries] + + def put(self, local_path: Path, uri: str) -> str: + """Upload local file to Oceanum storage.""" + remote_path = self._strip_scheme(uri) + + # Upload using fsspec + self._fs.put(str(local_path), remote_path, recursive=False) + + return f"oceanum://{remote_path}" + + def delete(self, uri: str, recursive: bool = False) -> None: + """Delete file/directory from Oceanum storage.""" + remote_path = self._strip_scheme(uri) + self._fs.rm(remote_path, recursive=recursive) + + def stat(self, uri: str) -> Dict[str, Any]: + """Return file metadata from Oceanum storage.""" + remote_path = self._strip_scheme(uri) + + # fsspec info() returns detailed metadata + info = self._fs.info(remote_path) + + return { + "size": info.get("size", 0), + "type": info.get("type", "unknown"), + "name": info.get("name", remote_path), + "mtime": info.get("mtime"), + **info, # Include all other metadata + } diff --git a/src/rompy/transfer/registry.py b/src/rompy/transfer/registry.py new file mode 100644 index 0000000..723c6a8 --- /dev/null +++ b/src/rompy/transfer/registry.py @@ -0,0 +1,110 @@ +"""Transfer registry for scheme-based dispatch.""" + +from __future__ import annotations + +import importlib.metadata +from pathlib import Path +from typing import TYPE_CHECKING, Any, Dict, Type, Union + +from .utils import parse_scheme + +if TYPE_CHECKING: + from .base import TransferBase + +# Global registry cache +_REGISTRY: Dict[str, Type["TransferBase"]] | None = None + + +def _load_transfer_registry() -> Dict[str, Type["TransferBase"]]: + """ + Load transfer implementations from entry points. + + Returns: + Dict mapping scheme names (lowercase) to TransferBase classes. + + Raises: + ValueError: If duplicate schemes are found. + """ + registry: Dict[str, Type["TransferBase"]] = {} + seen_schemes: Dict[str, str] = {} # scheme -> entry_point_name + + # Load entry points from rompy.transfer group + entry_points = importlib.metadata.entry_points(group="rompy.transfer") + + for ep in entry_points: + scheme = ep.name.lower() + + # Check for duplicates + if scheme in seen_schemes: + raise ValueError( + f"Duplicate transfer entry point for scheme '{scheme}': " + f"'{seen_schemes[scheme]}' and '{ep.name}'" + ) + + seen_schemes[scheme] = ep.name + + # Load the transfer class + try: + transfer_class = ep.load() + registry[scheme] = transfer_class + except Exception as e: + # Log warning but don't fail - allow optional transfers + import warnings + + warnings.warn( + f"Failed to load transfer for scheme '{scheme}' from entry point '{ep.name}': {e}", + UserWarning, + ) + + return registry + + +def get_registry() -> Dict[str, Type["TransferBase"]]: + """ + Get the cached transfer registry. + + Lazy-loads the registry on first access. + """ + global _REGISTRY + if _REGISTRY is None: + _REGISTRY = _load_transfer_registry() + return _REGISTRY + + +def get_transfer(uri_or_pathlike: Union[str, Path, Any]) -> "TransferBase": + """ + Get a transfer instance for the given URI or path-like object. + + Args: + uri_or_pathlike: URI string, Path, or cloudpathlib.AnyPath object + + Returns: + Instantiated transfer object for the scheme + + Raises: + KeyError: If no transfer is registered for the URI's scheme + """ + # Convert Path-like objects to string URIs + if isinstance(uri_or_pathlike, Path): + uri = str(uri_or_pathlike) + elif hasattr(uri_or_pathlike, "__str__"): + # Handle cloudpathlib.AnyPath and other path-like objects + uri = str(uri_or_pathlike) + else: + uri = uri_or_pathlike + + # Parse the scheme + scheme = parse_scheme(uri) + + # Look up transfer class in registry + registry = get_registry() + if scheme not in registry: + available = ", ".join(sorted(registry.keys())) + raise KeyError( + f"No transfer registered for scheme '{scheme}'. " + f"Available schemes: {available}" + ) + + # Instantiate and return the transfer + transfer_class = registry[scheme] + return transfer_class() diff --git a/src/rompy/transfer/utils.py b/src/rompy/transfer/utils.py new file mode 100644 index 0000000..bac8bc7 --- /dev/null +++ b/src/rompy/transfer/utils.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from urllib.parse import urlparse, urlunparse + + +def parse_scheme(uri: str) -> str: + """ + Return the lower-cased scheme for a URI. + - Empty scheme or empty input -> 'file' + - Windows drive letters (e.g., 'C:/path') are treated as file scheme + - Case-insensitive: returns lowercase + - For URIs with a real scheme (e.g., 'https', 's3'), return the scheme + """ + if uri is None: + return "file" + parsed = urlparse(uri) + scheme = (parsed.scheme or "").lower() + # Treat Windows drive letters like 'C:/path' as file scheme + if len(scheme) == 1 and scheme.isalpha(): + return "file" + if scheme == "": + return "file" + return scheme + + +def join_prefix(prefix: str, name: str) -> str: + """Join a destination prefix with a target filename to create a full URI. + + Handles various URI schemes (file://, s3://, gs://, http://, etc.) and + plain filesystem paths, ensuring proper path construction without + double slashes (except in scheme://authority). + + Args: + prefix: Destination prefix (folder-like). May be a URI with scheme + (e.g., "s3://bucket/outputs/") or plain path ("/local/outputs/"). + name: Target filename to append to the prefix (e.g., "output.nc"). + + Returns: + Complete destination URI or path. + + Examples: + >>> join_prefix("s3://bucket/outputs/", "file.nc") + 's3://bucket/outputs/file.nc' + + >>> join_prefix("s3://bucket/outputs", "file.nc") + 's3://bucket/outputs/file.nc' + + >>> join_prefix("/local/outputs/", "file.nc") + '/local/outputs/file.nc' + + >>> join_prefix("file:///tmp/data/", "file.nc") + 'file:///tmp/data/file.nc' + + >>> join_prefix("gs://bucket", "file.nc") + 'gs://bucket/file.nc' + """ + parsed = urlparse(prefix) + + if parsed.scheme: + path = parsed.path + if not path.endswith("/"): + path += "/" + + full_path = path + name + + return urlunparse( + ( + parsed.scheme, + parsed.netloc, + full_path, + parsed.params, + parsed.query, + parsed.fragment, + ) + ) + else: + prefix_clean = prefix.rstrip("/") + return f"{prefix_clean}/{name}" diff --git a/tests/test_cloud_transfer.py b/tests/test_cloud_transfer.py new file mode 100644 index 0000000..dcc6570 --- /dev/null +++ b/tests/test_cloud_transfer.py @@ -0,0 +1,309 @@ +"""Tests for cloud storage transfer backend using cloudpathlib.""" + +import pytest +from pathlib import Path +from unittest.mock import Mock, patch, MagicMock + +from rompy.transfer.cloud import CloudTransfer + + +@pytest.fixture +def cloud_transfer(): + """Create CloudTransfer instance for testing.""" + return CloudTransfer() + + +@pytest.fixture +def mock_s3_path(): + """Mock S3Path for testing.""" + with patch("rompy.transfer.cloud.S3Path") as mock: + yield mock + + +@pytest.fixture +def mock_gs_path(): + """Mock GSPath for testing.""" + with patch("rompy.transfer.cloud.GSPath") as mock: + yield mock + + +@pytest.fixture +def mock_az_path(): + """Mock AzureBlobPath for testing.""" + with patch("rompy.transfer.cloud.AzureBlobPath") as mock: + yield mock + + +class TestCloudTransferParsing: + """Test URI parsing and scheme detection.""" + + def test_parse_s3_uri(self, cloud_transfer, mock_s3_path): + """Test parsing S3 URI.""" + uri = "s3://bucket/key/file.txt" + result = cloud_transfer._parse_cloud_uri(uri) + mock_s3_path.assert_called_once_with(uri) + + def test_parse_gs_uri(self, cloud_transfer, mock_gs_path): + """Test parsing GCS URI.""" + uri = "gs://bucket/key/file.txt" + result = cloud_transfer._parse_cloud_uri(uri) + mock_gs_path.assert_called_once_with(uri) + + def test_parse_az_uri(self, cloud_transfer, mock_az_path): + """Test parsing Azure Blob URI.""" + uri = "az://container/blob/file.txt" + result = cloud_transfer._parse_cloud_uri(uri) + mock_az_path.assert_called_once_with(uri) + + def test_parse_unsupported_scheme(self, cloud_transfer): + """Test error on unsupported scheme.""" + with pytest.raises(ValueError, match="Unsupported cloud URI scheme"): + cloud_transfer._parse_cloud_uri("ftp://server/file.txt") + + +class TestCloudTransferGet: + """Test downloading files from cloud storage.""" + + def test_get_s3_file(self, cloud_transfer, mock_s3_path, tmp_path): + """Test downloading file from S3.""" + # Setup mock + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.name = "file.txt" + mock_cloud.download_to.return_value = None + mock_s3_path.return_value = mock_cloud + + # Execute + uri = "s3://bucket/path/file.txt" + result = cloud_transfer.get(uri, tmp_path) + + # Verify + assert result == tmp_path / "file.txt" + mock_cloud.download_to.assert_called_once_with(tmp_path / "file.txt") + + def test_get_with_custom_name(self, cloud_transfer, mock_s3_path, tmp_path): + """Test downloading with custom destination name.""" + # Setup mock + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.name = "original.txt" + mock_s3_path.return_value = mock_cloud + + # Execute + result = cloud_transfer.get("s3://bucket/file.txt", tmp_path, name="custom.txt") + + # Verify + assert result == tmp_path / "custom.txt" + mock_cloud.download_to.assert_called_once_with(tmp_path / "custom.txt") + + def test_get_nonexistent_file(self, cloud_transfer, mock_s3_path, tmp_path): + """Test error when downloading nonexistent file.""" + # Setup mock + mock_cloud = MagicMock() + mock_cloud.exists.return_value = False + mock_s3_path.return_value = mock_cloud + + # Execute and verify + with pytest.raises(FileNotFoundError, match="Cloud object does not exist"): + cloud_transfer.get("s3://bucket/missing.txt", tmp_path) + + def test_get_link_ignored(self, cloud_transfer, mock_s3_path, tmp_path): + """Test that link parameter is ignored for cloud storage.""" + # Setup mock + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.name = "file.txt" + mock_s3_path.return_value = mock_cloud + + # Execute with link=True (should be ignored) + result = cloud_transfer.get("s3://bucket/file.txt", tmp_path, link=True) + + # Verify download was called (not symlink) + mock_cloud.download_to.assert_called_once() + + +class TestCloudTransferExists: + """Test checking existence of cloud objects.""" + + def test_exists_true(self, cloud_transfer, mock_s3_path): + """Test exists returns True for existing object.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_s3_path.return_value = mock_cloud + + assert cloud_transfer.exists("s3://bucket/file.txt") is True + + def test_exists_false(self, cloud_transfer, mock_s3_path): + """Test exists returns False for nonexistent object.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = False + mock_s3_path.return_value = mock_cloud + + assert cloud_transfer.exists("s3://bucket/missing.txt") is False + + def test_exists_handles_exceptions(self, cloud_transfer, mock_s3_path): + """Test exists returns False on exceptions.""" + mock_s3_path.side_effect = Exception("Connection error") + + assert cloud_transfer.exists("s3://bucket/file.txt") is False + + +class TestCloudTransferList: + """Test listing cloud objects.""" + + def test_list_file(self, cloud_transfer, mock_s3_path): + """Test listing single file.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.is_file.return_value = True + mock_cloud.is_dir.return_value = False + mock_cloud.name = "file.txt" + mock_s3_path.return_value = mock_cloud + + result = cloud_transfer.list("s3://bucket/file.txt") + assert result == ["file.txt"] + + def test_list_directory(self, cloud_transfer, mock_s3_path): + """Test listing directory contents.""" + # Setup mock directory + mock_file1 = MagicMock() + mock_file1.name = "file1.txt" + mock_file2 = MagicMock() + mock_file2.name = "file2.txt" + + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.is_file.return_value = False + mock_cloud.is_dir.return_value = True + mock_cloud.iterdir.return_value = [mock_file1, mock_file2] + mock_s3_path.return_value = mock_cloud + + result = cloud_transfer.list("s3://bucket/prefix/") + assert result == ["file1.txt", "file2.txt"] + + def test_list_nonexistent(self, cloud_transfer, mock_s3_path): + """Test error listing nonexistent prefix.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = False + mock_s3_path.return_value = mock_cloud + + with pytest.raises(FileNotFoundError, match="Cloud prefix does not exist"): + cloud_transfer.list("s3://bucket/missing/") + + +class TestCloudTransferPut: + """Test uploading files to cloud storage.""" + + def test_put_file(self, cloud_transfer, mock_s3_path, tmp_path): + """Test uploading local file to cloud.""" + # Create local file + local_file = tmp_path / "test.txt" + local_file.write_text("content") + + # Setup mock + mock_cloud = MagicMock() + mock_cloud.__str__.return_value = "s3://bucket/dest.txt" + mock_s3_path.return_value = mock_cloud + + # Execute + result = cloud_transfer.put(local_file, "s3://bucket/dest.txt") + + # Verify + assert result == "s3://bucket/dest.txt" + mock_cloud.upload_from.assert_called_once_with(local_file) + + def test_put_nonexistent_file(self, cloud_transfer, mock_s3_path, tmp_path): + """Test error uploading nonexistent file.""" + with pytest.raises(FileNotFoundError, match="Local file does not exist"): + cloud_transfer.put(tmp_path / "missing.txt", "s3://bucket/dest.txt") + + def test_put_directory_not_supported(self, cloud_transfer, mock_s3_path, tmp_path): + """Test error uploading directory.""" + with pytest.raises(ValueError, match="Directory uploads not yet supported"): + cloud_transfer.put(tmp_path, "s3://bucket/dest/") + + +class TestCloudTransferDelete: + """Test deleting cloud objects.""" + + def test_delete_file(self, cloud_transfer, mock_s3_path): + """Test deleting single file.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.is_dir.return_value = False + mock_s3_path.return_value = mock_cloud + + cloud_transfer.delete("s3://bucket/file.txt") + mock_cloud.unlink.assert_called_once() + + def test_delete_nonexistent_ignored(self, cloud_transfer, mock_s3_path): + """Test deleting nonexistent object (no-op).""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = False + mock_s3_path.return_value = mock_cloud + + cloud_transfer.delete("s3://bucket/missing.txt") + mock_cloud.unlink.assert_not_called() + + def test_delete_directory_recursive(self, cloud_transfer, mock_s3_path): + """Test deleting directory recursively.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.is_dir.return_value = True + mock_s3_path.return_value = mock_cloud + + cloud_transfer.delete("s3://bucket/prefix/", recursive=True) + mock_cloud.rmtree.assert_called_once() + + def test_delete_directory_without_recursive(self, cloud_transfer, mock_s3_path): + """Test error deleting directory without recursive flag.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.is_dir.return_value = True + mock_s3_path.return_value = mock_cloud + + with pytest.raises( + ValueError, match="Cannot delete cloud directory without recursive=True" + ): + cloud_transfer.delete("s3://bucket/prefix/", recursive=False) + + +class TestCloudTransferStat: + """Test getting cloud object metadata.""" + + def test_stat_file(self, cloud_transfer, mock_s3_path): + """Test getting file metadata.""" + # Setup mock + mock_stat = MagicMock() + mock_stat.st_size = 1024 + mock_stat.st_mtime = 1234567890.0 + mock_stat.etag = "abc123" + + mock_cloud = MagicMock() + mock_cloud.exists.return_value = True + mock_cloud.stat.return_value = mock_stat + mock_cloud.is_file.return_value = True + mock_cloud.is_dir.return_value = False + mock_cloud.__str__.return_value = "s3://bucket/file.txt" + mock_s3_path.return_value = mock_cloud + + # Execute + result = cloud_transfer.stat("s3://bucket/file.txt") + + # Verify + assert result["size"] == 1024 + assert result["mtime"] == 1234567890.0 + assert result["is_file"] is True + assert result["is_dir"] is False + assert result["type"] == "file" + assert result["uri"] == "s3://bucket/file.txt" + assert result["etag"] == "abc123" + + def test_stat_nonexistent(self, cloud_transfer, mock_s3_path): + """Test error getting metadata for nonexistent object.""" + mock_cloud = MagicMock() + mock_cloud.exists.return_value = False + mock_s3_path.return_value = mock_cloud + + with pytest.raises(FileNotFoundError, match="Cloud object does not exist"): + cloud_transfer.stat("s3://bucket/missing.txt") diff --git a/tests/test_datablob_http.py b/tests/test_datablob_http.py index ee4b96d..2fe5cbb 100644 --- a/tests/test_datablob_http.py +++ b/tests/test_datablob_http.py @@ -117,16 +117,15 @@ def huge_content(): @pytest.mark.skipif(respx is None, reason="respx not installed") def test_datablob_accepts_http_url(): - """Test that DataBlob accepts HTTP URLs.""" + """Test that DataBlob accepts HTTP URLs as plain strings.""" from rompy.core.data import DataBlob - from pydantic import HttpUrl url = "https://example.com/data.nc" blob = DataBlob(source=url) - assert isinstance(blob.source, HttpUrl) - assert str(blob.source) == url + assert isinstance(blob.source, str) + assert blob.source == url assert blob.link is False @@ -138,7 +137,7 @@ def test_datablob_http_link_error(): url = "https://example.com/data.nc" - with pytest.raises(ValidationError, match="Cannot use link=True with HTTP URLs"): + with pytest.raises(ValidationError, match="Cannot use link=True with https://"): DataBlob(source=url, link=True) diff --git a/tests/test_join_prefix.py b/tests/test_join_prefix.py new file mode 100644 index 0000000..634270f --- /dev/null +++ b/tests/test_join_prefix.py @@ -0,0 +1,53 @@ +"""Tests for join_prefix utility.""" + +import pytest + +from rompy.transfer.utils import join_prefix + + +def test_join_prefix_s3_with_trailing_slash(): + """Test joining S3 prefix with trailing slash.""" + result = join_prefix("s3://bucket/outputs/", "file.nc") + assert result == "s3://bucket/outputs/file.nc" + + +def test_join_prefix_s3_without_trailing_slash(): + """Test joining S3 prefix without trailing slash.""" + result = join_prefix("s3://bucket/outputs", "file.nc") + assert result == "s3://bucket/outputs/file.nc" + + +def test_join_prefix_local_path_with_trailing_slash(): + """Test joining local path with trailing slash.""" + result = join_prefix("/local/outputs/", "file.nc") + assert result == "/local/outputs/file.nc" + + +def test_join_prefix_local_path_without_trailing_slash(): + """Test joining local path without trailing slash.""" + result = join_prefix("/local/outputs", "file.nc") + assert result == "/local/outputs/file.nc" + + +def test_join_prefix_file_uri(): + """Test joining file:// URI.""" + result = join_prefix("file:///tmp/data/", "file.nc") + assert result == "file:///tmp/data/file.nc" + + +def test_join_prefix_gs_bucket(): + """Test joining Google Cloud Storage bucket.""" + result = join_prefix("gs://bucket", "file.nc") + assert result == "gs://bucket/file.nc" + + +def test_join_prefix_nested_path(): + """Test joining with nested prefix.""" + result = join_prefix("s3://bucket/path/to/outputs/", "file.nc") + assert result == "s3://bucket/path/to/outputs/file.nc" + + +def test_join_prefix_relative_path(): + """Test joining relative path.""" + result = join_prefix("outputs/data", "file.nc") + assert result == "outputs/data/file.nc" diff --git a/tests/test_transfer_file.py b/tests/test_transfer_file.py new file mode 100644 index 0000000..0863c28 --- /dev/null +++ b/tests/test_transfer_file.py @@ -0,0 +1,145 @@ +"""Tests for FileTransfer implementation.""" + +import os +from pathlib import Path + +import pytest + +from rompy.transfer.file import FileTransfer +from rompy.transfer.exceptions import UnsupportedOperation + + +@pytest.fixture +def transfer(): + return FileTransfer() + + +@pytest.fixture +def temp_source_file(tmp_path): + source = tmp_path / "source.txt" + source.write_text("test content") + return source + + +@pytest.fixture +def temp_source_dir(tmp_path): + source_dir = tmp_path / "source_dir" + source_dir.mkdir() + (source_dir / "file1.txt").write_text("content1") + (source_dir / "file2.txt").write_text("content2") + return source_dir + + +def test_file_copy(transfer, temp_source_file, tmp_path): + """Test FileTransfer.get() copies file.""" + destdir = tmp_path / "dest" + destdir.mkdir() + + result = transfer.get(str(temp_source_file), destdir, link=False) + + assert result.exists() + assert result.read_text() == "test content" + assert result != temp_source_file + assert not result.is_symlink() + + +def test_file_copy_with_name(transfer, temp_source_file, tmp_path): + """Test FileTransfer.get() copies file with custom name.""" + destdir = tmp_path / "dest" + destdir.mkdir() + + result = transfer.get(str(temp_source_file), destdir, name="custom.txt", link=False) + + assert result.name == "custom.txt" + assert result.read_text() == "test content" + + +def test_file_link(transfer, temp_source_file, tmp_path): + """Test FileTransfer.get() creates symlink.""" + destdir = tmp_path / "dest" + destdir.mkdir() + + result = transfer.get(str(temp_source_file), destdir, link=True) + + assert result.exists() + assert result.is_symlink() + assert result.read_text() == "test content" + assert result.resolve() == temp_source_file.resolve() + + +def test_file_link_relative_path(transfer, temp_source_file, tmp_path): + """Test FileTransfer.get() creates relative symlink.""" + destdir = tmp_path / "dest" + destdir.mkdir() + + result = transfer.get(str(temp_source_file), destdir, link=True) + + link_target = os.readlink(result) + assert not Path(link_target).is_absolute() + + +def test_file_exists(transfer, temp_source_file): + """Test FileTransfer.exists() for existing file.""" + assert transfer.exists(str(temp_source_file)) is True + + +def test_file_not_exists(transfer, tmp_path): + """Test FileTransfer.exists() for non-existing file.""" + assert transfer.exists(str(tmp_path / "missing.txt")) is False + + +def test_file_list_directory(transfer, temp_source_dir): + """Test FileTransfer.list() lists directory contents.""" + items = transfer.list(str(temp_source_dir)) + + assert len(items) == 2 + assert "file1.txt" in items + assert "file2.txt" in items + + +def test_file_list_file(transfer, temp_source_file): + """Test FileTransfer.list() returns single item for file.""" + items = transfer.list(str(temp_source_file)) + + assert items == [temp_source_file.name] + + +def test_file_put(transfer, temp_source_file, tmp_path): + """Test FileTransfer.put() copies file to destination.""" + dest_uri = str(tmp_path / "dest" / "output.txt") + + result_uri = transfer.put(temp_source_file, dest_uri) + + assert result_uri == dest_uri + result_path = Path(dest_uri) + assert result_path.exists() + assert result_path.read_text() == "test content" + + +def test_file_delete(transfer, temp_source_file): + """Test FileTransfer.delete() removes file.""" + assert temp_source_file.exists() + + transfer.delete(str(temp_source_file), recursive=False) + + assert not temp_source_file.exists() + + +def test_file_delete_directory_recursive(transfer, temp_source_dir): + """Test FileTransfer.delete() removes directory recursively.""" + assert temp_source_dir.exists() + assert (temp_source_dir / "file1.txt").exists() + + transfer.delete(str(temp_source_dir), recursive=True) + + assert not temp_source_dir.exists() + + +def test_file_stat(transfer, temp_source_file): + """Test FileTransfer.stat() returns file metadata.""" + stat_dict = transfer.stat(str(temp_source_file)) + + assert "size" in stat_dict + assert "mtime" in stat_dict + assert stat_dict["size"] == len("test content") + assert stat_dict["type"] == "file" diff --git a/tests/test_transfer_http.py b/tests/test_transfer_http.py new file mode 100644 index 0000000..3be8350 --- /dev/null +++ b/tests/test_transfer_http.py @@ -0,0 +1,100 @@ +"""Tests for HttpTransfer implementation.""" + +from pathlib import Path + +import httpx +import pytest + +try: + import respx +except ImportError: + respx = None + +from rompy.transfer.http import HttpTransfer +from rompy.transfer.exceptions import UnsupportedOperation + + +@pytest.fixture +def transfer(): + return HttpTransfer() + + +@pytest.mark.skipif(respx is None, reason="respx not installed") +def test_http_get(transfer, tmp_path): + """Test HttpTransfer.get() downloads file via http_handler.""" + url = "https://example.com/test.nc" + content = b"test netcdf content" + + with respx.mock: + respx.get(url).mock(return_value=httpx.Response(200, content=content)) + + result = transfer.get(url, tmp_path, link=False) + + assert result.exists() + assert result.read_bytes() == content + assert result.name == "test.nc" + + +@pytest.mark.skipif(respx is None, reason="respx not installed") +def test_http_get_with_name(transfer, tmp_path): + """Test HttpTransfer.get() with custom filename.""" + url = "https://example.com/test.nc" + content = b"test netcdf content" + + with respx.mock: + respx.get(url).mock(return_value=httpx.Response(200, content=content)) + + result = transfer.get(url, tmp_path, name="custom.nc", link=False) + + assert result.exists() + assert result.name == "custom.nc" + assert result.read_bytes() == content + + +@pytest.mark.skipif(respx is None, reason="respx not installed") +def test_http_get_ignores_link(transfer, tmp_path): + """Test HttpTransfer.get() ignores link=True (always downloads).""" + url = "https://example.com/test.nc" + content = b"test netcdf content" + + with respx.mock: + respx.get(url).mock(return_value=httpx.Response(200, content=content)) + + result = transfer.get(url, tmp_path, link=True) + + assert result.exists() + assert not result.is_symlink() + assert result.read_bytes() == content + + +def test_http_exists_unsupported(transfer): + """Test HttpTransfer.exists() raises UnsupportedOperation.""" + with pytest.raises(UnsupportedOperation, match="exists.*http"): + transfer.exists("https://example.com/test.nc") + + +def test_http_list_unsupported(transfer): + """Test HttpTransfer.list() raises UnsupportedOperation.""" + with pytest.raises(UnsupportedOperation, match="list.*http"): + transfer.list("https://example.com/") + + +def test_http_put_unsupported(transfer, tmp_path): + """Test HttpTransfer.put() raises UnsupportedOperation.""" + test_file = tmp_path / "test.txt" + test_file.write_text("content") + + with pytest.raises(UnsupportedOperation, match="put.*http"): + transfer.put(test_file, "https://example.com/test.txt") + + +def test_http_delete_unsupported(transfer): + """Test HttpTransfer.delete() raises UnsupportedOperation.""" + with pytest.raises(UnsupportedOperation, match="delete.*http"): + transfer.delete("https://example.com/test.nc") + + +def test_http_stat_unsupported(transfer): + """Test HttpTransfer.stat() raises UnsupportedOperation.""" + with pytest.raises(UnsupportedOperation, match="stat.*http"): + transfer.stat("https://example.com/test.nc") diff --git a/tests/test_transfer_manager.py b/tests/test_transfer_manager.py new file mode 100644 index 0000000..e9d6207 --- /dev/null +++ b/tests/test_transfer_manager.py @@ -0,0 +1,142 @@ +"""Tests for TransferManager.""" + +import pytest +from pathlib import Path + +from rompy.transfer.manager import ( + TransferManager, + TransferFailurePolicy, + TransferItemResult, + TransferBatchResult, +) + + +@pytest.fixture +def temp_files(tmp_path): + """Create temporary test files.""" + file1 = tmp_path / "file1.txt" + file2 = tmp_path / "file2.txt" + file1.write_text("content1") + file2.write_text("content2") + return [file1, file2] + + +@pytest.fixture +def temp_dest(tmp_path): + """Create temporary destination directories.""" + dest1 = tmp_path / "dest1" + dest2 = tmp_path / "dest2" + dest1.mkdir() + dest2.mkdir() + return [str(dest1), str(dest2)] + + +def test_transfer_manager_single_file_single_dest(temp_files, temp_dest): + """Test transferring a single file to a single destination.""" + manager = TransferManager() + files = [temp_files[0]] + destinations = [temp_dest[0]] + name_map = {temp_files[0]: "renamed.txt"} + + result = manager.transfer_files(files, destinations, name_map) + + assert result.total == 1 + assert result.succeeded == 1 + assert result.failed == 0 + assert result.all_succeeded() + assert len(result.items) == 1 + + item = result.items[0] + assert item.local_path == temp_files[0] + assert item.dest_prefix == temp_dest[0] + assert item.target_name == "renamed.txt" + assert item.ok is True + assert item.error is None + + transferred = Path(item.dest_uri) + assert transferred.exists() + assert transferred.read_text() == "content1" + + +def test_transfer_manager_multiple_files_multiple_dest(temp_files, temp_dest): + """Test transferring multiple files to multiple destinations.""" + manager = TransferManager() + name_map = {temp_files[0]: "file1_renamed.txt", temp_files[1]: "file2_renamed.txt"} + + result = manager.transfer_files(temp_files, temp_dest, name_map) + + assert result.total == 4 + assert result.succeeded == 4 + assert result.failed == 0 + assert result.all_succeeded() + assert len(result.items) == 4 + + for dest in temp_dest: + assert (Path(dest) / "file1_renamed.txt").exists() + assert (Path(dest) / "file2_renamed.txt").exists() + assert (Path(dest) / "file1_renamed.txt").read_text() == "content1" + assert (Path(dest) / "file2_renamed.txt").read_text() == "content2" + + +def test_transfer_manager_continue_on_failure(temp_files, tmp_path): + """Test CONTINUE policy records failures but continues.""" + manager = TransferManager() + good_dest = str(tmp_path / "good") + bad_dest = "/nonexistent/readonly/path" + Path(good_dest).mkdir() + + destinations = [good_dest, bad_dest] + name_map = {temp_files[0]: "file.txt"} + + result = manager.transfer_files( + [temp_files[0]], destinations, name_map, policy=TransferFailurePolicy.CONTINUE + ) + + assert result.total == 2 + assert result.succeeded == 1 + assert result.failed == 1 + assert not result.all_succeeded() + + assert result.items[0].ok is True + assert result.items[1].ok is False + assert result.items[1].error is not None + + +def test_transfer_manager_fail_fast_on_error(temp_files, tmp_path): + """Test FAIL_FAST policy raises on first failure.""" + manager = TransferManager() + good_dest = str(tmp_path / "good") + bad_dest = "/nonexistent/readonly/path" + Path(good_dest).mkdir() + + destinations = [bad_dest, good_dest] + name_map = {temp_files[0]: "file.txt"} + + with pytest.raises(Exception): + manager.transfer_files( + [temp_files[0]], + destinations, + name_map, + policy=TransferFailurePolicy.FAIL_FAST, + ) + + +def test_transfer_batch_result_all_succeeded(): + """Test TransferBatchResult.all_succeeded() helper.""" + result_success = TransferBatchResult(total=2, succeeded=2, failed=0, items=[]) + assert result_success.all_succeeded() + + result_failure = TransferBatchResult(total=2, succeeded=1, failed=1, items=[]) + assert not result_failure.all_succeeded() + + +def test_transfer_manager_empty_lists(tmp_path): + """Test transfer with empty file list.""" + manager = TransferManager() + result = manager.transfer_files([], [str(tmp_path)], {}) + + assert result.total == 0 + assert result.succeeded == 0 + assert result.failed == 0 + assert result.all_succeeded() + assert len(result.items) == 0 diff --git a/tests/test_transfer_oceanum.py b/tests/test_transfer_oceanum.py new file mode 100644 index 0000000..271fbf6 --- /dev/null +++ b/tests/test_transfer_oceanum.py @@ -0,0 +1,115 @@ +"""Tests for OceanumTransfer implementation.""" + +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from rompy.transfer.oceanum import OceanumTransfer + + +@pytest.fixture +def mock_fs(): + """Mock fsspec filesystem.""" + fs = MagicMock() + fs.exists.return_value = True + fs.ls.return_value = ["file1.nc", "file2.nc"] + fs.info.return_value = {"size": 1024, "type": "file"} + return fs + + +@pytest.fixture +def transfer_with_token(monkeypatch, mock_fs): + """Create OceanumTransfer with DATAMESH_TOKEN set and mocked fsspec.""" + monkeypatch.setenv("DATAMESH_TOKEN", "test-token-123") + with patch("fsspec.filesystem", return_value=mock_fs): + transfer = OceanumTransfer() + transfer._fs = mock_fs # Replace with mock after creation + return transfer + + +def test_oceanum_missing_token(): + """Test OceanumTransfer raises error without DATAMESH_TOKEN.""" + with patch.dict(os.environ, {}, clear=True): + with pytest.raises(ValueError, match="DATAMESH_TOKEN"): + OceanumTransfer() + + +def test_oceanum_get(transfer_with_token, mock_fs, tmp_path): + """Test OceanumTransfer.get() calls fs.get().""" + uri = "oceanum://bucket/path/file.nc" + result = transfer_with_token.get(uri, tmp_path, link=False) + + mock_fs.get.assert_called_once() + assert result == tmp_path / "file.nc" + + +def test_oceanum_get_with_name(transfer_with_token, mock_fs, tmp_path): + """Test OceanumTransfer.get() with custom filename.""" + uri = "oceanum://bucket/path/file.nc" + result = transfer_with_token.get(uri, tmp_path, name="custom.nc", link=False) + + mock_fs.get.assert_called_once() + assert result == tmp_path / "custom.nc" + + +def test_oceanum_exists(transfer_with_token, mock_fs): + """Test OceanumTransfer.exists() calls fs.exists().""" + uri = "oceanum://bucket/path/file.nc" + result = transfer_with_token.exists(uri) + + mock_fs.exists.assert_called_once_with("bucket/path/file.nc") + assert result is True + + +def test_oceanum_list(transfer_with_token, mock_fs): + """Test OceanumTransfer.list() calls fs.ls().""" + uri = "oceanum://bucket/path/" + result = transfer_with_token.list(uri) + + mock_fs.ls.assert_called_once_with("bucket/path/", detail=False) + assert result == ["file1.nc", "file2.nc"] + + +def test_oceanum_put(transfer_with_token, mock_fs, tmp_path): + """Test OceanumTransfer.put() calls fs.put().""" + test_file = tmp_path / "test.nc" + test_file.write_text("data") + + uri = "oceanum://bucket/path/test.nc" + result = transfer_with_token.put(test_file, uri) + + mock_fs.put.assert_called_once_with( + str(test_file), "bucket/path/test.nc", recursive=False + ) + assert result == uri + + +def test_oceanum_delete(transfer_with_token, mock_fs): + """Test OceanumTransfer.delete() calls fs.rm().""" + uri = "oceanum://bucket/path/file.nc" + transfer_with_token.delete(uri, recursive=False) + + mock_fs.rm.assert_called_once_with("bucket/path/file.nc", recursive=False) + + +def test_oceanum_delete_recursive(transfer_with_token, mock_fs): + """Test OceanumTransfer.delete() with recursive=True.""" + uri = "oceanum://bucket/path/dir/" + transfer_with_token.delete(uri, recursive=True) + + mock_fs.rm.assert_called_once_with("bucket/path/dir/", recursive=True) + + +def test_oceanum_stat(transfer_with_token, mock_fs): + """Test OceanumTransfer.stat() calls fs.info().""" + uri = "oceanum://bucket/path/file.nc" + result = transfer_with_token.stat(uri) + + mock_fs.info.assert_called_once_with("bucket/path/file.nc") + # Check that result contains expected fields (implementation adds mtime, name, and spreads **info) + assert result["size"] == 1024 + assert result["type"] == "file" + assert "mtime" in result + assert "name" in result diff --git a/tests/test_transfer_registry.py b/tests/test_transfer_registry.py new file mode 100644 index 0000000..feb25e6 --- /dev/null +++ b/tests/test_transfer_registry.py @@ -0,0 +1,163 @@ +import pytest +from pathlib import Path + + +def _import_registry(): + try: + from rompy.transfer import get_transfer, parse_scheme, TransferBase + + return get_transfer, parse_scheme, TransferBase + except Exception as exc: + pytest.skip( + f"ROMPY transfer registry not importable in this environment: {exc}", + allow_module_level=False, + ) + + +# Mock transfer implementations for testing entry-point wiring +class MockFileTransfer(object): + def __init__(self, *a, **k): + pass + + def get(self, uri, destdir, name=None, link=False): + return Path(destdir) / (name or "mock_file.txt") + + def exists(self, uri) -> bool: + return False + + def list(self, uri): + return [] + + def put(self, local_path, uri) -> str: + return uri + + def delete(self, uri: str, recursive: bool = False) -> None: + pass + + def stat(self, uri): + return {} + + +class MockHttpTransfer(object): + def __init__(self, *a, **k): + pass + + def get(self, uri, destdir, name=None, link=False): + return Path(destdir) / (name or "mock_http.txt") + + def exists(self, uri) -> bool: + return False + + def list(self, uri): + return [] + + def put(self, local_path, uri) -> str: + return uri + + def delete(self, uri: str, recursive: bool = False) -> None: + pass + + def stat(self, uri): + return {} + + +class MockEP: + def __init__(self, name: str, cls): + self.name = name + self._cls = cls + + def load(self): + return self._cls + + +def _set_entry_points(monkeypatch, mapping): + eps = [MockEP(name, cls) for name, cls in mapping.items()] + # Patch importlib.metadata.entry_points so the registry uses the mocks + monkeypatch.setattr("importlib.metadata.entry_points", lambda group=None: eps) + # Clear internal registry cache so subsequent lookups reload from mocks + try: + import rompy.transfer.registry as _reg # type: ignore + + _reg._REGISTRY = None # type: ignore[attr-defined] + except Exception: + pass + return eps + + +def test_scheme_normalization_and_get_transfer(monkeypatch): + _set_entry_points( + monkeypatch, + { + "file": MockFileTransfer, + "http": MockHttpTransfer, + }, + ) + + # http URIs should map to the HttpTransfer implementation + get_transfer, parse_scheme, TransferBase = _import_registry() + t = get_transfer("http://example.com/data") + assert isinstance(t, MockHttpTransfer) + + +def test_parse_scheme_normalization_and_file_path(monkeypatch): + # Ensure parse_scheme normalizes various inputs; this relies on the real + # implementation in production. If the fallback is used in this environment, + # this test will still exercise the API surface. + get_transfer, parse_scheme, TransferBase = _import_registry() + s = parse_scheme("HTTP://EXAMPLE.COM/path") + assert isinstance(s, str) + # Expect lowercase scheme for the normalized value + assert s == "http" + + # Empty/scheme-less path should yield the "file" scheme + s2 = parse_scheme("/tmp/file.txt") + assert s2 == "file" + + +def test_duplicate_schemes_fail_fast(monkeypatch): + # Create two entry-points with the same scheme name to simulate a duplicate + _set_entry_points( + monkeypatch, + { + "file1": MockFileTransfer, + "file2": MockFileTransfer, + }, + ) + get_transfer, parse_scheme, TransferBase = _import_registry() + get_transfer, parse_scheme, TransferBase = _import_registry() + with pytest.raises(Exception) as exc: + get_transfer("file:///tmp/test.txt") + # The exact error type/message may vary by implementation; ensure a meaningful + # error is raised and mentions a duplicate scheme + assert any(tok in str(exc.value).lower() for tok in ["duplicate", "scheme"]) + + +def test_unsupported_scheme_error(monkeypatch): + # Only register a single, known scheme and request an unsupported one + _set_entry_points(monkeypatch, {"file": MockFileTransfer}) + get_transfer, parse_scheme, TransferBase = _import_registry() + with pytest.raises(Exception) as exc: + get_transfer("ftp://example.com/resource") + msg = str(exc.value).lower() + # The registry should raise a clear message about missing transfer and available schemes + assert "no transfer registered" in msg + assert "available schemes" in msg + + +def test_get_transfer_accepts_path_and_anypath(monkeypatch): + _set_entry_points(monkeypatch, {"file": MockFileTransfer, "http": MockHttpTransfer}) + # Path input should resolve to the file transfer + get_transfer, parse_scheme, TransferBase = _import_registry() + t1 = get_transfer(Path("/tmp/data.txt")) + assert isinstance(t1, MockFileTransfer) + + # Optional: test AnyPath input if cloudpathlib is available in the runtime + try: + from cloudpathlib import AnyPath # type: ignore + + ap = AnyPath("/tmp/data.txt") + t2 = get_transfer(ap) + assert isinstance(t2, MockFileTransfer) + except Exception: + # If cloudpathlib is unavailable, skip this path gracefully + pass