diff --git a/src/quack/cache.py b/src/quack/cache.py index 447d715..4ed608d 100644 --- a/src/quack/cache.py +++ b/src/quack/cache.py @@ -9,12 +9,13 @@ from pathlib import Path from typing import final, override +import zstandard as zstd from loguru import logger from xdg_base_dirs import xdg_cache_home from quack.config import Config from quack.consts import CACHE_METADATA_FILENAME -from quack.exceptions import ChecksumError +from quack.exceptions import CacheCorruptionError, CloudStorageTransientError from quack.models.target import Target from quack.utils.archiver import Archiver from quack.utils.ci_environment import CIEnvironment @@ -68,7 +69,10 @@ def load(self, target: Target) -> None: archive_path = self.get_archive_path(target) size = os.path.getsize(archive_path) logger.info(f"正在从本地加载 Target {target.name} 的缓存（大小：{format_size(size)}）...") - Archiver.extract(archive_path) + try: + Archiver.extract(archive_path) + except zstd.ZstdError as e: + raise CacheCorruptionError(f"缓存归档解压失败：{archive_path}") from e metadata_path = self.get_metadata_path(target) if os.path.exists(metadata_path): os.utime(metadata_path, None) @@ -192,10 +196,13 @@ def exists(self, target: Target) -> bool: def update_access_time(self, target: Target) -> None: """重新上传一次 metadata 文件，来标识其被访问过""" - self.cloud_client.upload( - self.local_backend.get_metadata_path(target), - self.get_metadata_path(target), - ) + try: + self.cloud_client.upload( + self.local_backend.get_metadata_path(target), + self.get_metadata_path(target), + ) + except CloudStorageTransientError as e: + logger.warning(f"更新缓存访问时间失败，将跳过：{e}") def load(self, target: Target, update_access_time: bool = True) -> None: if self.local_backend.exists(target): @@ -204,13 +211,22 @@ def load(self, target: Target, update_access_time: bool = True) -> None: if update_access_time: self.update_access_time(target) return - except ChecksumError: + except CacheCorruptionError: logger.warning("本地缓存已损坏，从云存储重新下载") - - logger.info(f"正在从云存储加载 Target {target.name} 的缓存...") - self.cloud_client.download(self.get_archive_path(target), self.local_backend.get_archive_path(target)) - self.cloud_client.download(self.get_metadata_path(target), self.local_backend.get_metadata_path(target)) - self.local_backend.load(target) + shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) + + try: + logger.info(f"正在从云存储加载 Target {target.name} 的缓存...") + self.cloud_client.download(self.get_archive_path(target), self.local_backend.get_archive_path(target)) + self.cloud_client.download(self.get_metadata_path(target), self.local_backend.get_metadata_path(target)) + self.local_backend.load(target) + except CacheCorruptionError: + logger.warning(f"云存储中 Target {target.name} 的缓存已损坏，将重新生成") + shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) + raise + except CloudStorageTransientError: + shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) + raise if update_access_time: self.update_access_time(target) diff --git a/src/quack/cache_test.py b/src/quack/cache_test.py index 1bb1b10..85a47e6 100644 --- a/src/quack/cache_test.py +++ b/src/quack/cache_test.py @@ -1,8 +1,28 @@ import os from unittest import mock -from quack.cache import TargetCacheBackendTypeCloud +import pytest + +from quack.cache import TargetCacheBackendTypeCloud, TargetCacheBackendTypeLocal from quack.config import Config +from quack.exceptions import CacheCorruptionError, CloudStorageError, CloudStorageTransientError + + +class TestTargetCacheBackendTypeLocal: + def test_load_wraps_corrupt_archive_as_cache_corruption(self, tmp_path, monkeypatch, mock_test_spec: mock.Mock): + monkeypatch.setattr("quack.cache.xdg_cache_home", lambda: tmp_path) + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeLocal(config, mock_test_spec.app_name) + + os.makedirs(backend.get_cache_path(target), exist_ok=True) + with open(backend.get_archive_path(target), "wb") as f: + _ = f.write(b"not a zstd archive") + + with pytest.raises(CacheCorruptionError): + backend.load(target) class TestTargetCacheBackendTypeCloud: @@ -29,6 +49,55 @@ def test_load_exists( # 验证 update_access_time 被调用（上传 metadata） assert mock_cloud_client.upload.called + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_exists_ignores_transient_access_time_update_failure( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client.upload.side_effect = CloudStorageTransientError( + "上传文件失败", + "IncompleteBody", + code="IncompleteBody", + ) + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = True + + backend.load(target) + + mock_local_backend.return_value.load.assert_called_once() + + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_exists_raises_non_transient_access_time_update_failure( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client.upload.side_effect = CloudStorageError("上传文件失败", "AccessDenied", code="AccessDenied") + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = True + + with pytest.raises(CloudStorageError): + backend.load(target) + @mock.patch("quack.cache.CloudClient") @mock.patch("quack.cache.TargetCacheBackendTypeLocal") def test_load_not_exists( @@ -53,6 +122,58 @@ def test_load_not_exists( # 验证本地加载被调用 assert mock_local_backend.return_value.load.call_count == 1 + @mock.patch("quack.cache.shutil.rmtree") + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_corrupt_local_cache_falls_back_to_cloud( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_rmtree: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = True + mock_local_backend.return_value.load.side_effect = [CacheCorruptionError("缓存归档解压失败"), None] + + backend.load(target) + + assert mock_cloud_client.download.call_count == 2 + assert mock_rmtree.called + + @mock.patch("quack.cache.shutil.rmtree") + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_corrupt_cloud_cache_raises( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_rmtree: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = False + mock_local_backend.return_value.load.side_effect = CacheCorruptionError("缓存归档解压失败") + + with pytest.raises(CacheCorruptionError): + backend.load(target) + + assert mock_rmtree.called + @mock.patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}, clear=True) @mock.patch("quack.cache.CloudClient") @mock.patch("quack.cache.TargetCacheBackendTypeLocal") diff --git a/src/quack/exceptions.py b/src/quack/exceptions.py index 5c5d916..f8e12bd 100644 --- a/src/quack/exceptions.py +++ b/src/quack/exceptions.py @@ -1,4 +1,8 @@ -class ChecksumError(Exception): +class CacheCorruptionError(Exception): + pass + + +class ChecksumError(CacheCorruptionError): pass @@ -7,7 +11,12 @@ class ConfigError(Exception): class CloudStorageError(Exception): - def __init__(self, message: str, details: str = ""): + def __init__(self, message: str, details: str = "", code: str | None = None): self.message = message self.details = details + self.code = code super().__init__(f"{message}: {details}" if details else message) + + +class CloudStorageTransientError(CloudStorageError): + pass diff --git a/src/quack/models/target.py b/src/quack/models/target.py index 37e54b5..1898df9 100644 --- a/src/quack/models/target.py +++ b/src/quack/models/target.py @@ -13,7 +13,7 @@ from pydantic import Field from quack.config import Config -from quack.exceptions import CloudStorageError +from quack.exceptions import CacheCorruptionError, CloudStorageTransientError from quack.models.base import BaseModel from quack.models.command import Command from quack.models.dependency import Dependency, DependencyTypeTarget @@ -91,36 +91,49 @@ def execute( logger.info(f"正在执行 Target {self.name}...") logger.info(f"Target {self.name} Checksum 值：{self.checksum_value}") - logger.info(f"正在查找 Target {self.name} 的缓存...") - - cache = TargetCache(config, app_name, self, cache_backend) - cache_exists = cache.hit() if mode == TargetExecutionMode.DEPS_ONLY: self.prepare_deps(config, app_name, cache_backend) - elif mode == TargetExecutionMode.LOAD_ONLY: - if cache_exists: - logger.info("找到缓存，直接从缓存加载...") - cache.load() - else: - logger.error("未找到缓存，无法进行加载") - sys.exit(1) else: - if not cache_exists: - logger.info(f"未找到对应的缓存，开始重新生成缓存：{self.operations.build.command}") - self.prepare_deps(config, app_name, cache_backend) - self.operations.build.execute() - - if cache_exists: - logger.info("找到缓存，直接从缓存加载...") - cache.load() - else: - logger.info(f"正在存入缓存，路径：{self.cache_path}") - try: - cache.save() - except CloudStorageError as e: - logger.error(f"存入缓存失败：{e}") + logger.info(f"正在查找 Target {self.name} 的缓存...") + cache = TargetCache(config, app_name, self, cache_backend) + try: + cache_exists = cache.hit() + except CloudStorageTransientError as e: + if mode == TargetExecutionMode.LOAD_ONLY: + raise + logger.warning(f"缓存命中检查失败，将重新生成 Target {self.name}：{e}") + cache_exists = False + + if mode == TargetExecutionMode.LOAD_ONLY: + if cache_exists: + logger.info("找到缓存，直接从缓存加载...") + cache.load() + else: + logger.error("未找到缓存，无法进行加载") sys.exit(1) + else: + if not cache_exists: + logger.info(f"未找到对应的缓存，开始重新生成缓存：{self.operations.build.command}") + self.prepare_deps(config, app_name, cache_backend) + self.operations.build.execute() + + if cache_exists: + logger.info("找到缓存，直接从缓存加载...") + try: + cache.load() + except (CloudStorageTransientError, CacheCorruptionError) as e: + logger.warning(f"缓存加载失败，将重新生成 Target {self.name}：{e}") + self.prepare_deps(config, app_name, cache_backend) + self.operations.build.execute() + cache_exists = False + + if not cache_exists: + logger.info(f"正在存入缓存，路径：{self.cache_path}") + try: + cache.save() + except CloudStorageTransientError as e: + logger.warning(f"上传缓存失败，将跳过云端缓存：{e}") elapsed = time.time() - start_time logger.success(f"Target {self.name} 执行完毕！") diff --git a/src/quack/models/target_test.py b/src/quack/models/target_test.py index affb76e..258f916 100644 --- a/src/quack/models/target_test.py +++ b/src/quack/models/target_test.py @@ -1,12 +1,65 @@ +from typing import ClassVar, cast from unittest import mock import pytest from pydantic import ValidationError +from quack.cache import TargetCacheBackendType from quack.config import Config +from quack.exceptions import CacheCorruptionError, CloudStorageError, CloudStorageTransientError from quack.models.target import Target, TargetExecutionMode +class RecordingCacheBackend: + NAME: ClassVar[str] = "recording" + exists_result: ClassVar[bool] = False + exists_error: ClassVar[Exception | None] = None + load_error: ClassVar[Exception | None] = None + save_error: ClassVar[Exception | None] = None + exists_calls: ClassVar[int] = 0 + load_calls: ClassVar[int] = 0 + save_calls: ClassVar[int] = 0 + + def __init__(self, _config: Config, _app_name: str) -> None: + pass + + def exists(self, _target: Target) -> bool: + type(self).exists_calls += 1 + error = type(self).exists_error + if error is not None: + raise error + return type(self).exists_result + + def load(self, _target: Target) -> None: + type(self).load_calls += 1 + error = type(self).load_error + if error is not None: + raise error + + def save(self, _target: Target) -> None: + type(self).save_calls += 1 + error = type(self).save_error + if error is not None: + raise error + + +def reset_recording_cache_backend( + *, + exists_result: bool = False, + exists_error: Exception | None = None, + load_error: Exception | None = None, + save_error: Exception | None = None, +) -> type[TargetCacheBackendType]: + RecordingCacheBackend.exists_result = exists_result + RecordingCacheBackend.exists_error = exists_error + RecordingCacheBackend.load_error = load_error + RecordingCacheBackend.save_error = save_error + RecordingCacheBackend.exists_calls = 0 + RecordingCacheBackend.load_calls = 0 + RecordingCacheBackend.save_calls = 0 + return cast(type[TargetCacheBackendType], RecordingCacheBackend) + + class TestTarget: def test_init(self): with pytest.raises(ValidationError) as exc_info: @@ -39,45 +92,154 @@ def test_cache_path(self, mock_test_spec: mock.Mock): def test_cache_archive_filename(self, mock_test_spec: mock.Mock): assert mock_test_spec.targets["quack:test"].cache_archive_filename == "quack:test.tar.zst" - @mock.patch("quack.cache.TargetCache") - def test_execute_deps_only(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_deps_only(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" + cache_backend = reset_recording_cache_backend() # 当 mode=TargetExecutionMode.DEPS_ONLY 仅构建依赖项 - mock_target_cache.return_value.hit.return_value = False target.execute( config, mock_test_spec.app_name, - mock.Mock, + cache_backend, mode=TargetExecutionMode.DEPS_ONLY, ) - @mock.patch("quack.cache.TargetCache") - def test_execute_cache_hit(self, mock_target_cache, mock_test_spec: mock.Mock): + assert RecordingCacheBackend.exists_calls == 0 + assert RecordingCacheBackend.load_calls == 0 + assert RecordingCacheBackend.save_calls == 0 + + def test_execute_cache_hit(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" + cache_backend = reset_recording_cache_backend(exists_result=True) # 当缓存命中时，直接加载缓存 - mock_target_cache.return_value.hit.return_value = True - target.execute(config, mock_test_spec.app_name, mock.Mock) - mock_target_cache.return_value.load.assert_called_once() + target.execute(config, mock_test_spec.app_name, cache_backend) + assert RecordingCacheBackend.load_calls == 1 + + def test_execute_rebuilds_after_transient_cache_hit_failure(self, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_error=CloudStorageTransientError( + "检查文件是否存在失败", + "Could not connect to the endpoint URL", + ) + ) + + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, cache_backend) + + mock_build.assert_called_once() + assert RecordingCacheBackend.load_calls == 0 + assert RecordingCacheBackend.save_calls == 1 + + def test_execute_rebuilds_after_transient_cache_load_failure(self, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_result=True, + load_error=CloudStorageTransientError( + "下载文件失败", + "IncompleteBody", + code="IncompleteBody", + ), + ) + + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, cache_backend) + + assert RecordingCacheBackend.load_calls == 1 + mock_build.assert_called_once() + assert RecordingCacheBackend.save_calls == 1 + + def test_execute_rebuilds_after_corrupt_cache_load(self, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_result=True, + load_error=CacheCorruptionError("缓存归档解压失败"), + ) + + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, cache_backend) + + assert RecordingCacheBackend.load_calls == 1 + mock_build.assert_called_once() + assert RecordingCacheBackend.save_calls == 1 + + def test_execute_raises_non_transient_cache_load_failure(self, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_result=True, + load_error=CloudStorageError( + "下载文件失败", + "AccessDenied", + code="AccessDenied", + ), + ) + + with mock.patch("quack.models.command.Command.execute") as mock_build, pytest.raises(CloudStorageError): + target.execute(config, mock_test_spec.app_name, cache_backend) + + mock_build.assert_not_called() + assert RecordingCacheBackend.save_calls == 0 + + def test_execute_ignores_transient_cache_save_failure(self, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_result=False, + save_error=CloudStorageTransientError( + "上传文件失败", + "IncompleteBody", + code="IncompleteBody", + ), + ) + + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, cache_backend) + + mock_build.assert_called_once() + assert RecordingCacheBackend.save_calls == 1 + + def test_execute_raises_non_transient_cache_save_failure(self, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_result=False, + save_error=CloudStorageError( + "上传文件失败", + "AccessDenied", + code="AccessDenied", + ), + ) + + with mock.patch("quack.models.command.Command.execute"), pytest.raises(CloudStorageError): + target.execute(config, mock_test_spec.app_name, cache_backend) - @mock.patch("quack.cache.TargetCache") - def test_execute_load_only(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_load_only(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" + cache_backend = reset_recording_cache_backend(exists_result=False) # 当 mode=TargetExecutionMode.LOAD_ONLY 且缓存未命中时，应该退出 - mock_target_cache.return_value.hit.return_value = False with pytest.raises(SystemExit): target.execute( config, mock_test_spec.app_name, - mock.Mock, + cache_backend, mode=TargetExecutionMode.LOAD_ONLY, ) diff --git a/src/quack/utils/cloud.py b/src/quack/utils/cloud.py index 53fa551..077c6af 100644 --- a/src/quack/utils/cloud.py +++ b/src/quack/utils/cloud.py @@ -2,15 +2,53 @@ import fnmatch import os +import re from collections.abc import Iterator from dataclasses import dataclass from datetime import datetime +from typing import NoReturn import boto3 +from boto3.exceptions import Boto3Error from botocore.config import Config as BotocoreConfig -from botocore.exceptions import ClientError, NoCredentialsError - -from quack.exceptions import CloudStorageError +from botocore.exceptions import ( + BotoCoreError, + ClientError, + ConnectionClosedError, + ConnectTimeoutError, + EndpointConnectionError, + HTTPClientError, + NoCredentialsError, + ProxyConnectionError, + ReadTimeoutError, +) + +from quack.exceptions import CloudStorageError, CloudStorageTransientError + +TRANSIENT_ERROR_CODES = { + "500", + "503", + "IncompleteBody", + "InternalError", + "RequestTimeout", + "RequestTimeoutException", + "ServiceUnavailable", + "SlowDown", + "Throttling", + "ThrottlingException", + "TooManyRequestsException", +} + +TRANSIENT_NETWORK_ERRORS = ( + ConnectionClosedError, + ConnectTimeoutError, + EndpointConnectionError, + HTTPClientError, + ProxyConnectionError, + ReadTimeoutError, +) + +CLIENT_ERROR_MESSAGE_RE = re.compile(r"An error occurred \(([^)]+)\) when calling the \w+ operation") @dataclass @@ -20,6 +58,57 @@ class CloudFileMetadata: size: int +def _get_client_error_code(error: ClientError) -> str: + code = error.response.get("Error", {}).get("Code", "") + return str(code) + + +def _iter_error_chain(error: Exception) -> Iterator[Exception]: + seen: set[int] = set() + pending = [error] + while pending: + current_error = pending.pop() + if id(current_error) in seen: + continue + + seen.add(id(current_error)) + yield current_error + + pending.extend( + wrapped_error + for wrapped_error in [ + current_error.__context__, + current_error.__cause__, + getattr(current_error, "last_exception", None), + ] + if isinstance(wrapped_error, Exception) + ) + + +def _extract_error_code(error: Exception) -> str: + for current_error in _iter_error_chain(error): + if isinstance(current_error, ClientError): + return _get_client_error_code(current_error) + + match = CLIENT_ERROR_MESSAGE_RE.search(str(current_error)) + if match: + return match.group(1) + + return "" + + +def _is_transient_error(error: Exception) -> bool: + if any(isinstance(current_error, TRANSIENT_NETWORK_ERRORS) for current_error in _iter_error_chain(error)): + return True + return _extract_error_code(error) in TRANSIENT_ERROR_CODES + + +def _raise_cloud_storage_error(message: str, path: str, error: Exception) -> NoReturn: + code = _extract_error_code(error) + exception_class = CloudStorageTransientError if _is_transient_error(error) else CloudStorageError + raise exception_class(f"{message}：{path}", str(error), code=code or None) from error + + class CloudClient: """统一的云存储客户端（支持 OSS 和 S3，使用 boto3）""" @@ -117,7 +206,9 @@ def exists(self, path: str) -> bool: except ClientError as e: if e.response["Error"]["Code"] == "404": return False - raise CloudStorageError(f"检查文件是否存在失败：{path}", str(e)) from e + _raise_cloud_storage_error("检查文件是否存在失败", path, e) + except (Boto3Error, BotoCoreError) as e: + _raise_cloud_storage_error("检查文件是否存在失败", path, e) def upload(self, path: str, dest: str) -> None: """上传文件或目录""" @@ -137,8 +228,8 @@ def upload(self, path: str, dest: str) -> None: self._client.upload_file(local_file, self._bucket_name, object_key) else: raise CloudStorageError(f"路径不存在或不是文件/目录：{path}") - except ClientError as e: - raise CloudStorageError(f"上传文件失败：{path}", str(e)) from e + except (ClientError, Boto3Error, BotoCoreError) as e: + _raise_cloud_storage_error("上传文件失败", path, e) def download(self, path: str, dest: str) -> None: """下载文件或目录""" @@ -175,8 +266,8 @@ def download(self, path: str, dest: str) -> None: if dir_path: os.makedirs(dir_path, exist_ok=True) self._client.download_file(self._bucket_name, key, dest) - except ClientError as e: - raise CloudStorageError(f"下载文件失败：{path}", str(e)) from e + except (ClientError, Boto3Error, BotoCoreError) as e: + _raise_cloud_storage_error("下载文件失败", path, e) def read(self, path: str) -> str | None: """读取文件内容""" diff --git a/src/quack/utils/cloud_test.py b/src/quack/utils/cloud_test.py new file mode 100644 index 0000000..83cc029 --- /dev/null +++ b/src/quack/utils/cloud_test.py @@ -0,0 +1,106 @@ +from pathlib import Path +from unittest import mock + +import pytest +from boto3.exceptions import RetriesExceededError, S3UploadFailedError +from botocore.exceptions import ClientError, EndpointConnectionError, ReadTimeoutError + +from quack.exceptions import CloudStorageError, CloudStorageTransientError +from quack.utils.cloud import CloudClient + + +def _cloud_client_with_upload_error(error: Exception) -> CloudClient: + client = CloudClient.__new__(CloudClient) + client._base_path = "" + client._bucket_name = "bucket" + client._client = mock.Mock() + client._client.upload_file.side_effect = error + return client + + +def _cloud_client_with_download_error(error: Exception) -> CloudClient: + client = CloudClient.__new__(CloudClient) + client._base_path = "" + client._bucket_name = "bucket" + client._client = mock.Mock() + client._client.get_paginator.return_value.paginate.return_value = [{"Contents": [{"Key": "cache.tar.zst"}]}] + client._client.download_file.side_effect = error + return client + + +def _s3_upload_error_from_client_error(code: str) -> S3UploadFailedError: + client_error = ClientError( + {"Error": {"Code": code, "Message": code}}, + "PutObject", + ) + try: + raise S3UploadFailedError("Failed to upload cache.tar.zst") from client_error + except S3UploadFailedError as e: + return e + + +def _s3_upload_error_from_message(code: str) -> S3UploadFailedError: + return S3UploadFailedError( + f"Failed to upload cache.tar.zst: An error occurred ({code}) when calling the PutObject operation: {code}" + ) + + +def test_upload_wraps_incomplete_body_as_transient_error(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + local_file.write_bytes(b"cache") + + client = _cloud_client_with_upload_error(_s3_upload_error_from_client_error("IncompleteBody")) + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.upload(str(local_file), "dest/cache.tar.zst") + + assert exc_info.value.code == "IncompleteBody" + + +def test_upload_extracts_error_code_from_s3_transfer_message(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + local_file.write_bytes(b"cache") + + client = _cloud_client_with_upload_error(_s3_upload_error_from_message("IncompleteBody")) + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.upload(str(local_file), "dest/cache.tar.zst") + + assert exc_info.value.code == "IncompleteBody" + + +def test_upload_keeps_access_denied_as_non_transient_error(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + local_file.write_bytes(b"cache") + + client = _cloud_client_with_upload_error(_s3_upload_error_from_client_error("AccessDenied")) + + with pytest.raises(CloudStorageError) as exc_info: + client.upload(str(local_file), "dest/cache.tar.zst") + + assert not isinstance(exc_info.value, CloudStorageTransientError) + assert exc_info.value.code == "AccessDenied" + + +def test_download_wraps_retries_exceeded_timeout_as_transient_error(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + retry_error = RetriesExceededError(ReadTimeoutError(endpoint_url="https://s3.example")) + client = _cloud_client_with_download_error(retry_error) + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.download("cache.tar.zst", str(local_file)) + + assert exc_info.value.code is None + + +def test_exists_wraps_endpoint_connection_error_as_transient_error(): + client = CloudClient.__new__(CloudClient) + client._base_path = "" + client._bucket_name = "bucket" + client._client = mock.Mock() + client._client.head_object.side_effect = EndpointConnectionError(endpoint_url="https://s3.example") + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.exists("cache-metadata.json") + + assert exc_info.value.code is None