business-assistant/scripts/tests/test_qcc_transport.py

"""Production QCC adapter against a local synthetic gateway only."""
import hashlib
import json
import os
from pathlib import Path
import tempfile
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from unittest.mock import patch
from urllib.parse import parse_qs, urlsplit

from src.step3_question_answer import qcc
from scripts import deploy, deploy_check


class QccTransportTests(unittest.TestCase):
    def setUp(self):
        self.calls = []
        self.mode = 'ok'
        test = self
        class Handler(BaseHTTPRequestHandler):
            protocol_version = 'HTTP/1.1'
            def handle(self):
                try:
                    super().handle()
                except (ConnectionAbortedError, ConnectionResetError):
                    pass  # Client deliberately closes immediately on redirects/errors.
            def log_message(self, *_):
                pass
            def do_GET(self):
                params = parse_qs(urlsplit(self.path).query)
                test.calls.append((self.path, params, dict(self.headers)))
                route = params['servertype'][0]
                company = {'KeyNo': 'synthetic-id', 'Name': '合成企业', 'CreditCode': 'synthetic-credit'}
                if route == 'fuzzySearch':
                    result = [company]
                elif route == 'basicDetails':
                    result = company
                    if test.mode == 'identity':
                        result = {**company, 'KeyNo': 'different'}
                else:
                    page = int(params['pageIndex'][0])
                    result = {'VerifyResult': 1, 'Data': [{'Name': 'synthetic'}] * (20 if page == 1 else 1)}
                payload = {'Status': '115' if test.mode == 'auth' else '200', 'Result': result}
                if route == 'honor':
                    payload['Paging'] = {'TotalRecords': 21}
                body = json.dumps(payload).encode()
                self.send_response(302 if test.mode == 'redirect' else 200)
                if test.mode == 'redirect':
                    self.send_header('Location', '/do-not-follow')
                if test.mode == 'chunk':
                    self.send_header('Transfer-Encoding', 'chunked')
                    self.end_headers()
                    self.wfile.write(b'bad-chunk\r\n')
                    self.close_connection = True
                else:
                    self.send_header('Content-Length', str(len(body) + (10 if test.mode == 'short' else 0)))
                    self.end_headers()
                    self.wfile.write(body)
                    if test.mode == 'short':
                        self.close_connection = True
        self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        base = f'http://127.0.0.1:{self.server.server_port}/proxy/?proxyToken=fake%2B%26token&servertype='
        self.config = qcc.Config('synthetic-key', 'synthetic-secret', timeout=2,
            search_url=base + 'fuzzySearch', basic_url=base + 'basicDetails', honor_url=base + 'honor')

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join()

    def test_production_three_interfaces_and_deploy_check(self):
        client = qcc.QccClient(self.config)
        with patch.dict(os.environ, {'HTTP_PROXY': 'http://invalid:1'}):
            report = deploy_check.qcc(Path('.'), '合成企业', client)
        self.assertTrue(report['ok'], report)
        self.assertEqual(report['interfaces']['1001']['records'], 21)
        self.assertEqual(report['interfaces']['1001']['pages'], 2)
        self.assertEqual(len(self.calls), 4)
        for path, params, headers in self.calls:
            self.assertIn('proxyToken=fake%2B%26token&', path)
            self.assertNotIn('合成企业', path)
            self.assertEqual(headers['Connection'], 'keep-alive')
            self.assertEqual(headers['Accept'], '*/*')
            self.assertEqual(params['key'], ['synthetic-key'])
            expected = hashlib.md5(('synthetic-key' + headers['Timespan'] + 'synthetic-secret').encode()).hexdigest().upper()
            self.assertEqual(headers['Token'], expected)
        self.assertEqual(self.calls[1][1]['keyword'], ['合成企业'])

    def test_errors_and_identity_do_not_become_success(self):
        client = qcc.QccClient(self.config)
        for mode, code in [('auth', 'qcc_auth_failed'), ('redirect', 'http_302'),
                           ('short', 'response_incomplete'), ('chunk', 'response_incomplete')]:
            self.mode = mode
            before = len(self.calls)
            with self.assertRaisesRegex(qcc.QccError, '^' + code + '$'):
                client.search('合成企业')
            self.assertEqual(len(self.calls), before + 1)
        self.mode = 'identity'
        with self.assertRaisesRegex(qcc.QccError, 'company_identity_mismatch'):
            client.basic({'KeyNo': 'synthetic-id', 'Name': '合成企业'})

    def test_config_validation_and_environment_isolation(self):
        for suffix in ('&key=x', '&pageIndex=2', '&proxyToken=duplicate'):
            self.assertFalse(qcc.valid_api_url(self.config.search_url + suffix))
        self.assertFalse(qcc.valid_api_url(self.config.search_url.replace('servertype', 'serverType')))
        self.assertTrue(qcc.valid_api_url(qcc.DEFAULT_SEARCH_URL))
        self.assertTrue(qcc.valid_api_url(self.config.search_url.replace('http:', 'https:')))
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / '.env'
            path.write_text('QCC_KEY=synthetic-key\nQCC_SECRET_KEY=synthetic-secret\nQCC_SEARCH_URL=' + self.config.search_url)
            with patch.dict(os.environ, {'DEPLOY_API_ENV_FILE_ONLY': '1', 'QCC_KEY': 'old',
                                        'DEEPSEEK_BASE_URL': 'old', 'DMS_TOKEN': 'preserved'}):
                deploy.apply_api_env_policy()
                self.assertEqual(qcc.load_config(path).key, 'synthetic-key')
                self.assertNotIn('DEEPSEEK_BASE_URL', os.environ)
                self.assertNotIn('DMS_TOKEN', os.environ)


if __name__ == '__main__':
    unittest.main()