a5d980a5e9
实现独立加密设备身份、网关 agent、持久生命周期任务和 Surge/FlClash 专属订阅凭证。 完善 Shadowrocket 规则转换与 Clash AI DNS，增加 Surge FFSB 适配、规则快照、运维工具及验收文档。
61 lines
2.7 KiB
Python
61 lines
2.7 KiB
Python
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import reconcile
|
|
|
|
|
|
class RefreshLoggingTests(unittest.TestCase):
|
|
def check_refresh(self, manual, download_ok=True, valid=True):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / 'domains.json').write_text('[]')
|
|
(root / 'Proxy Station.conf').write_text('#!MANAGED-CONFIG https://example.invalid/private')
|
|
(root / 'managed-source.conf').write_text('source')
|
|
(root / 'Proxy Station Claude.conf').write_text('candidate')
|
|
(root / 'state.json').write_text(json.dumps({'profile_sha256': hashlib.sha256(b'candidate').hexdigest()}))
|
|
calls = []
|
|
|
|
def run(args, **kwargs):
|
|
calls.append(args)
|
|
if args[0] == '/usr/bin/curl':
|
|
(root / 'download.conf').write_text('source')
|
|
return SimpleNamespace(returncode=0 if download_ok else 1)
|
|
return SimpleNamespace(returncode=0 if valid else 1, stdout='', stderr='')
|
|
|
|
stdout, stderr = io.StringIO(), io.StringIO()
|
|
with patch.object(reconcile, 'ROOT', root), patch.object(reconcile, 'PROFILES', root), patch.object(reconcile, 'build', return_value='candidate'), patch.object(reconcile, 'run', side_effect=run), patch.object(reconcile.subprocess, 'run', side_effect=run), patch.object(reconcile.sys, 'argv', ['reconcile.py'] + (['--refresh'] if manual else [])), contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
|
reconcile.main()
|
|
self.assertFalse(any('reload' in call for call in calls))
|
|
self.assertNotIn('example.invalid', stdout.getvalue() + stderr.getvalue())
|
|
return stdout.getvalue(), stderr.getvalue()
|
|
|
|
def test_manual_unchanged_success(self):
|
|
out, err = self.check_refresh(True)
|
|
self.assertIn('Refresh succeeded; configuration unchanged', out)
|
|
self.assertEqual(err, '')
|
|
|
|
def test_background_unchanged_quiet(self):
|
|
self.assertEqual(self.check_refresh(False), ('', ''))
|
|
|
|
def test_download_failure_does_not_claim_success(self):
|
|
out, err = self.check_refresh(True, download_ok=False)
|
|
self.assertIn('Refresh failed', out)
|
|
self.assertIn('using last validated cache', err)
|
|
self.assertNotIn('succeeded', out)
|
|
|
|
def test_validation_failure_does_not_claim_success(self):
|
|
out, err = self.check_refresh(True, valid=False)
|
|
self.assertIn('Refresh failed', out)
|
|
self.assertIn('using last validated cache', err)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|