"""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_milvus_checks_talent_three_libraries_and_preserves_business_fields(self): self._check_multiple_agents() def test_milvus_talent_differences_do_not_hide_business_report(self): for fault in ('database', 'count', 'collection', 'identity', 'prefix'): with self.subTest(fault=fault): self._check_multiple_agents(fault) def _check_multiple_agents(self, fault=None): prepared = self.root / 'knowledge/prepared-talent' (prepared / 'ingestion').mkdir(parents=True, exist_ok=True) counts = {'article': 2, 'qa': 4, 'policy': 1} manifest = {'complete': True, 'snapshot': 'talent-synthetic', 'counts': counts, 'databases': {kind: {'database': 'talent_knowledge_' + kind, 'index_collection': 'index_x', 'content_collection': 'content_x'} for kind in counts}} (prepared / 'ingestion/manifest.json').write_text(json.dumps(manifest)) data = [{'kind': 'qa'}] * 3 talent = [{'kind': kind if kind != 'article' else 'general', 'knowledge_agent': 'talent', 'data': {}, 'collection': {'article': '人才政策和服务', 'qa': '人才政策问答', 'policy': '青浦区惠企政策事项'}[kind]} for kind, count in counts.items() for _ in range(count)] if fault == 'identity': talent[0] = {**talent[0], 'knowledge_agent': 'business'} clients = [] class MultiMilvus(FakeMilvus): def __init__(self, **kwargs): super().__init__(**kwargs) self.closed, self.queries = False, [] clients.append(self) def list_databases(self, **kwargs): names = ['test_qa'] + ['talent_knowledge_' + kind for kind in counts] return names[:-1] if fault == 'database' else names def list_collections(self, **kwargs): return [] if fault == 'collection' and self.db == 'talent_knowledge_policy' else ['index_x', 'content_x'] def query(self, name, **kwargs): self.queries.append(kwargs['output_fields']) total = 3 if self.db == 'test_qa' else counts[self.db.removeprefix('talent_knowledge_')] if fault == 'count' and self.db == 'talent_knowledge_policy': total -= 1 return [{'count(*)': total}] def close(self): self.closed = True settings = {'MILVUS_URI': 'http://synthetic.invalid', 'MILVUS_TIMEOUT_SECONDS': 2, 'MILVUS_DB_PREFIX': 'test'} if fault == 'prefix': settings['TALENT_MILVUS_DB_PREFIX'] = 'test' with (patch('src.step2_knowledge_ingest.ingest.config', return_value=settings), patch('src.step2_knowledge_ingest.ingest.load', side_effect=lambda path: talent if path == prepared else data), patch('src.step2_knowledge_ingest.ingest.content_parts', return_value=[0])): report = deploy_check.milvus(self.root, MultiMilvus) self.assertTrue(report['agents']['business']['ok']) self.assertEqual(report['expected_total'], 3) self.assertEqual(report['agents']['talent']['ok'], fault is None) self.assertEqual(report['ok'], fault is None) self.assertTrue(all(client.closed for client in clients)) self.assertTrue(all(fields == ['count(*)'] for client in clients for fields in client.queries)) if fault is None: self.assertEqual(report['expected_total_all_agents'], 10) self.assertEqual(report['indexed_total_all_agents'], 10) self.assertEqual(report['agents']['talent']['libraries']['article']['database'], 'talent_knowledge_article') elif fault == 'count': self.assertEqual(report['agents']['talent']['libraries']['policy']['index_delta'], -1) def test_milvus_existing_talent_directory_without_manifest_is_not_ignored(self): (self.root / 'knowledge/prepared-talent').mkdir() 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]])): report = deploy_check.milvus(self.root, FakeMilvus) self.assertTrue(report['agents']['business']['ok']) self.assertFalse(report['ok']) self.assertEqual(report['agents']['talent']['code'], 'manifest_missing') 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)))