tests/test_downloads_clouddrive.py
Rama master · réplica de lectura del repositorio original.
"""115 云下载经 CloudDrive2 gRPC：起一个进程内的假服务端，走真实的序列化与元数据。
假服务端只实现那八个方法，状态放在内存里。断言落在「Peach 发出去的是什么」上：令牌只在
`authorization` 元数据里，提交前先查目录与配额，已有同一 infohash 的任务就不再提交，
地址留空时只用不带令牌的 `GetSystemInfo` 探测，目标目录留空时按 CloudDrive2 报的云盘类型
从推送前缀里推，只有用户点了才建目录。
"""
from __future__ import annotations
import socket
import unittest
from concurrent import futures
import grpc
from peach import downloads as dl
from peach import downloads_clouddrive as cd
from peach import downloads_clouddrive_pb2 as pb
from peach.push_discovery import CloudPrefix
HASH = "c9e15763f722f23e98a29decdfae341b98d53056"
MAGNET = dl.parse_magnet(f"magnet:?xt=urn:btih:{HASH}&dn=ABC-123")
TOKEN = "token-for-tests"
TARGET = "/115/云下载"
class FakeCloudDrive:
def __init__(self):
self.calls: list[tuple[str, object, str]] = []
self.folders = {TARGET: True}
#: 挂载树里的云盘根 → CloudDrive2 报的云盘类型。没登记的目录都算在 115 上。
self.clouds = {"/115": "115open"}
self.create_error = ""
self.offline: list[pb.OfflineFile] = []
self.quota = pb.OfflineQuotaInfo(total=1500, used=3, left=1497)
self.add_error = ""
self.files: set[str] = set()
self.files_after_refresh: set[str] = set()
self.list_error = None
self.can_manage = True
def _auth(self, name, request, context):
header = dict(context.invocation_metadata()).get("authorization", "")
self.calls.append((name, request, header))
if header != f"Bearer {TOKEN}":
context.abort(grpc.StatusCode.UNAUTHENTICATED, "bad token")
def GetSystemInfo(self, request, context):
header = dict(context.invocation_metadata()).get("authorization", "")
self.calls.append(("GetSystemInfo", request, header))
return pb.CloudDriveSystemInfo(IsLogin=True, UserName="u1", SystemReady=True)
def GetSystemSettings(self, request, context):
self._auth('GetSystemSettings', request, context)
return pb.SystemSettings()
def GetApiTokenInfo(self, request, context):
self._auth("GetApiTokenInfo", request, context)
return pb.TokenInfo(token=request.value, rootDir="/", permissions=pb.TokenPermissions(
allow_add_offline_download=True, allow_list_offline_downloads=True))
def GetAllCloudApis(self, request, context):
self._auth('GetAllCloudApis', request, context)
return pb.CloudAPIList()
def GetCloudAPIConfig(self, request, context):
self._auth('GetCloudAPIConfig', request, context)
return pb.CloudAPIConfig()
def SetCloudAPIConfig(self, request, context):
self._auth('SetCloudAPIConfig', request, context)
from google.protobuf.empty_pb2 import Empty
return Empty()
def GetMountPoints(self, request, context):
self._auth('GetMountPoints', request, context)
return pb.GetMountPointsResult()
def FindFileByPath(self, request, context):
self._auth("FindFileByPath", request, context)
full = f"{request.parentPath.rstrip('/')}/{request.path}"
if full in self.folders:
root = max((root for root in self.clouds if full == root or full.startswith(root + "/")),
key=len, default="/115")
return pb.CloudDriveFile(name=request.path, fullPathName=full, isDirectory=True,
isCloudRoot=full == root, canOfflineDownload=self.folders[full],
supportOfflineDownloadManagement=self.can_manage,
CloudAPI=pb.CloudAPI(name=self.clouds.get(root, "115open"),
userName="u1", path=root))
if full in self.files:
return pb.CloudDriveFile(name=request.path, fullPathName=full)
context.abort(grpc.StatusCode.NOT_FOUND, "no such file")
def GetSubFiles(self, request, context):
self._auth('GetSubFiles', request, context)
if self.list_error is not None:
context.abort(self.list_error, 'listing unavailable')
self.files.update(self.files_after_refresh)
yield pb.SubFilesReply()
def CreateFolder(self, request, context):
self._auth("CreateFolder", request, context)
if self.create_error:
return pb.CreateFolderResult(result=pb.FileOperationResult(
success=False, errorMessage=self.create_error))
full = f"{request.parentPath.rstrip('/')}/{request.folderName}"
self.folders[full] = True
return pb.CreateFolderResult(
folderCreated=pb.CloudDriveFile(name=request.folderName, fullPathName=full, isDirectory=True),
result=pb.FileOperationResult(success=True))
def AddOfflineFiles(self, request, context):
self._auth("AddOfflineFiles", request, context)
if self.add_error:
return pb.FileOperationResult(success=False, errorMessage=self.add_error)
self.offline.append(pb.OfflineFile(name="ABC-123", url=request.urls, infoHash=HASH,
status=pb.OFFLINE_DOWNLOADING, percendDone=12.5))
return pb.FileOperationResult(success=True)
def RemoveOfflineFiles(self, request, context):
self._auth("RemoveOfflineFiles", request, context)
self.offline = [item for item in self.offline if item.infoHash not in request.infoHashes]
return pb.FileOperationResult(success=True)
def ListOfflineFilesByPath(self, request, context):
self._auth("ListOfflineFilesByPath", request, context)
return pb.OfflineFileListResult(offlineFiles=self.offline)
def GetOfflineQuotaInfo(self, request, context):
self._auth("GetOfflineQuotaInfo", request, context)
return self.quota
def closed_address() -> str:
"""本机一个刚放开、没人在听的端口。"""
with socket.socket() as probe:
probe.bind(("127.0.0.1", 0))
port = probe.getsockname()[1]
return f"http://127.0.0.1:{port}"
class _Server(unittest.TestCase):
def setUp(self):
self.fake = FakeCloudDrive()
self.server, self.address = self.serve(self.fake)
self.provider = cd.CloudDriveProvider(self.address, TOKEN)
def serve(self, fake: FakeCloudDrive) -> tuple[grpc.Server, str]:
handlers = {
name: grpc.unary_unary_rpc_method_handler(
getattr(fake, name), request_deserializer=request.FromString,
response_serializer=response.SerializeToString)
for name, (request, response) in cd._METHODS.items()}
handlers['GetSubFiles'] = grpc.unary_stream_rpc_method_handler(
fake.GetSubFiles, request_deserializer=pb.FileRequest.FromString,
response_serializer=pb.SubFilesReply.SerializeToString)
server = grpc.server(futures.ThreadPoolExecutor(max_workers=2))
server.add_generic_rpc_handlers(
(grpc.method_handlers_generic_handler("clouddrive.CloudDriveFileSrv", handlers),))
port = server.add_insecure_port("127.0.0.1:0")
server.start()
self.addCleanup(server.stop, None)
return server, f"http://127.0.0.1:{port}"
def methods(self) -> list[str]:
return [name for name, _request, _header in self.fake.calls]
class SubmitTests(_Server):
def test_cloud_without_task_management_submits_and_waits_for_the_file(self):
self.fake.can_manage = False
self.assertEqual(self.provider.submit(MAGNET, TARGET).state, dl.RUNNING)
self.assertNotIn("ListOfflineFilesByPath", self.methods())
self.assertNotIn("GetOfflineQuotaInfo", self.methods())
task = dl.DownloadTask(1, HASH, "clouddrive:guangya:test", MAGNET.uri, "ABC-123", TARGET, dl.SUBMITTED)
self.assertEqual(self.provider.status(task).state, dl.RUNNING)
self.fake.files.add(f"{TARGET}/ABC-123")
self.assertEqual(self.provider.status(task).state, dl.DONE)
with self.assertRaisesRegex(dl.DownloadError, "到网盘中处理"):
self.provider.cancel(task)
self.assertNotIn("RemoveOfflineFiles", self.methods())
def test_a_submission_checks_the_folder_and_quota_then_adds(self):
status = self.provider.submit(MAGNET, TARGET)
self.assertEqual((status.state, status.adopted), (dl.RUNNING, False))
self.assertEqual(self.methods(), ["FindFileByPath", "ListOfflineFilesByPath",
"GetOfflineQuotaInfo", "AddOfflineFiles"])
add = self.fake.calls[-1][1]
self.assertEqual((add.urls, add.toFolder, add.checkFolderAfterSecs),
(MAGNET.uri, TARGET, cd.CHECK_FOLDER_AFTER_SECS))
quota = self.fake.calls[2][1]
self.assertEqual((quota.cloudName, quota.cloudAccountId, quota.path), ("115open", "u1", TARGET))
def test_the_token_travels_only_in_the_metadata(self):
self.provider.submit(MAGNET, TARGET)
for name, request, header in self.fake.calls:
with self.subTest(name=name):
self.assertEqual(header, f"Bearer {TOKEN}")
if name != "GetApiTokenInfo":
self.assertNotIn(TOKEN.encode(), request.SerializeToString())
def test_a_task_already_in_the_folder_is_adopted_without_adding(self):
self.provider.submit(MAGNET, TARGET)
self.fake.calls.clear()
status = self.provider.submit(MAGNET, TARGET)
self.assertTrue(status.adopted)
self.assertNotIn("AddOfflineFiles", self.methods())
self.assertNotIn("GetOfflineQuotaInfo", self.methods())
def test_an_exhausted_quota_stops_before_adding(self):
self.fake.quota = pb.OfflineQuotaInfo(total=200, used=200, left=0)
with self.assertRaises(dl.DownloadError) as caught:
self.provider.submit(MAGNET, TARGET)
self.assertEqual(caught.exception.failure, "quota")
self.assertNotIn("AddOfflineFiles", self.methods())
def test_the_50038_block_is_flagged(self):
self.fake.add_error = "errno: 50038, 该资源涉嫌违规"
with self.assertRaises(dl.DownloadError) as caught:
self.provider.submit(MAGNET, TARGET)
self.assertEqual((caught.exception.failure, caught.exception.block), ("rejected", True))
def test_a_folder_that_cannot_take_offline_tasks_is_a_configuration_problem(self):
self.fake.folders[TARGET] = False
with self.assertRaises(dl.DownloadError) as caught:
self.provider.submit(MAGNET, TARGET)
self.assertEqual(caught.exception.failure, "config")
with self.assertRaises(dl.DownloadError) as caught:
self.provider.submit(MAGNET, "/115/没有这个目录")
self.assertIn("没有", caught.exception.detail)
def test_a_wrong_token_and_an_unreachable_server_are_told_apart(self):
with self.assertRaises(dl.DownloadError) as caught:
cd.CloudDriveProvider(self.address, "wrong").submit(MAGNET, TARGET)
self.assertEqual(caught.exception.failure, "config")
self.server.stop(None)
with self.assertRaises(dl.DownloadError) as caught:
self.provider.status(dl.DownloadTask(1, HASH, "115", MAGNET.uri, "", TARGET, dl.SUBMITTED))
self.assertEqual(caught.exception.failure, "network")
class TrackingTests(_Server):
def task(self) -> dl.DownloadTask:
return dl.DownloadTask(1, HASH, "115", MAGNET.uri, "ABC-123", TARGET, dl.SUBMITTED)
def test_status_is_matched_by_infohash_or_the_original_link(self):
self.provider.submit(MAGNET, TARGET)
status = self.provider.status(self.task())
self.assertEqual((status.state, status.progress), (dl.RUNNING, 0.125))
self.fake.offline = [pb.OfflineFile(name="ABC-123", url=MAGNET.uri, status=pb.OFFLINE_FINISHED)]
self.assertEqual(self.provider.status(self.task()).state, dl.DONE)
self.fake.offline = []
self.assertEqual(self.provider.status(self.task()).state, dl.MISSING)
def test_landed_looks_for_the_name_in_the_target_folder(self):
self.assertFalse(self.provider.landed(self.task()))
self.fake.files.add(f"{TARGET}/ABC-123")
self.assertTrue(self.provider.landed(self.task()))
def test_landed_refreshes_the_official_directory_cache(self):
self.fake.files_after_refresh.add(f'{TARGET}/ABC-123')
self.assertTrue(self.provider.landed(self.task()))
name, request, header = self.fake.calls[0]
self.assertEqual((name, request.path, request.forceRefresh), ('GetSubFiles', TARGET, True))
self.assertEqual(header, f'Bearer {TOKEN}')
def test_directory_refresh_network_failure_remains_retryable(self):
self.fake.list_error = grpc.StatusCode.UNAVAILABLE
with self.assertRaises(dl.DownloadError) as caught:
self.provider.landed(self.task())
self.assertEqual(caught.exception.failure, 'network')
def test_cancelling_keeps_the_downloaded_files(self):
self.provider.submit(MAGNET, TARGET)
self.provider.cancel(self.task())
remove = [request for name, request, _ in self.fake.calls if name == "RemoveOfflineFiles"][0]
self.assertEqual((list(remove.infoHashes), remove.deleteFiles, remove.cloudName),
([HASH], False, "115open"))
self.assertEqual(self.fake.offline, [])
class CheckTests(_Server):
def test_the_check_reports_permissions_folder_and_quota(self):
report = cd.check(self.address, TOKEN, TARGET)
self.assertEqual(report["missing"], ["取消离线任务"])
self.assertEqual(report["folder"], {"path": TARGET, "can_offline": True, "cloud": "115open"})
self.assertEqual(report["quota"], {"total": 1500, "used": 3, "left": 1497})
self.assertEqual(report["problems"], [])
self.assertEqual(report["address"], self.address)
self.assertNotIn("AddOfflineFiles", self.methods())
self.assertNotIn("GetSystemInfo", self.methods())
def test_without_a_token_nothing_is_called(self):
report = cd.check(self.address, "", TARGET)
self.assertFalse(report["ok"])
self.assertIn("API 令牌", report["problems"][0])
self.assertEqual(self.fake.calls, [])
def test_a_blank_address_takes_the_first_local_address_that_answers(self):
second = FakeCloudDrive()
_server, second_address = self.serve(second)
report = cd.check("", TOKEN, TARGET, local_addresses=(self.address, second_address))
self.assertEqual((report["ok"], report["address"]), (True, self.address))
self.assertEqual(self.methods()[0], "GetSystemInfo")
self.assertEqual(self.fake.calls[0][2], "")
self.assertEqual(second.calls, [])
def test_a_blank_address_falls_through_to_the_second_local_address(self):
report = cd.check("", TOKEN, TARGET, local_addresses=(closed_address(), self.address))
self.assertEqual(report["address"], self.address)
self.assertEqual(report["quota"], {"total": 1500, "used": 3, "left": 1497})
def test_a_blank_address_without_a_token_still_finds_the_address(self):
report = cd.check("", "", TARGET, local_addresses=(self.address,))
self.assertEqual((report["ok"], report["address"]), (False, self.address))
self.assertEqual(self.methods(), ["GetSystemInfo"])
def test_no_local_address_answering_lists_every_address_tried(self):
tried = (closed_address(), closed_address())
with self.assertRaises(dl.DownloadError) as caught:
cd.check("", TOKEN, TARGET, local_addresses=tried)
self.assertEqual(caught.exception.failure, "network")
for address in tried:
self.assertIn(address, caught.exception.detail)
def test_addresses_turn_into_grpc_targets(self):
self.assertEqual(cd.channel_target("http://127.0.0.1"), ("127.0.0.1:19798", False))
self.assertEqual(cd.channel_target("https://192.0.2.10:8443"), ("192.0.2.10:8443", True))
self.assertEqual(cd.channel_target("http://[::1]:19798"), ("[::1]:19798", False))
for address in ("http://localhost:bad-port", "http://localhost:0", "http://localhost:65536"):
with self.subTest(address=address), self.assertRaises(dl.DownloadError) as raised:
cd.channel_target(address)
self.assertEqual(raised.exception.failure, "config")
def test_discovery_returns_only_validated_service_addresses(self):
from unittest.mock import patch
with patch.object(cd, "answers", return_value=True) as answers:
address = cd.discover(("http://localhost:bad-port", "127.0.0.1:19798"))
self.assertEqual(address, "http://127.0.0.1:19798")
answers.assert_called_once_with(address)
self.assertTrue(all(dl.clean_address(address) == address for address in cd.LOCAL_ADDRESSES))
self.assertEqual(cd.split_remote("/115/云下载/"), ("/115", "云下载"))
class SuggestionTests(_Server):
"""前缀名字故意和云盘对不上：`/甲` 是 115 的挂载根，`/115假` 是 PikPak 的。"""
def setUp(self):
super().setUp()
self.fake.clouds.update({"/甲": "115open", "/115假": "PikPak"})
self.fake.folders.update({"/甲": True, "/115假": False})
self.prefixes = (CloudPrefix("/115假", "A:\\"), CloudPrefix("/甲", "B:\\"))
self.roots = {"115": ("B:\\",), "pikpak": ("A:\\",), "local": ("C:\\媒体",)}
def hints(self, *, prefixes=None, roots=None, pikpak_root="", pikpak_account=True) -> cd.Hints:
return cd.Hints(self.prefixes if prefixes is None else prefixes,
self.roots if roots is None else roots, pikpak_root, pikpak_account)
def finds(self) -> list[str]:
return [f"{request.parentPath.rstrip('/')}/{request.path}"
for name, request, _ in self.fake.calls if name == "FindFileByPath"]
def test_a_blank_target_takes_the_folder_under_the_prefix_clouddrive_reports_as_115(self):
self.fake.folders["/甲/云下载"] = True
report = cd.check(self.address, TOKEN, "", hints=self.hints())
self.assertEqual(report["suggested_target"], {"path": "/甲/云下载", "exists": True})
self.assertEqual(report["folder"], {"path": "/甲/云下载", "can_offline": True, "cloud": "115open"})
self.assertEqual(report["quota"], {"total": 1500, "used": 3, "left": 1497})
self.assertEqual(report["suggested_pikpak_root"], "A:\\")
self.assertEqual(report["problems"], [])
def test_a_missing_suggested_folder_is_reported_and_never_created(self):
report = cd.check(self.address, TOKEN, "", hints=self.hints(pikpak_root="A:\\"))
self.assertEqual(report["suggested_target"], {"path": "/甲/云下载", "exists": False})
self.assertEqual((report["folder"], report["quota"], report["problems"]), (None, None, []))
self.assertNotIn("CreateFolder", self.methods())
self.assertNotIn("GetOfflineQuotaInfo", self.methods())
def test_a_declared_115_root_wins_over_another_115_prefix(self):
self.fake.clouds["/乙"] = "115"
self.fake.folders.update({"/乙": True, "/乙/云下载": True, "/甲/云下载": True})
prefixes = (CloudPrefix("/乙", "C:\\媒体"), CloudPrefix("/甲/深一层", "B:\\"))
self.fake.folders["/甲/深一层"] = True
report = cd.check(self.address, TOKEN, "", hints=self.hints(prefixes=prefixes, pikpak_root="A:\\"))
self.assertEqual(report["suggested_target"]["path"], "/甲/深一层/云下载")
def test_without_a_115_prefix_the_problem_points_at_push_discovery(self):
only_pikpak = (CloudPrefix("/115假", "A:\\"),)
report = cd.check(self.address, TOKEN, "", hints=self.hints(prefixes=only_pikpak, pikpak_root="A:\\"))
self.assertIsNone(report["suggested_target"])
self.assertEqual(len(report["problems"]), 1)
self.assertIn("「推送发现」", report["problems"][0])
self.assertIn("/115假", report["problems"][0])
report = cd.check(self.address, TOKEN, "", hints=self.hints(prefixes=(), pikpak_root="A:\\"))
self.assertIn("没有登记云端路径前缀", report["problems"][0])
def test_a_pikpak_prefix_below_the_cloud_root_is_not_the_pikpak_root(self):
self.fake.folders["/115假/影视"] = False
roots = {"115": ("B:\\",), "pikpak": ("A:\\影视",)}
prefixes = (CloudPrefix("/115假/影视", "A:\\影视"), CloudPrefix("/甲", "B:\\"))
report = cd.check(self.address, TOKEN, "/115/云下载", hints=self.hints(prefixes=prefixes, roots=roots))
self.assertEqual(report["suggested_pikpak_root"], "")
self.assertEqual(len(report["problems"]), 1)
self.assertIn("PikPak 根目录对应的媒体目录", report["problems"][0])
def test_without_a_pikpak_account_the_pikpak_root_is_left_alone(self):
report = cd.check(self.address, TOKEN, TARGET, hints=self.hints(prefixes=(), pikpak_account=False))
self.assertEqual((report["suggested_pikpak_root"], report["problems"]), ("", []))
self.assertEqual(self.finds(), [TARGET])
def test_filled_fields_ask_nothing_about_the_prefixes(self):
report = cd.check(self.address, TOKEN, TARGET, hints=self.hints(pikpak_root="A:\\"))
self.assertEqual(self.finds(), [TARGET])
self.assertEqual((report["suggested_target"], report["suggested_pikpak_root"]), (None, ""))
def test_creating_the_folder_happens_only_on_request_and_checks_it_again(self):
report = cd.create_folder(self.address, TOKEN, "/甲/云下载", hints=self.hints(pikpak_root="A:\\"))
created = [request for name, request, _ in self.fake.calls if name == "CreateFolder"]
self.assertEqual([(request.parentPath, request.folderName) for request in created],
[("/甲", "云下载")])
self.assertEqual(report["folder"], {"path": "/甲/云下载", "can_offline": True, "cloud": "115open"})
self.fake.create_error = "没有新建文件夹的权限"
with self.assertRaises(dl.DownloadError) as caught:
cd.create_folder(self.address, TOKEN, "/甲/另一个", hints=self.hints())
self.assertIn("没有新建文件夹的权限", caught.exception.detail)
if __name__ == "__main__":
unittest.main()