姚章浩
51 lines
2.8 KiB
Python
51 lines
2.8 KiB
Python
import io,json,unittest,contextlib
|
|
import live_catalog_refresh as refresh
|
|
|
|
|
|
class S4ProgressPayloadTests(unittest.TestCase):
|
|
def test_source_key_preserves_legacy_directory_identity(self):
|
|
from source_catalog_import import stable_connection_id
|
|
from unittest.mock import patch
|
|
legacy = {'roots': {'company': {'configured_root_id': 'root'}}}
|
|
with patch.object(refresh.pathlib.Path, 'read_text', return_value=json.dumps(legacy)):
|
|
self.assertEqual(refresh.dingtalk_root_key(stable_connection_id('dingtalk-company'), 'space'), 'company')
|
|
self.assertEqual(refresh.dingtalk_root_key(stable_connection_id('dingtalk-space-space'), 'space'), 'space-space')
|
|
with patch.object(refresh.pathlib.Path, 'read_text', return_value=json.dumps(legacy)):
|
|
with self.assertRaises(RuntimeError):
|
|
refresh.dingtalk_root_key('sc-unrecognized', 'space')
|
|
|
|
def collect(self, fn):
|
|
buffer=io.StringIO()
|
|
with contextlib.redirect_stdout(buffer):
|
|
fn()
|
|
return [json.loads(line) for line in buffer.getvalue().splitlines() if line.strip()]
|
|
|
|
def test_progress_lines_are_strict_fields_and_carry_scanned_counts(self):
|
|
lines=self.collect(lambda: refresh.emit('scanning',40,'正在扫描目录：已完成 12 个文件夹，待处理 3 个',12,15,current_directory='/产品资料/CR系列',scanned_directories=12,discovered_directories=15,scanned_files=240,estimated_total_files=520,total_is_estimate=True))
|
|
payload=lines[0]
|
|
self.assertEqual(set(payload),{'stage','progress','message','processed','total','current_directory','scanned_directories','discovered_directories','scanned_files','estimated_total_files','total_is_estimate'})
|
|
self.assertEqual(payload['scanned_files'],240)
|
|
self.assertTrue(payload['total_is_estimate'])
|
|
|
|
def test_sanitizer_strips_control_chars_and_refuses_secrets_urls(self):
|
|
self.assertEqual(refresh.sanitize_directory('/产品\x07资料'),'/产品资料')
|
|
self.assertEqual(refresh.sanitize_directory('https://internal.example/x'),'')
|
|
self.assertEqual(refresh.sanitize_directory('/run/source-secrets/pp'),'')
|
|
self.assertEqual(refresh.sanitize_directory(''),'')
|
|
self.assertEqual(len(refresh.sanitize_directory('目'*2000)),1024)
|
|
|
|
def test_source_name_only_appears_when_set_and_never_paths_with_credentials(self):
|
|
refresh._source_name=''
|
|
plain=self.collect(lambda: refresh.emit('building',80,'正在校验目录结构',5,5))
|
|
self.assertNotIn('source_name',plain[0])
|
|
refresh._source_name='国内工业'
|
|
try:
|
|
named=self.collect(lambda: refresh.emit('building',80,'正在校验目录结构',5,5))
|
|
self.assertEqual(named[0]['source_name'],'国内工业')
|
|
finally:
|
|
refresh._source_name=''
|
|
|
|
|
|
if __name__=='__main__':
|
|
unittest.main()
|