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()