tests/test_downloads_clouddrive_organize.py
Rama master · réplica de lectura del repositorio original.
"""已声明来源的官方移动：冲突、成员核验、确定性拒绝与状态未知、路径边界。"""
import ssl
import unittest
from pathlib import PureWindowsPath
from types import SimpleNamespace
from unittest.mock import Mock, patch
from urllib.parse import unquote, urlsplit
import httpx
import grpc
from peach.organize_clouddrive import (ACCEPTED, DAV_ORIGIN, REJECTED, MoveNotExecuted, PikPakDav,
UnconfirmedMove)
from peach import organize_clouddrive
from peach.task_retry import transient_delay
MAPPING = organize_clouddrive.PikPakPaths('A:\\', '/Pikpak')
class Status(grpc.RpcError):
def __init__(self, code):
self._code = code
def code(self):
return self._code
class ConfiguredPaths(unittest.TestCase):
def setUp(self):
self.mapping = organize_clouddrive.PikPakPaths('P:\\Library', '/My cloud/Movies', '/Movies')
def test_parent_creation_keeps_server_retry_after_but_stops_for_authentication(self):
for status, delay in ((429, 1200), (503, 1200), (401, None), (403, None)):
with self.subTest(status=status), httpx.Client(transport=httpx.MockTransport(
lambda request, status=status: httpx.Response(status, headers={'Retry-After': '1200'}))) as http:
dav = PikPakDav(http, self.mapping, Mock(), listing=lambda path: [], rename=Mock(), move_remote=Mock())
with self.assertRaises(OSError) as failure:
dav.ensure_parent('P:\\Library\\new\\film.mp4')
self.assertEqual(transient_delay(failure.exception), delay)
def test_certificate_failure_does_not_become_a_retryable_file_error(self):
error = httpx.ConnectError('证书校验失败')
error.__cause__ = ssl.SSLCertVerificationError('证书不可信')
with httpx.Client(transport=httpx.MockTransport(Mock(side_effect=error))) as http:
dav = PikPakDav(http, self.mapping, Mock(), listing=lambda path: [], rename=Mock(), move_remote=Mock())
with self.assertRaises(OSError) as failure:
dav.ensure_parent('P:\\Library\\new\\film.mp4')
self.assertIsNone(transient_delay(failure.exception))
def test_directory_network_failure_retries_but_permission_failure_stops(self):
for code, delay in ((grpc.StatusCode.UNAVAILABLE, 0), (grpc.StatusCode.DEADLINE_EXCEEDED, 0),
(grpc.StatusCode.PERMISSION_DENIED, None)):
with self.subTest(code=code):
stream = Mock()
stream.__iter__ = Mock(side_effect=Status(code))
cloud = Mock(token='test-token')
cloud.channel.unary_stream.return_value.return_value = stream
with self.assertRaises(OSError) as failure:
organize_clouddrive._cloud_listing(cloud, 'P:\\Library', self.mapping)
self.assertEqual(transient_delay(failure.exception), delay)
stream.cancel.assert_called_once()
def test_custom_drive_subdirectory_and_remote_name(self):
self.assertEqual(self.mapping.cloud_path('p:\\library\\人名\\a.mp4'), '/My cloud/Movies/人名/a.mp4')
self.assertEqual(self.mapping.dav_path('P:\\Library\\人名\\a.mp4'), '/Movies/人名/a.mp4')
dav = PikPakDav(Mock(), self.mapping, Mock(), listing=Mock(), rename=Mock(), move_remote=Mock())
self.assertEqual(unquote(dav.url('P:\\Library\\a b.mp4')), DAV_ORIGIN + '/Movies/a b.mp4')
dav.ensure_parent('P:\\Library\\a.mp4')
dav.listing.assert_not_called()
def test_paths_outside_the_declared_directory_are_rejected(self):
for path in ('A:\\a.mp4', 'P:\\LibraryOther\\a.mp4', 'P:\\Library\\..\\a.mp4',
'P:Library\\a.mp4', 'P:\\Library\\a.mp4:secret'):
with self.subTest(path=path), self.assertRaises(ValueError):
self.mapping.cloud_path(path)
def test_mapping_comes_from_the_samples_declared_root(self):
roots = {'pikpak': ('P:\\Library', 'Q:\\Second')}
prefixes = (organize_clouddrive.CloudPrefix('/My cloud/Movies', 'P:\\Library'),
organize_clouddrive.CloudPrefix('/Another account', 'Q:\\Second'))
selected = organize_clouddrive._configured_paths('Q:\\Second\\a.mp4', roots, prefixes)
self.assertEqual((selected.root, selected.remote), ('Q:\\Second', '/Another account'))
for mappings in ((), prefixes + (organize_clouddrive.CloudPrefix('/Duplicate', 'Q:\\Second'),)):
with self.subTest(mappings=mappings), self.assertRaises(ValueError):
organize_clouddrive._configured_paths('Q:\\Second\\a.mp4', roots, mappings)
def test_protocol_requests_use_the_configured_remote_directory(self):
cloud = Mock(token='test-token')
cloud.channel.unary_unary.return_value.return_value = SimpleNamespace(success=True)
organize_clouddrive._move_remote(cloud, 'P:\\Library\\old\\a.mp4',
'P:\\Library\\new\\a.mp4', self.mapping)
request = cloud.channel.unary_unary.return_value.call_args.args[0]
self.assertEqual(list(request.theFilePaths), ['/My cloud/Movies/old/a.mp4'])
self.assertEqual(request.destPath, '/My cloud/Movies/new')
organize_clouddrive._rename_remote(cloud, 'P:\\Library\\old\\a.mp4',
'P:\\Library\\old\\b.mp4', self.mapping)
request = cloud.channel.unary_unary.return_value.call_args.args[0]
self.assertEqual((request.path, request.newName), ('/My cloud/Movies/old/a.mp4', 'b.mp4'))
def test_custom_root_directory_listing_stays_within_mapping(self):
cloud = Mock(token='test-token')
stream = OfficialDirectoryEvidence.Stream([
SimpleNamespace(fullPathName='/My cloud/Movies/a.mp4', size=20, isDirectory=False)])
cloud.channel.unary_stream.return_value.return_value = stream
self.assertEqual(organize_clouddrive._cloud_listing(cloud, 'P:\\Library', self.mapping),
[dict(path='P:\\Library\\a.mp4', size=20, directory=False)])
request = cloud.channel.unary_stream.return_value.call_args.args[0]
self.assertEqual(request.path, '/My cloud/Movies')
self.assertTrue(stream.cancelled)
def test_custom_root_authentication_checks_account_relative_path(self):
cloud = Mock(token='test-token')
cloud.channel.unary_unary.return_value.return_value = SimpleNamespace(
directUrl=DAV_ORIGIN + '/Movies/a.mp4', userAgent='',
additionalHeaders={'Authorization': 'test-credential'})
headers = organize_clouddrive._download_headers(cloud, 'P:\\Library\\a.mp4', self.mapping)
self.assertEqual(headers, {'Authorization': 'test-credential'})
request = cloud.channel.unary_unary.return_value.call_args.args[0]
self.assertEqual(request.path, '/My cloud/Movies/a.mp4')
def test_declared_root_cannot_be_a_move_target(self):
cloud = Mock(token='test-token')
with self.assertRaises(ValueError):
organize_clouddrive._move_remote(cloud, 'P:\\Library\\nested\\Library',
'P:\\Library', self.mapping)
cloud.channel.unary_unary.assert_not_called()
def test_channel_uses_the_configured_webdav_account_subdirectory(self):
from peach import downloads_clouddrive_pb2 as pb
roots = {'pikpak': ('P:\\Library',)}
prefix = organize_clouddrive.CloudPrefix('/Renamed account/Movies', 'P:\\Library')
with patch.object(organize_clouddrive, 'PushDiscoverySettings') as settings, \
patch.object(organize_clouddrive, 'CredentialStore') as credentials, \
patch.object(organize_clouddrive, 'DownloadSettings') as downloads, \
patch.object(organize_clouddrive, 'CloudDriveClient') as client, \
patch.object(organize_clouddrive, '_download_headers', return_value={}) as headers, \
patch.object(organize_clouddrive, 'http_client'), \
patch.object(organize_clouddrive.organize, 'verified_renames'):
settings.return_value.load.return_value = SimpleNamespace(prefixes=(prefix,))
credentials.return_value.load.return_value = SimpleNamespace(values={'token': 'test-token'})
downloads.return_value.load.return_value = SimpleNamespace(clouddrive_address='http://example.test')
cloud = client.return_value.__enter__.return_value
cloud.find.return_value = pb.CloudDriveFile(isDirectory=True,
CloudAPI=pb.CloudAPI(name='WebDAV', path='/Renamed account'))
with organize_clouddrive.pikpak_renames('P:\\Library\\a.mp4', roots) as dav:
self.assertEqual(dav.url('P:\\Library\\a.mp4'), DAV_ORIGIN + '/Movies/a.mp4')
cloud.find.assert_called_once_with('/Renamed account/Movies')
self.assertEqual(headers.call_args.args[2].dav_root, '/Movies')
class OfficialMoves(unittest.TestCase):
def setUp(self):
self.entries = {'A:\\old': True, 'A:\\old\\film.mp4': 20, 'A:\\new': True}
self.requests = []
self.refresh = Mock()
self.http = httpx.Client(transport=httpx.MockTransport(self.handle), follow_redirects=False)
self.addCleanup(self.http.close)
self.dav = PikPakDav(self.http, MAPPING, self.refresh, listing=self.listing,
rename=Mock(side_effect=self.relocate),
move_remote=Mock(side_effect=self.relocate))
def handle(self, request):
path = 'A:' + unquote(urlsplit(str(request.url)).path).replace('/', '\\').rstrip('\\')
self.requests.append((request.method, path))
if request.method != 'MKCOL':
return httpx.Response(405)
self.entries[path] = True
return httpx.Response(201)
def listing(self, path):
if path != 'A:\\' and self.entries.get(path) is not True:
raise FileNotFoundError('官方父目录未列出所需目录')
return [dict(path=name, size=0 if value is True else value, directory=value is True)
for name, value in self.entries.items()
if name != path and str(PureWindowsPath(name).parent) == path]
def relocate(self, source, target):
for name in list(self.entries):
if name == source or name.startswith(source + '\\'):
self.entries[target + name[len(source):]] = self.entries.pop(name)
return ACCEPTED
def test_official_move_preserves_size_and_refreshes_both_parents(self):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.assertEqual(self.entries['A:\\new\\film.mp4'], 20)
self.assertNotIn('A:\\old\\film.mp4', self.entries)
self.dav.move_remote.assert_called_once_with('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.rename.assert_not_called()
self.assertEqual({c.args[0] for c in self.refresh.call_args_list}, {'A:\\old', 'A:\\new'})
def test_target_collision_never_submits_move(self):
self.entries['A:\\new\\film.mp4'] = 40
with self.assertRaises(FileExistsError):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.move_remote.assert_not_called()
self.assertEqual(self.entries['A:\\new\\film.mp4'], 40)
def test_rejected_submission_is_a_failure_with_the_source_in_place(self):
self.dav.move_remote = Mock(return_value=REJECTED)
self.dav.confirm_attempts = 4
with patch.object(organize_clouddrive.time, 'sleep') as sleep, self.assertRaises(MoveNotExecuted):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.move_remote.assert_called_once()
sleep.assert_not_called()
self.assertEqual(self.entries['A:\\old\\film.mp4'], 20)
def test_unchanged_source_after_the_confirmation_budget_is_a_failure(self):
for status in (None, ACCEPTED):
with self.subTest(status=status):
self.dav.move_remote = Mock(return_value=status)
self.dav.confirm_attempts = 4
with patch.object(organize_clouddrive.time, 'sleep') as sleep, \
self.assertRaises(MoveNotExecuted):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.move_remote.assert_called_once()
self.assertEqual(sleep.call_count, 3)
self.assertEqual(self.entries['A:\\old\\film.mp4'], 20)
def test_rename_acknowledgement_without_path_change_is_a_failure(self):
self.dav.rename = Mock(return_value=ACCEPTED)
with self.assertRaises(MoveNotExecuted):
self.dav.move('A:\\old\\film.mp4', 'A:\\old\\title.mp4')
self.dav.rename.assert_called_once()
self.assertEqual(self.entries['A:\\old\\film.mp4'], 20)
def test_same_parent_rename_uses_the_official_rename(self):
self.dav.move('A:\\old\\film.mp4', 'A:\\old\\title.mp4')
self.dav.rename.assert_called_once_with('A:\\old\\film.mp4', 'A:\\old\\title.mp4')
self.dav.move_remote.assert_not_called()
self.assertEqual(self.entries['A:\\old\\title.mp4'], 20)
def test_collection_move_checks_all_members(self):
self.dav.move('A:\\old', 'A:\\new\\old')
self.assertEqual(self.entries['A:\\new\\old\\film.mp4'], 20)
self.assertNotIn('A:\\old', self.entries)
def test_incomplete_collection_stops_without_repeating_move(self):
def lossy(source, target):
self.relocate(source, target)
self.entries.pop(target + '\\film.mp4')
return ACCEPTED
self.dav.move_remote = Mock(side_effect=lossy)
with self.assertRaises(UnconfirmedMove):
self.dav.move('A:\\old', 'A:\\new\\old')
self.dav.move_remote.assert_called_once()
def test_source_and_target_both_present_stop_the_batch(self):
def copy(source, target):
self.entries[target] = self.entries[source]
return ACCEPTED
self.dav.move_remote = Mock(side_effect=copy)
with self.assertRaises(UnconfirmedMove):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.move_remote.assert_called_once()
def test_source_and_target_both_absent_stop_the_batch(self):
def lose(source, target):
self.entries.pop(source)
return None
self.dav.move_remote = Mock(side_effect=lose)
with self.assertRaises(UnconfirmedMove):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.move_remote.assert_called_once()
def test_empty_collection_requires_destination_entry(self):
del self.entries['A:\\old\\film.mp4']
self.dav.move('A:\\old', 'A:\\new\\old')
self.assertIs(self.entries['A:\\new\\old'], True)
def test_nested_parent_is_created_in_order(self):
self.dav.move('A:\\old\\film.mp4', 'A:\\deep\\nested\\film.mp4')
self.assertEqual([r[1] for r in self.requests if r[0] == 'MKCOL'], ['A:\\deep', 'A:\\deep\\nested'])
def test_cross_drive_is_rejected_before_network(self):
with self.assertRaises(ValueError):
self.dav.move('A:\\old\\film.mp4', 'R:\\film.mp4')
self.assertFalse(self.requests)
def test_cross_parent_rename_is_rejected_before_submission(self):
with self.assertRaises(ValueError):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\title.mp4')
self.dav.move_remote.assert_not_called()
self.dav.rename.assert_not_called()
self.assertFalse(self.requests)
def test_unicode_path_is_utf8_percent_encoded(self):
self.assertEqual(self.dav.url('A:\\创作者\\a b.mp4'), DAV_ORIGIN + '/%E5%88%9B%E4%BD%9C%E8%80%85/a%20b.mp4')
def test_creator_handle_keeps_the_literal_at_sign(self):
self.assertEqual(self.dav.url('A:\\old\\creator@handle'), DAV_ORIGIN + '/old/creator@handle')
def test_delayed_listing_only_rechecks_the_single_submitted_move(self):
self.dav.move_remote = Mock(return_value=ACCEPTED)
self.dav.confirm_attempts = 4
reads = 0
def refresh(path):
nonlocal reads
reads += 1
if reads == 3:
self.entries['A:\\new\\film.mp4'] = self.entries.pop('A:\\old\\film.mp4')
self.dav.refresh = refresh
with patch.object(organize_clouddrive.time, 'sleep'):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.move_remote.assert_called_once()
self.assertEqual(self.entries['A:\\new\\film.mp4'], 20)
def test_cache_failure_after_submission_preserves_intent(self):
self.dav.move_remote = Mock(return_value=ACCEPTED)
self.dav.refresh = Mock(side_effect=OSError('未取得'))
with self.assertRaises(UnconfirmedMove):
self.dav.move('A:\\old\\film.mp4', 'A:\\new\\film.mp4')
self.dav.move_remote.assert_called_once()
class OfficialMoveProtocol(unittest.TestCase):
def test_conflict_skip_and_false_optional_flags_are_encoded(self):
cloud = Mock(token='test-token')
cloud.channel.unary_unary.return_value.return_value = SimpleNamespace(success=True)
self.assertEqual(organize_clouddrive._move_remote(cloud, 'A:\\old\\film.mp4', 'A:\\new\\film.mp4', MAPPING), ACCEPTED)
request = cloud.channel.unary_unary.return_value.call_args.args[0]
self.assertEqual(list(request.theFilePaths), ['/Pikpak/old/film.mp4'])
self.assertEqual(request.destPath, '/Pikpak/new')
self.assertEqual(request.conflictPolicy, 2)
for flag in ('moveAcrossClouds', 'handleConflictRecursively'):
self.assertTrue(request.HasField(flag))
self.assertFalse(getattr(request, flag))
def test_path_boundaries_stop_before_rpc(self):
cloud = Mock(token='test-token')
for target in ('R:\\film.mp4', 'A:\\new\\other.mp4', 'A:\\old\\film.mp4', 'A:\\new\\..\\film.mp4'):
with self.subTest(target=target), self.assertRaises(ValueError):
organize_clouddrive._move_remote(cloud, 'A:\\old\\film.mp4', target, MAPPING)
for target in ('A:\\new\\title.mp4', 'R:\\old\\title.mp4'):
with self.subTest(target=target), self.assertRaises(ValueError):
organize_clouddrive._rename_remote(cloud, 'A:\\old\\film.mp4', target, MAPPING)
cloud.channel.unary_unary.assert_not_called()
def test_rpc_timeout_has_no_automatic_resubmission(self):
for submit in (organize_clouddrive._move_remote, organize_clouddrive._rename_remote):
with self.subTest(submit=submit.__name__):
cloud = Mock(token='test-token')
cloud.channel.unary_unary.return_value.side_effect = Status(grpc.StatusCode.DEADLINE_EXCEEDED)
target = 'A:\\new\\film.mp4' if submit is organize_clouddrive._move_remote else 'A:\\old\\title.mp4'
self.assertIsNone(submit(cloud, 'A:\\old\\film.mp4', target, MAPPING))
cloud.channel.unary_unary.return_value.assert_called_once()
def test_request_rejections_are_deterministic(self):
for code in (grpc.StatusCode.INVALID_ARGUMENT, grpc.StatusCode.PERMISSION_DENIED):
for submit, target in ((organize_clouddrive._move_remote, 'A:\\new\\film.mp4'),
(organize_clouddrive._rename_remote, 'A:\\old\\title.mp4')):
with self.subTest(code=code, submit=submit.__name__):
cloud = Mock(token='test-token')
cloud.channel.unary_unary.return_value.side_effect = Status(code)
self.assertEqual(submit(cloud, 'A:\\old\\film.mp4', target, MAPPING), REJECTED)
def test_rename_reply_maps_to_submission_status(self):
for success, expected in ((True, ACCEPTED), (False, REJECTED)):
with self.subTest(success=success):
cloud = Mock(token='test-token')
cloud.channel.unary_unary.return_value.return_value = SimpleNamespace(success=success)
self.assertEqual(organize_clouddrive._rename_remote(cloud, 'A:\\old\\film.mp4', 'A:\\old\\title.mp4', MAPPING), expected)
request = cloud.channel.unary_unary.return_value.call_args.args[0]
self.assertEqual((request.path, request.newName), ('/Pikpak/old/film.mp4', 'title.mp4'))
def test_cache_scope_rejects_another_drive(self):
cloud = Mock(token='test-token')
with self.assertRaises(ValueError):
organize_clouddrive._refresh_listing(cloud, 'B:\\MVP', MAPPING)
cloud.channel.unary_unary.assert_not_called()
class OfficialDirectoryEvidence(unittest.TestCase):
class Stream:
def __init__(self, rows):
self.rows, self.cancelled = rows, False
def __iter__(self):
yield SimpleNamespace(subFiles=self.rows)
def cancel(self):
self.cancelled = True
def setUp(self):
self.paths, self.streams = [], []
self.entries = {'/Pikpak': [('old', True, 0)], '/Pikpak/old': [('creator@handle', True, 0)],
'/Pikpak/old/creator@handle': [('film.mp4', False, 20)]}
def read(request, **kwargs):
self.paths.append((request.path, request.forceRefresh))
rows = [SimpleNamespace(fullPathName=request.path + '/' + name, isDirectory=directory, size=size)
for name, directory, size in self.entries[request.path]]
stream = self.Stream(rows)
self.streams.append(stream)
return stream
self.cloud = Mock(token='test-token')
self.cloud.channel.unary_stream.return_value = read
def test_ancestor_listings_confirm_the_complete_unicode_directory(self):
rows = organize_clouddrive._cloud_listing(self.cloud, 'A:\\old\\creator@handle', MAPPING)
self.assertEqual(rows, [dict(path='A:\\old\\creator@handle\\film.mp4', directory=False, size=20)])
self.assertEqual(self.paths, [('/Pikpak', True), ('/Pikpak/old', True), ('/Pikpak/old/creator@handle', True)])
self.assertTrue(all(stream.cancelled for stream in self.streams))
def test_missing_ancestor_does_not_read_an_unverified_child(self):
self.entries['/Pikpak/old'] = []
with self.assertRaises(FileNotFoundError):
organize_clouddrive._cloud_listing(self.cloud, 'A:\\old\\creator@handle', MAPPING)
self.assertEqual(len(self.paths), 2)
def test_foreign_drive_never_uses_the_existing_credential(self):
with self.assertRaises(ValueError):
organize_clouddrive._cloud_listing(self.cloud, 'R:\\Media', MAPPING)
self.assertFalse(self.paths)
def test_duplicate_members_stop_and_cancel_the_stream(self):
self.entries['/Pikpak/old/creator@handle'] *= 2
with self.assertRaises(ValueError):
organize_clouddrive._cloud_listing(self.cloud, 'A:\\old\\creator@handle', MAPPING)
self.assertTrue(all(stream.cancelled for stream in self.streams))
class AuthenticationSamples(unittest.TestCase):
class Missing(grpc.RpcError):
def code(self):
return grpc.StatusCode.NOT_FOUND
def test_moved_sample_uses_existing_catalog_sample(self):
cloud = Mock(token='test-token')
reply = SimpleNamespace(directUrl=DAV_ORIGIN + '/available/image.jpg', userAgent='', additionalHeaders={'Authorization': 'test-credential'})
stub = cloud.channel.unary_unary.return_value
stub.side_effect = [self.Missing(), reply]
with patch.object(organize_clouddrive, '_fallback_samples', return_value=['A:\\available\\image.jpg']):
headers = organize_clouddrive._download_headers(cloud, 'A:\\moved\\image.jpg', MAPPING)
self.assertEqual(headers, {'Authorization': 'test-credential'})
self.assertEqual([c.args[0].path for c in stub.call_args_list], ['/Pikpak/moved/image.jpg', '/Pikpak/available/image.jpg'])
def test_missing_samples_stop_after_three_calls(self):
cloud = Mock(token='test-token')
stub = cloud.channel.unary_unary.return_value
stub.side_effect = self.Missing()
with patch.object(organize_clouddrive, '_fallback_samples', return_value=['A:\\first.jpg', 'A:\\second.jpg']), self.assertRaises(ValueError):
organize_clouddrive._download_headers(cloud, 'A:\\moved.jpg', MAPPING)
self.assertEqual(stub.call_count, 3)
def test_foreign_download_source_is_rejected(self):
cloud = Mock(token='test-token')
cloud.channel.unary_unary.return_value.return_value = SimpleNamespace(directUrl='https://example.test/image.jpg', additionalHeaders={}, userAgent='')
with self.assertRaises(ValueError):
organize_clouddrive._download_headers(cloud, 'A:\\image.jpg', MAPPING)