tests/test_downloads_channels.py
Rama master · réplica de lectura del repositorio original.
"""媒体库渠道、只读能力探测与缓存的隔离回归。"""
from types import SimpleNamespace
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import Mock, patch
from peach.downloads import DownloadConfig, DownloadError, DownloadTask, Landing, REMOTE_DONE
from peach.downloads_channels import LibraryChannels
from peach.downloads_clouddrive import Folder
from peach.push_discovery import CloudPrefix
from peach.follow_secrets import CredentialError
class LibraryChannelTests(unittest.TestCase):
def test_unavailable_capabilities_leave_phase_and_failure_evidence_without_exposing_credentials(self):
from peach import error_reporting
for operation in ('token_permissions', 'folder_capabilities', 'mount_points'):
with self.subTest(operation=operation), tempfile.TemporaryDirectory() as directory:
self.setUp()
error = DownloadError('config', 'Authorization: Bearer test-only-token')
if operation == 'token_permissions':
self.client.token_info.side_effect = error
elif operation == 'folder_capabilities':
self.client.folder.side_effect = error
else:
self.client.token_info.return_value.permissions.allow_get_mounts = True
self.client.call.side_effect = error
with patch.object(error_reporting, '_log_dir', Path(directory)):
result = self.catalog()
records = [json.loads(line) for path in Path(directory).glob('errors-*.log')
for line in path.read_text(encoding='utf-8').splitlines()]
record = next(row for row in records if row.get('operation') == operation)
self.assertEqual(record['stage'], 'download-channel-probe')
self.assertEqual(record['failure'], 'config')
self.assertTrue(record['exception'])
self.assertNotIn('test-only-token', json.dumps(records))
self.assertNotIn('test-only-token', str(result))
if operation == 'mount_points':
self.assertEqual(len(result['providers']), 1, '可用的路径映射仍应继续探测')
else:
self.assertEqual(result['providers'], [])
self.assertTrue(result['problems'])
def test_a_broken_clouddrive_credential_keeps_a_healthy_pikpak_channel(self):
self.credentials.load.side_effect = CredentialError("test-only-secret-detail")
self.credentials.describe.return_value = {"fields": ["refresh_token"]}
self.service.config = DownloadConfig("http://localhost:bad-port", targets={"pikpak": "/Movies"},
pikpak_root="P:\\")
self.service.landing.declared_roots["pikpak"] = ("P:\\",)
result = self.catalog()
self.assertEqual([row["key"] for row in result["providers"]], ["pikpak"])
self.assertIn("CloudDrive2 令牌无法读取", result["problems"][0])
self.assertNotIn("test-only-secret-detail", str(result))
self.factory.assert_not_called()
def setUp(self):
self.credentials = Mock()
self.credentials.load.return_value = SimpleNamespace(values={"token": "test-only-token"})
self.credentials.describe.return_value = {"fields": []}
self.push = SimpleNamespace(config=SimpleNamespace(prefixes=(
CloudPrefix("/duck/Movies", "G:\\Movies"), CloudPrefix("/drive", "O:\\Movies"))))
self.service = SimpleNamespace(
config=DownloadConfig("http://127.0.0.1:19798"), credentials=self.credentials,
landing=SimpleNamespace(push_discovery=self.push,
declared_roots={"guangya": ("G:\\Movies",), "onedrive": ("O:\\Movies",)}))
self.names = {"G:\\Movies": "我的影片", "O:\\Movies": "文档"}
self.client = Mock()
self.client.__enter__ = Mock(return_value=self.client)
self.client.__exit__ = Mock(return_value=False)
self.client.token_info.return_value = SimpleNamespace(rootDir="/", permissions=SimpleNamespace(
allow_add_offline_download=True, allow_list_offline_downloads=True, allow_modify_offline_downloads=True))
self.client.folder.side_effect = lambda path: Folder(path, path.startswith("/duck"), "cloud", "test-account")
self.factory = Mock(return_value=self.client)
self.clock = 0
self.catalog = LibraryChannels(self.service, lambda: self.names, client=self.factory, clock=lambda: self.clock)
def test_only_declared_cloud_libraries_with_offline_capability_are_listed(self):
result = self.catalog()
self.assertEqual(len(result["providers"]), 1)
channel = result["providers"][0]
self.assertTrue(channel["key"].startswith("clouddrive:guangya:"))
self.assertEqual((channel["source"], channel["label"], channel["target"]),
("guangya", "我的影片 · 光鸭云盘", "/duck/Movies"))
self.assertEqual(result["problems"], [])
self.client.add.assert_not_called()
self.client.create_folder.assert_not_called()
self.client.remove.assert_not_called()
def test_library_removal_and_rename_invalidate_the_cache(self):
key = self.catalog()["providers"][0]["key"]
self.names["G:\\Movies"] = "收藏"
channel = self.catalog()["providers"][0]
self.assertEqual((channel["key"], channel["label"]), (key, "收藏 · 光鸭云盘"))
self.service.landing.declared_roots = {"onedrive": ("O:\\Movies",)}
self.assertEqual(self.catalog()["providers"], [])
def test_submission_only_cloud_is_available_without_a_cancel_action(self):
self.client.folder.side_effect = lambda path: Folder(path, path.startswith("/duck"), "cloud", "test-account", False)
channel = self.catalog()["providers"][0]
self.assertEqual(channel["source"], "guangya")
self.assertFalse(channel["can_cancel"])
self.assertEqual(channel["cloud_root"], "/duck/Movies")
def test_capability_cache_expires_and_credential_changes_invalidate_it(self):
self.catalog()
self.catalog()
self.assertEqual(self.factory.call_count, 1)
self.clock = 31
self.catalog()
self.assertEqual(self.factory.call_count, 2)
self.credentials.load.return_value = SimpleNamespace(values={"token": "test-only-rotated"})
self.catalog()
self.assertEqual(self.factory.call_count, 3)
def test_missing_permissions_or_unavailable_connection_never_enable_a_channel(self):
self.client.token_info.return_value.permissions.allow_add_offline_download = False
self.assertEqual(self.catalog()["providers"], [])
self.client.folder.assert_not_called()
self.clock = 31
self.client.token_info.side_effect = DownloadError("network", "test-only-failure")
result = self.catalog()
self.assertEqual(result["providers"], [])
self.assertTrue(result["problems"])
self.assertNotIn("test-only-token", str(result))
def test_pikpak_direct_channel_requires_its_configured_media_root(self):
self.credentials.describe.return_value = {"fields": ["refresh_token"]}
self.service.config = DownloadConfig(targets={"pikpak": "/Movies"}, pikpak_root="P:\\")
self.assertEqual(self.catalog()["providers"], [])
self.service.landing.declared_roots["pikpak"] = ("P:\\",)
self.names["P:\\"] = "PikPak"
channel = self.catalog()["providers"][0]
self.assertEqual((channel["key"], channel["target"]), ("pikpak", "/Movies"))
def test_saved_target_is_used_only_inside_the_library_mapping(self):
self.service.config.targets["guangya"] = "/duck/Movies/收藏"
channel = self.catalog()["providers"][0]
self.assertEqual(channel["target"], "/duck/Movies/收藏")
self.service.config.targets["guangya"] = "/other/Movies"
self.assertEqual(self.catalog()["providers"][0]["target"], "/duck/Movies")
def test_clouddrive_mount_points_resolve_media_without_a_push_prefix(self):
self.push.config.prefixes = ()
self.service.landing.mounts = {"guangya": ("/mnt/cloud/Movies",)}
self.client.token_info.return_value.permissions.allow_get_mounts = True
self.client.call.return_value = SimpleNamespace(mountPoints=[SimpleNamespace(
mountPoint="/mnt/cloud", sourceDir="/duck", localMount=False, isMounted=True)])
result = self.catalog()
self.assertEqual(result["providers"][0]["target"], "/duck/Movies")
self.assertTrue(result["problems"])
channel = result["providers"][0]
landing = Landing(self.push, declared_roots=self.service.landing.declared_roots)
landing.channel_catalog = self.catalog
task = DownloadTask(1, None, channel["key"], "", "SSIS-519.mkv", channel["target"], REMOTE_DONE)
self.assertEqual(landing.ledger_path(task, self.service.config), ("guangya", "G:\\Movies\\SSIS-519.mkv"))
def test_checks_skip_provider_specific_work_without_its_media_source(self):
from peach.downloads_clouddrive import _inspect, empty_report
from peach.web_downloads import _hints
self.credentials.describe.return_value = {"fields": ["refresh_token"]}
hints = _hints(self.service, {"pikpak_account": True})
self.assertFalse(hints.target_enabled)
self.assertFalse(hints.pikpak_account)
_inspect(empty_report("http://127.0.0.1:19798"), self.client, "", hints)
self.client.folder.assert_not_called()
self.client.quota.assert_not_called()