mirror of
https://github.com/Shu-Ying/main_plugin_farm.git
synced 2026-10-28 05:00:01 +08:00
231 lines
11 KiB
Python
231 lines
11 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from io import BytesIO, StringIO
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from farm_render import bundle
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ZIP = ROOT / 'src/farm_render/data/assets.zip'
|
|
|
|
|
|
class BundleTests(unittest.TestCase):
|
|
def test_manifest_check_without_third_party_packages(self):
|
|
result = subprocess.run(
|
|
[sys.executable, '-S',
|
|
str(ROOT / 'scripts/check_asset_manifest.py')],
|
|
cwd=ROOT, capture_output=True, text=True, timeout=30,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertIn('Verified external asset manifest', result.stdout)
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.expected = bundle.expected_manifest()
|
|
|
|
def download(self, url, destination, limit):
|
|
content = ZIP.read_bytes()
|
|
self.assertEqual(limit, len(content))
|
|
destination.write(content)
|
|
|
|
def test_download_deduplicates_and_cache_hash_checked_each_start(self):
|
|
with patch.object(bundle, '_copy_remote',
|
|
side_effect=self.download) as fetch:
|
|
with ThreadPoolExecutor(4) as pool:
|
|
results = list(pool.map(
|
|
lambda _: bundle.ensure_assets(self.root), range(4),
|
|
))
|
|
self.assertEqual(len(set(results)), 1)
|
|
self.assertEqual(fetch.call_count, 1)
|
|
path = results[0]
|
|
content = bytearray(path.read_bytes())
|
|
content[-1] ^= 1
|
|
path.write_bytes(content)
|
|
bundle.ensure_assets(self.root)
|
|
self.assertEqual(fetch.call_count, 2)
|
|
self.assertEqual(path.read_bytes(), ZIP.read_bytes())
|
|
|
|
def test_gitea_and_service_use_service_repository_and_configured_base(self):
|
|
with patch.object(bundle, '_copy_remote',
|
|
side_effect=self.download) as fetch:
|
|
bundle.ensure_assets(self.root / 'gitea', 'Gitea')
|
|
self.assertEqual(fetch.call_args.args[0],
|
|
'https://gitea.diuse.work/Diuse/farm_service/'
|
|
'raw/branch/main/catalogue/releases/'
|
|
+ self.expected['catalogue_release'] + '/assets.zip')
|
|
bundle.ensure_assets(self.root / 'service', 'farm_service',
|
|
'https://farm.example/prefix/')
|
|
self.assertEqual(fetch.call_args.args[0],
|
|
'https://farm.example/prefix/v1/catalogue/releases/'
|
|
+ self.expected['catalogue_release'] + '/assets.zip')
|
|
|
|
def test_failed_download_leaves_no_partial_and_wrong_hash_is_rejected(self):
|
|
def corrupt(url, destination, limit):
|
|
data = bytearray(ZIP.read_bytes())
|
|
data[-1] ^= 1
|
|
destination.write(data)
|
|
with patch.object(bundle, '_copy_remote', side_effect=corrupt):
|
|
with self.assertRaisesRegex(ValueError, 'SHA256'):
|
|
bundle.ensure_assets(self.root)
|
|
self.assertFalse(list(self.root.rglob('assets.zip')))
|
|
self.assertFalse(list(self.root.rglob('*.download')))
|
|
with patch.object(bundle, '_copy_remote', side_effect=OSError()):
|
|
with self.assertRaises(OSError):
|
|
bundle.ensure_assets(self.root)
|
|
self.assertFalse(list(self.root.rglob('*.download')))
|
|
|
|
def test_invalid_source_and_service_configuration(self):
|
|
for source, url in [('unknown', ''), ('farm_service', ''),
|
|
('farm_service', 'http://farm.example'),
|
|
('Github', 'https://user:pass@farm.example'),
|
|
('Gitea', 'https://farm.example/?token=secret')]:
|
|
with self.subTest(source=source, url=url):
|
|
with self.assertRaises(ValueError):
|
|
bundle.ensure_assets(self.root, source, url)
|
|
|
|
def test_download_enforces_response_limit_and_https_redirect(self):
|
|
class Response(BytesIO):
|
|
headers = {'Content-Length': '20'}
|
|
def geturl(self):
|
|
return 'https://farm.example/assets.zip'
|
|
with patch.object(bundle, 'urlopen', return_value=Response(b'x'*20)):
|
|
with self.assertRaisesRegex(ValueError, 'limit'):
|
|
bundle._copy_remote('https://farm.example/a', BytesIO(), 10)
|
|
response = Response(b'')
|
|
response.geturl = lambda: 'http://farm.example/a'
|
|
with patch.object(bundle, 'urlopen', return_value=response):
|
|
with self.assertRaises(ValueError):
|
|
bundle._copy_remote('https://farm.example/a', BytesIO(), 10)
|
|
|
|
def test_download_refreshes_terminal_progress_on_same_line(self):
|
|
from rich.console import Console
|
|
|
|
class Response(BytesIO):
|
|
headers = {'Content-Length': '262144'}
|
|
def geturl(self):
|
|
return 'https://farm.example/private/assets.zip'
|
|
terminal = StringIO()
|
|
terminal.isatty = lambda: True
|
|
with patch.dict('os.environ', {'TERM': 'xterm-256color'}), \
|
|
patch.object(bundle.sys, 'stderr', terminal), \
|
|
patch('rich.console.Console', side_effect=lambda **kwargs:
|
|
Console(**kwargs, force_terminal=True,
|
|
force_interactive=True, legacy_windows=False,
|
|
color_system='standard', width=120)), \
|
|
patch.object(bundle, 'urlopen', return_value=Response(
|
|
b'x' * 262144)), patch.object(
|
|
bundle, 'time', SimpleNamespace(monotonic=Mock(
|
|
side_effect=[0, 1, 2, 10, 11]))), \
|
|
self.assertLogs(bundle.logger, level='INFO') as captured:
|
|
bundle._copy_remote('https://farm.example/private/assets.zip',
|
|
BytesIO(), 262144)
|
|
messages = '\n'.join(captured.output)
|
|
self.assertIn('下载开始', messages)
|
|
self.assertIn('下载完成', messages)
|
|
self.assertNotIn('下载进度', messages)
|
|
raw = terminal.getvalue()
|
|
progress = re.sub(r'\x1b\[[0-9;?]*[a-zA-Z]', '', raw)
|
|
self.assertIn('0.0%', progress)
|
|
self.assertIn('50.0%', progress)
|
|
self.assertIn('100.0%', progress)
|
|
self.assertIn('eta', progress)
|
|
self.assertIn('kB', progress)
|
|
self.assertIn('/s', progress)
|
|
self.assertGreater(raw.count('\r'), 2)
|
|
self.assertIn('\x1b[2K', raw)
|
|
self.assertNotIn('farm.example', progress)
|
|
self.assertNotIn('farm.example', messages)
|
|
self.assertNotIn('private', messages)
|
|
|
|
def test_unknown_length_and_failed_download_report_truthfully(self):
|
|
class Response(BytesIO):
|
|
headers = {}
|
|
def geturl(self):
|
|
return 'https://farm.example/a'
|
|
output = StringIO()
|
|
with patch.object(bundle.sys, 'stderr', output), \
|
|
patch.object(bundle, 'urlopen', return_value=Response(b'abc')), \
|
|
patch.object(bundle, 'time', SimpleNamespace(
|
|
monotonic=Mock(side_effect=[0, 10]))), \
|
|
self.assertLogs(bundle.logger, level='INFO') as captured:
|
|
bundle._copy_remote('https://farm.example/a', BytesIO(), 10)
|
|
self.assertIn('下载完成', '\n'.join(captured.output))
|
|
self.assertEqual(output.getvalue(), '')
|
|
self.assertNotIn('%', '\n'.join(captured.output))
|
|
with patch.object(bundle, 'urlopen', side_effect=OSError('secret')), \
|
|
self.assertLogs(bundle.logger, level='INFO') as captured:
|
|
with self.assertRaises(OSError):
|
|
bundle._copy_remote('https://farm.example/a', BytesIO(), 10)
|
|
messages = '\n'.join(captured.output)
|
|
self.assertIn('下载失败', messages)
|
|
self.assertNotIn('下载完成', messages)
|
|
self.assertNotIn('secret', messages)
|
|
|
|
def test_terminal_progress_is_throttled_and_unknown_size_has_no_percent(self):
|
|
terminal = StringIO()
|
|
terminal.isatty = lambda: True
|
|
with patch.object(bundle.sys, 'stderr', terminal), \
|
|
patch('rich.progress.Progress') as rich_progress:
|
|
progress = bundle._DownloadProgress('字体包')
|
|
progress.update(0, 0, 0)
|
|
progress.update(65536, 0, 0.1)
|
|
progress.update(131072, 0, 0.2)
|
|
progress.close()
|
|
display = rich_progress.return_value
|
|
self.assertEqual(display.update.call_count, 2)
|
|
self.assertEqual(display.update.call_args.kwargs['completed'], 131072)
|
|
self.assertEqual(display.update.call_args.kwargs['percentage'], '')
|
|
self.assertIsNone(display.add_task.call_args.kwargs['total'])
|
|
self.assertFalse(rich_progress.call_args.kwargs['auto_refresh'])
|
|
self.assertFalse(rich_progress.call_args.kwargs['redirect_stdout'])
|
|
self.assertFalse(rich_progress.call_args.kwargs['redirect_stderr'])
|
|
display.stop.assert_called_once()
|
|
|
|
def test_download_failure_ends_progress_line(self):
|
|
class Response(BytesIO):
|
|
headers = {'Content-Length': '10'}
|
|
|
|
def geturl(self):
|
|
return 'https://farm.example/a'
|
|
|
|
def read(self, size):
|
|
raise OSError('private error')
|
|
|
|
terminal = StringIO()
|
|
terminal.isatty = lambda: True
|
|
with patch.object(bundle.sys, 'stderr', terminal), \
|
|
patch.object(bundle, 'urlopen', return_value=Response()), \
|
|
self.assertLogs(bundle.logger, level='INFO') as captured:
|
|
with self.assertRaises(OSError):
|
|
bundle._copy_remote('https://farm.example/a', BytesIO(), 10)
|
|
self.assertTrue(terminal.getvalue().endswith('\n'))
|
|
self.assertNotIn('100.0%', terminal.getvalue())
|
|
self.assertNotIn('下载完成', '\n'.join(captured.output))
|
|
|
|
def test_progress_display_failure_does_not_abort_download(self):
|
|
class Response(BytesIO):
|
|
headers = {'Content-Length': '3'}
|
|
|
|
def geturl(self):
|
|
return 'https://farm.example/a'
|
|
|
|
terminal = StringIO()
|
|
terminal.isatty = lambda: True
|
|
destination = BytesIO()
|
|
with patch.object(bundle.sys, 'stderr', terminal), \
|
|
patch.object(bundle, 'urlopen', return_value=Response(b'abc')), \
|
|
patch('rich.progress.Progress') as rich_progress:
|
|
display = rich_progress.return_value
|
|
display.refresh.side_effect = OSError('broken terminal')
|
|
bundle._copy_remote('https://farm.example/a', destination, 10)
|
|
self.assertEqual(destination.getvalue(), b'abc')
|
|
display.stop.assert_called_once()
|