business-assistant/scripts/tests/test_deploy_check.py

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