feat: 添加 WireGuard 设备管理并完善客户端分流
实现独立加密设备身份、网关 agent、持久生命周期任务和 Surge/FlClash 专属订阅凭证。 完善 Shadowrocket 规则转换与 Clash AI DNS,增加 Surge FFSB 适配、规则快照、运维工具及验收文档。
This commit is contained in:
60
scripts/ffsb/test_reconcile.py
Normal file
60
scripts/ffsb/test_reconcile.py
Normal file
@@ -0,0 +1,60 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user