tests/test_clouddrive_proxy.py
Rama master · réplica de lectura del repositorio original.
"""官方 gRPC 序列化、账号绑定、凭据保留与本机设置路由。"""
import tempfile
from dataclasses import replace
from pathlib import Path
from unittest.mock import patch
from fastapi import FastAPI
from fastapi.testclient import TestClient
from peach import clouddrive_proxy as proxy, downloads_clouddrive_pb2 as pb
from peach.downloads import CLOUDDRIVE_CREDENTIAL, DownloadSettings
from peach.follow_secrets import CredentialStore
from peach.push_discovery import PushDiscoverySettings
from peach.routes_auth import require_auth
from peach.routes_configuration import router
from peach.settings_file import PeachConfig
from test_downloads_clouddrive import FakeCloudDrive, TOKEN, _Server
class ProxyServer(FakeCloudDrive):
def __init__(self):
super().__init__()
self.permissions = pb.TokenPermissions(allow_get_cloud_apis=True, allow_modify_cloud_apis=True,
allow_get_mounts=True)
self.configs = {'one': pb.CloudAPIConfig(maxDownloadThreads=7, defaultReadLengthKB=4096,
apiProxy=pb.ProxyInfo(proxyType=pb.NOPROXY),
dataProxy=pb.ProxyInfo(proxyType=pb.HTTP, host='localhost', port=7890,
username='sample', password='private-proxy-password')),
'two': pb.CloudAPIConfig(dataProxy=pb.ProxyInfo(proxyType=pb.NOPROXY))}
self.configs['one'].MergeFromString(b'\xa0\x06\x07')
def GetApiTokenInfo(self, request, context):
self._auth('GetApiTokenInfo', request, context)
return pb.TokenInfo(permissions=self.permissions)
def GetAllCloudApis(self, request, context):
self._auth('GetAllCloudApis', request, context)
return pb.CloudAPIList(apis=[pb.CloudAPI(name='115open', userName='one', nickName='账号一', path='/drive1'),
pb.CloudAPI(name='115open', userName='two', nickName='账号二', path='/drive2'),
pb.CloudAPI(name='PikPak', userName='unused', nickName='未添加的账号', path='/unused')])
def GetCloudAPIConfig(self, request, context):
self._auth('GetCloudAPIConfig', request, context)
return self.configs[request.userName]
def SetCloudAPIConfig(self, request, context):
self._auth('SetCloudAPIConfig', request, context)
self.configs[request.userName].CopyFrom(request.config)
from google.protobuf.empty_pb2 import Empty
return Empty()
def GetMountPoints(self, request, context):
self._auth('GetMountPoints', request, context)
return pb.GetMountPointsResult(mountPoints=[pb.MountPoint(
mountPoint='B:\\', sourceDir='/drive2', isMounted=True)])
class ProxyTests(_Server):
def setUp(self):
self.fake = ProxyServer()
self.server, self.address = self.serve(self.fake)
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
root = Path(temporary.name).resolve()
self.config = PeachConfig(root, root / 'config.toml', present=True,
locations={'115': ('A:\\Movies', 'B:\\Movies')})
DownloadSettings(self.config.directory('state')).save({'clouddrive_address': self.address}, {})
CredentialStore(self.config.directory('secrets')).save(CLOUDDRIVE_CREDENTIAL, {'token': TOKEN})
PushDiscoverySettings(self.config.directory('state')).save(
{'prefixes': [{'prefix': '/drive1', 'root': 'A:\\Movies'}]}, self.config.locations)
def test_local_libraries_need_no_clouddrive_connection(self):
config = replace(self.config, locations={'local': ('C:\\Movies',)})
with patch.object(proxy, 'client_for') as create_client:
self.assertEqual(proxy.read(config), {'accounts': [], 'unlinked': []})
create_client.assert_not_called()
def test_cloud_libraries_without_clouddrive_address_read_as_not_configured(self):
DownloadSettings(self.config.directory('state')).save({'clouddrive_address': ''}, {})
with patch.object(proxy, 'client_for') as create_client:
self.assertEqual(proxy.read(self.config), {'accounts': [], 'unlinked': [], 'configured': False})
create_client.assert_not_called()
app = FastAPI()
app.include_router(router)
app.dependency_overrides[require_auth] = lambda: {}
with patch('peach.routes_configuration.settings_file.load_config', return_value=self.config), TestClient(
app, base_url='http://127.0.0.1', client=('127.0.0.1', 8000)) as client:
response = client.get('/api/configuration/cloud-proxies')
self.assertEqual(response.status_code, 200, response.text)
self.assertIs(response.json()['configured'], False)
def test_reads_are_fresh_and_passwords_are_not_exposed(self):
state = proxy.read(self.config)
self.assertEqual(len(state['accounts']), 2)
self.assertEqual(state['unlinked'], [])
account = state['accounts'][0]
self.assertTrue(account['data']['password_set'])
self.assertNotIn('private-proxy-password', str(account))
self.fake.configs['one'].dataProxy.proxyType = pb.SOCKS5
updated = proxy.read(self.config)['accounts'][0]
self.assertEqual(updated['data']['mode'], 'socks5')
self.assertNotEqual(account['revision'], updated['revision'])
self.assertNotIn('SetCloudAPIConfig', self.methods())
def test_libraries_share_an_account_and_unlinked_libraries_are_visible(self):
config = replace(self.config, locations={'115': ('A:\\Movies', 'A:\\Movies\\Extra', 'Z:\\Missing')},
library_names={'A:\\Movies': '收藏', 'A:\\Movies\\Extra': '精选'})
result = proxy.read(config)
self.assertEqual(len(result['accounts']), 1)
self.assertEqual([row['name'] for row in result['accounts'][0]['libraries']], ['收藏', '精选'])
self.assertEqual(result['unlinked'][0]['root'], 'Z:\\Missing')
def test_save_preserves_other_fields_unknown_fields_and_password(self):
before = self.fake.configs['one'].SerializeToString()
account = proxy.read(self.config)['accounts'][0]
result = proxy.save(self.config, dict(id=account['id'], revision=account['revision'],
data={**account['data'], 'port': 7891}))
expected = pb.CloudAPIConfig.FromString(before)
expected.dataProxy.port = 7891
self.assertEqual(self.fake.configs['one'], expected)
self.assertEqual(result['data']['port'], 7891)
self.assertNotIn('private-proxy-password', str(result))
self.assertEqual(self.fake.configs['two'].dataProxy.proxyType, pb.NOPROXY)
self.assertEqual(self.methods().count('SetCloudAPIConfig'), 1)
def test_conflict_and_missing_permission_do_not_write(self):
account = proxy.read(self.config)['accounts'][0]
self.fake.configs['one'].maxDownloadThreads = 8
with self.assertRaises(proxy.ProxyConflict):
proxy.save(self.config, {**account, 'data': account['data']})
self.fake.permissions.allow_modify_cloud_apis = False
with self.assertRaisesRegex(ValueError, '修改云盘'):
proxy.save(self.config, {**account, 'data': account['data']})
self.assertNotIn('SetCloudAPIConfig', self.methods())
def test_api_and_data_credentials_can_be_changed_and_cleared_independently(self):
account = proxy.read(self.config)['accounts'][0]
result = proxy.save(self.config, dict(id=account['id'], revision=account['revision'],
api={'mode': 'socks5', 'host': 'localhost', 'port': 1080, 'password': 'new-secret'},
data={**account['data'], 'clear_password': True}))
self.assertEqual(self.fake.configs['one'].apiProxy.password, 'new-secret')
self.assertFalse(result['data']['password_set'])
def test_proxy_host_and_port_validation_prevents_writes(self):
account = proxy.read(self.config)['accounts'][0]
for data in ({'mode': 'http', 'host': 'http://localhost', 'port': 7890},
{'mode': 'socks5', 'host': 'localhost', 'port': 0}, {'mode': 'anything'}):
with self.subTest(data=data), self.assertRaises(ValueError):
proxy.save(self.config, dict(id=account['id'], revision=account['revision'], data=data))
self.assertNotIn('SetCloudAPIConfig', self.methods())
def test_exact_paths_bind_accounts_for_separate_traffic_totals(self):
self.assertEqual(proxy.connections(self.config), {'a:\\movies': 'proxy', 'b:\\movies': 'direct'})
self.fake.configs['two'].dataProxy.proxyType = pb.SYSTEM
self.assertEqual(proxy.connections(self.config)['b:\\movies'], 'system')
self.fake.permissions.allow_get_mounts = False
self.assertEqual(proxy.connections(self.config), {'a:\\movies': 'proxy'})
def test_settings_routes_roundtrip_and_reject_stale_revision(self):
app = FastAPI()
app.include_router(router)
app.dependency_overrides[require_auth] = lambda: {}
with patch('peach.routes_configuration.settings_file.load_config', return_value=self.config), TestClient(
app, base_url='http://127.0.0.1', client=('127.0.0.1', 8000)) as client:
response = client.get('/api/configuration/cloud-proxies')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.headers['cache-control'], 'no-store')
account = response.json()['accounts'][0]
body = dict(id=account['id'], revision=account['revision'], data={'mode': 'direct'})
saved = client.post('/api/configuration/cloud-proxies', json=body)
self.assertEqual(saved.status_code, 200, saved.text)
self.assertEqual(saved.json()['data']['mode'], 'direct')
self.assertEqual(client.post('/api/configuration/cloud-proxies', json=body).status_code, 409)
self.assertNotIn('private-proxy-password', saved.text)