"""Synthetic deployment diagnostics; no external traffic or personal data.""" import json from pathlib import Path import tempfile import unittest from unittest.mock import Mock, patch from scripts import deploy_check from src.dms_schema import DEFAULT_SCHEMA class FakeMilvus: def __init__(self, **kwargs): self.db = None self.index_count = 3 def list_databases(self, **kwargs): return ['test_qa'] def use_database(self, name, **kwargs): self.db = name def list_collections(self, **kwargs): return ['index_x', 'content_x'] def query(self, name, **kwargs): return [{'count(*)': self.index_count if name == 'index_x' else 5}] class FakeQcc: def search(self, name): return {'candidates': [{'Name': name, 'KeyNo': 'synthetic'}]} def basic(self, company): return {'data': {'KeyNo': company['KeyNo']}} def honors(self, company): return {'complete': True, 'records': [], 'sources': [{}], 'error': None} class FakeDmsSessions: def __init__(self, config): self.config = config def connect(self): return self class FakeDmsCompany: connected = [] def __init__(self, config): self.config = config def connect(self, identity): self.connected.append(identity) return self class DeployCheckTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) (self.root / 'knowledge/prepared/ingestion').mkdir(parents=True) self.manifest = {'complete': True, 'snapshot': 'synthetic', 'counts': {'qa': 3}, 'databases': {'qa': {'database': 'test_qa', 'index_collection': 'index_x', 'content_collection': 'content_x'}}} (self.root / 'knowledge/prepared/ingestion/manifest.json').write_text(json.dumps(self.manifest)) (self.root / 'config').mkdir() self.schema = json.loads(json.dumps(DEFAULT_SCHEMA)) (self.root / 'config/dms_schema.json').write_text(json.dumps(self.schema), encoding='utf-8') (self.root / '.env').write_text('DMS_BASE_URL=https://example.invalid/dms\nDMS_TOKEN=synthetic\n') def test_milvus_compares_published_counts(self): with (patch('src.step2_knowledge_ingest.ingest.config', return_value={ 'MILVUS_URI': 'http://synthetic.invalid', 'MILVUS_TIMEOUT_SECONDS': 2, 'MILVUS_DB_PREFIX': 'test'}), patch('src.step2_knowledge_ingest.ingest.load', return_value=[{'kind': 'qa'}] * 3), patch('src.step2_knowledge_ingest.ingest.content_parts', side_effect=[[0, 1], [0, 1], [0]] * 3)): self._check_milvus_counts() def _check_milvus_counts(self): healthy = deploy_check.milvus(self.root, FakeMilvus) self.assertTrue(healthy['ok']) self.assertEqual(healthy['libraries']['qa']['content_segments'], 5) client = FakeMilvus() client.index_count = 2 changed = deploy_check.milvus(self.root, lambda **kwargs: client) self.assertFalse(changed['ok']) self.assertEqual(changed['libraries']['qa']['index_delta'], -1) self.manifest['databases']['qa']['database'] = 'wrong_qa' (self.root / 'knowledge/prepared/ingestion/manifest.json').write_text(json.dumps(self.manifest)) self.assertEqual(deploy_check.milvus(self.root, FakeMilvus)['libraries']['qa']['code'], 'database_missing_or_mismatched') def test_redis_pings_once_and_closes(self): client = Mock() client.ping.return_value = True with patch('src.step3_question_answer.context.load_context_config', return_value=Mock(url='redis://synthetic.invalid')): self.assertTrue(deploy_check.redis_check(self.root, lambda *args, **kwargs: client)['ok']) client.ping.assert_called_once() client.close.assert_called_once() def test_dms_checks_session_and_company_schema_without_content_reads(self): FakeDmsCompany.connected = [] with patch.dict('os.environ', {}, clear=True): report = deploy_check.dms(self.root, FakeDmsSessions, FakeDmsCompany) self.assertTrue(report['ok']) self.assertEqual(report['scopes']['sessions']['columns'], 3) self.assertEqual(report['scopes']['company']['columns'], 2) self.assertEqual(FakeDmsCompany.connected, [self.schema['columns'][alias]['id'] for alias in ('basic', 'honor')]) def test_dms_uses_configured_column_ids_and_names(self): self.schema['columns']['basic']['id'] = 2888 self.schema['columns']['honor']['id'] = 2886 self.schema['columns']['basic']['name'] = '企业档案' self.schema['columns']['basic']['fields']['c_name'] = 'c_business_name' (self.root / 'config/dms_schema.json').write_text(json.dumps(self.schema), encoding='utf-8') FakeDmsCompany.connected = [] observed = [] class CapturingCompany(FakeDmsCompany): def __init__(self, config): super().__init__(config) observed.append(config) with patch.dict('os.environ', {}, clear=True): report = deploy_check.dms(self.root, FakeDmsSessions, CapturingCompany) self.assertTrue(report['ok']) self.assertEqual(FakeDmsCompany.connected, [2888, 2886]) self.assertEqual(observed[0].scope_names, ('企业档案', '企业荣誉信息')) self.assertEqual(observed[0].schema['columns']['basic']['fields']['c_name'], 'c_business_name') def test_dms_reports_independent_scopes_and_redacts_errors(self): class BrokenSession(FakeDmsSessions): def connect(self): raise RuntimeError('secret-url') with patch.dict('os.environ', {}, clear=True): report = deploy_check.dms(self.root, BrokenSession, FakeDmsCompany) self.assertFalse(report['ok']) self.assertEqual(report['scopes']['sessions']['code'], 'request_failed') self.assertTrue(report['scopes']['company']['ok']) self.assertNotIn('secret-url', json.dumps(report)) def test_deepseek_requires_actual_content(self): config = Mock(model='synthetic', api_key='secret') with patch('src.step1_knowledge_prepare.summarize.load_config', return_value=config): self.assertFalse(deploy_check.deepseek(self.root, lambda payload, cfg: {'choices': []})['ok']) self.assertTrue(deploy_check.deepseek(self.root, lambda payload, cfg: { 'choices': [{'message': {'content': 'OK'}}]})['ok']) def test_qcc_checks_three_interfaces_without_echoing_company(self): report = deploy_check.qcc(self.root, 'synthetic company', FakeQcc()) self.assertTrue(report['ok']) self.assertEqual(set(report['interfaces']), {'886', '410', '1001'}) self.assertNotIn('synthetic company', json.dumps(report)) fake = FakeQcc() fake.search = lambda name: {'candidates': []} report = deploy_check.qcc(self.root, 'synthetic company', fake) self.assertFalse(report['ok']) self.assertEqual(report['interfaces']['410']['code'], 'exact_company_not_found') def test_run_continues_on_failure_and_hides_exception_message(self): with (patch.object(deploy_check, 'milvus', side_effect=RuntimeError('secret-url')), patch.object(deploy_check, 'redis_check', return_value=deploy_check.result(True, 'connected')), patch.object(deploy_check, 'dms', return_value=deploy_check.result(True, 'usable')), patch.object(deploy_check, 'deepseek', return_value=deploy_check.result(True, 'usable')), patch.object(deploy_check, 'qcc', return_value=deploy_check.result(True, 'usable'))): with patch('builtins.print') as printed: self.assertEqual(deploy_check.run(self.root, 'synthetic'), 1) output = printed.call_args.args[0] self.assertNotIn('secret-url', output) self.assertEqual(json.loads(output)['checks']['milvus']['code'], 'request_failed') def test_independent_checks_only_call_selected_service(self): for action, selected in (('check-milvus', 'milvus'), ('check-redis', 'redis'), ('check-dms', 'dms'), ('check-deepseek', 'deepseek'), ('check-qcc', 'qcc')): with self.subTest(action=action): probes = {name: Mock(return_value=deploy_check.result(True, 'usable')) for name in ('milvus', 'redis_check', 'dms', 'deepseek', 'qcc')} with (patch.object(deploy_check, 'milvus', probes['milvus']), patch.object(deploy_check, 'redis_check', probes['redis_check']), patch.object(deploy_check, 'dms', probes['dms']), patch.object(deploy_check, 'deepseek', probes['deepseek']), patch.object(deploy_check, 'qcc', probes['qcc']), patch('builtins.print') as printed): self.assertEqual(deploy_check.run(self.root, 'synthetic', action), 0) self.assertEqual(set(json.loads(printed.call_args.args[0])['checks']), {selected}) for name, probe in probes.items(): self.assertEqual(probe.call_count, int(name == ('redis_check' if selected == 'redis' else selected)))