import os import unittest from unittest.mock import Mock, patch from qcloud_cos.cos_exception import CosClientError, CosServiceError from app.features.memoir.memoir_images.storage import ( CosDownloadUrlError, CosUploadError, TencentCosStorageService, _is_retryable_cos_error, normalize_cos_url, resolve_image_storage_key, ) class MemoirImageStorageTest(unittest.TestCase): @patch.dict( os.environ, { "TENCENT_COS_SECRET_ID": "id", "TENCENT_COS_SECRET_KEY": "key", "TENCENT_COS_REGION": "ap-shanghai", "TENCENT_COS_BUCKET": "memoir-1250000000", "TENCENT_COS_BASE_URL": "https://memoir-1250000000.cos.ap-shanghai.myqcloud.com", }, clear=False, ) @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_from_env_reuses_singleton_for_same_config(self, client_cls): TencentCosStorageService._instance = None TencentCosStorageService._instance_config = None client_cls.return_value = Mock() first = TencentCosStorageService.from_env() second = TencentCosStorageService.from_env() self.assertIs(first, second) client_cls.assert_called_once() @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_upload_bytes_returns_persistent_cos_url(self, client_cls): client = Mock() client.put_object.return_value = { "ETag": '"abc123"', "x-cos-request-id": "req-001", } client_cls.return_value = client storage = TencentCosStorageService( secret_id="id", secret_key="key", region="ap-shanghai", bucket="memoir-1250000000", base_url="https://memoir-1250000000.cos.ap-shanghai.myqcloud.com", ) url = storage.upload_bytes( image_bytes=b"png-bytes", key="memoirs/u1/c1/0-demo.png", content_type="image/png", ) self.assertEqual( url, "https://memoir-1250000000.cos.ap-shanghai.myqcloud.com/memoirs/u1/c1/0-demo.png", ) client.put_object.assert_called_once() @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_upload_bytes_normalizes_duplicate_appid_suffix_in_base_url(self, client_cls): client = Mock() client_cls.return_value = client storage = TencentCosStorageService( secret_id="id", secret_key="key", region="ap-shanghai", bucket="life-echo-dev-1319381411", base_url="https://life-echo-dev-1319381411-appid.cos.ap-shanghai.myqcloud.com", ) url = storage.upload_bytes( image_bytes=b"png-bytes", key="memoirs/u1/c1/0-demo.png", content_type="image/png", ) self.assertEqual( url, "https://life-echo-dev-1319381411.cos.ap-shanghai.myqcloud.com/memoirs/u1/c1/0-demo.png", ) client.put_object.assert_called_once() def test_normalize_cos_url_repairs_existing_duplicate_appid_host(self): normalized = normalize_cos_url( "https://life-echo-dev-1319381411-appid.cos.ap-shanghai.myqcloud.com/memoirs/u1/c1/0-demo.png", bucket="life-echo-dev-1319381411", region="ap-shanghai", ) self.assertEqual( normalized, "https://life-echo-dev-1319381411.cos.ap-shanghai.myqcloud.com/memoirs/u1/c1/0-demo.png", ) @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_get_download_url_returns_presigned_download_url(self, client_cls): client = Mock() client.get_presigned_download_url.return_value = "https://cos.example.com/0.png?q-sign-algorithm=sha1" client_cls.return_value = client storage = TencentCosStorageService( secret_id="id", secret_key="key", region="ap-shanghai", bucket="life-echo-dev-1319381411", base_url="https://life-echo-dev-1319381411.cos.ap-shanghai.myqcloud.com", ) url = storage.get_download_url("memoirs/u1/c1/0-demo.png", expires=1800) self.assertEqual(url, "https://cos.example.com/0.png?q-sign-algorithm=sha1") client.get_presigned_download_url.assert_called_once_with( Bucket="life-echo-dev-1319381411", Key="memoirs/u1/c1/0-demo.png", Expired=1800, ) def test_resolve_image_storage_key_prefers_explicit_storage_key(self): key = resolve_image_storage_key( { "storage_key": "memoirs/u1/c1/0-demo.png", "url": "https://life-echo-dev-1319381411.cos.ap-shanghai.myqcloud.com/other.png", } ) self.assertEqual(key, "memoirs/u1/c1/0-demo.png") def test_resolve_image_storage_key_derives_key_from_existing_url(self): key = resolve_image_storage_key( { "url": "https://life-echo-dev-1319381411.cos.ap-shanghai.myqcloud.com/memoirs/u1/c1/0-demo.png?q-sign-algorithm=sha1" } ) self.assertEqual(key, "memoirs/u1/c1/0-demo.png") @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_upload_bytes_raises_cos_upload_error_on_client_error(self, client_cls): client = Mock() client.put_object.side_effect = CosClientError("network timeout") client_cls.return_value = client storage = TencentCosStorageService( secret_id="id", secret_key="key", region="ap-shanghai", bucket="memoir-1250000000", base_url="https://memoir-1250000000.cos.ap-shanghai.myqcloud.com", ) with self.assertRaises(CosUploadError) as ctx: storage.upload_bytes(b"data", "key.png", "image/png") self.assertTrue(ctx.exception.retryable) self.assertIsInstance(ctx.exception.__cause__, CosClientError) @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_upload_bytes_raises_non_retryable_on_403(self, client_cls): client = Mock() svc_error = CosServiceError("PUT", "AccessDenied", 403) client.put_object.side_effect = svc_error client_cls.return_value = client storage = TencentCosStorageService( secret_id="id", secret_key="key", region="ap-shanghai", bucket="memoir-1250000000", base_url="https://memoir-1250000000.cos.ap-shanghai.myqcloud.com", ) with self.assertRaises(CosUploadError) as ctx: storage.upload_bytes(b"data", "key.png", "image/png") self.assertFalse(ctx.exception.retryable) @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_get_download_url_raises_cos_download_url_error(self, client_cls): client = Mock() client.get_presigned_download_url.side_effect = CosClientError("dns failure") client_cls.return_value = client storage = TencentCosStorageService( secret_id="id", secret_key="key", region="ap-shanghai", bucket="memoir-1250000000", base_url="https://memoir-1250000000.cos.ap-shanghai.myqcloud.com", ) with self.assertRaises(CosDownloadUrlError) as ctx: storage.get_download_url("key.png") self.assertTrue(ctx.exception.retryable) def test_is_retryable_returns_true_for_client_error(self): self.assertTrue(_is_retryable_cos_error(CosClientError("timeout"))) def test_is_retryable_returns_false_for_4xx_service_error(self): self.assertFalse(_is_retryable_cos_error(CosServiceError("GET", "Forbidden", 403))) def test_is_retryable_returns_true_for_5xx_service_error(self): self.assertTrue(_is_retryable_cos_error(CosServiceError("GET", "Internal", 500))) @patch("app.features.memoir.memoir_images.storage.CosS3Client") def test_cos_config_includes_scheme_and_token(self, client_cls): client_cls.return_value = Mock() with patch("app.features.memoir.memoir_images.storage.CosConfig") as config_cls: TencentCosStorageService( secret_id="id", secret_key="key", region="ap-shanghai", bucket="memoir-1250000000", base_url="https://memoir-1250000000.cos.ap-shanghai.myqcloud.com", token="tmp-token", ) config_cls.assert_called_once_with( Region="ap-shanghai", SecretId="id", SecretKey="key", Token="tmp-token", Scheme="https", )