baiyue-firstleg/scripts/client-xlsx-address-probe.test.py

import copy
import importlib.util
import io
from pathlib import Path
import unittest
import zipfile

spec = importlib.util.spec_from_file_location('probe', Path(__file__).with_name('client-xlsx-address-probe.py'))
probe = importlib.util.module_from_spec(spec)
spec.loader.exec_module(probe)


class AddressProbeTests(unittest.TestCase):
    def test_snapshot_requires_complete_tables_and_row_counts(self):
        rows = [{'table': t, 'count': 0} for t in probe.SNAPSHOT_TABLES]
        before = probe.snapshot_summary(rows)
        for bad in [rows[:-1], rows + [rows[0]], rows + [{'table': probe.SNAPSHOT_TABLES[0], 'hash': 'a' * 64}]]:
            with self.assertRaises(AssertionError):
                probe.snapshot_summary(bad)
        changed = copy.deepcopy(rows)
        changed[0]['count'] = 1
        changed.append({'table': probe.SNAPSHOT_TABLES[0], 'hash': 'a' * 64})
        after = probe.snapshot_summary(changed)
        self.assertNotEqual(before, after)
        changed[-1]['hash'] = 'b' * 64
        self.assertNotEqual(after, probe.snapshot_summary(changed))

    def test_snapshot_sql_is_read_only_complete_and_customer_scoped(self):
        columns = {t: ['id', 'tenant_id', 'deleted', 'version'] for t in probe.SNAPSHOT_TABLES}
        sql = probe.snapshot_sql(columns, ['123', '456'])
        self.assertIn('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY', sql)
        self.assertEqual(sql.count('tenant_id IN (123,456)'), 14)
        self.assertNotIn('deleted=0', sql)
        self.assertIn('CAST(`version` AS CHAR)', sql)
        for ids in [['123', '123'], ['123 OR 1=1', '456'], ['123']]:
            with self.assertRaises(AssertionError):
                probe.snapshot_sql(columns, ids)
        columns['ord_order'] = ['id', 'unsafe`field']
        with self.assertRaises(AssertionError):
            probe.snapshot_sql(columns, ['123', '456'])

    def response(self):
        return {'code': '200', 'data': {'mode': 'PARTIAL', 'sourceFileId': 's', 'errorReportFileId': 'e',
                'result': {'total': 1, 'succeeded': 0, 'failed': 1, 'items': [
                    {'success': False, 'clientRowId': 'excel-row-3', 'errorCode': 'A0422',
                     'errorMessage': '取货地址不存在、已停用或不属于当前客户'}]}}}

    def test_precise_address_rejection(self):
        probe.assert_rejection(self.response(), '取货地址')
        for change in [{'success': True}, {'orderId': 'created'}, {'errorCode': 'A0404'},
                       {'clientRowId': 'excel-row-4'}, {'errorMessage': '国内收货仓不能为空'}]:
            changed = copy.deepcopy(self.response())
            changed['data']['result']['items'][0].update(change)
            with self.assertRaises(AssertionError):
                probe.assert_rejection(changed, '取货地址')

    def test_not_setup_or_partial_success(self):
        for part, change in [('root', {'code': 'A0422'}), ('data', {'errorReportFileId': None}),
                             ('result', {'succeeded': 1}), ('result', {'total': 2})]:
            changed = self.response()
            value = changed if part == 'root' else changed['data'] if part == 'data' else changed['data']['result']
            value.update(change)
            with self.assertRaises(AssertionError):
                probe.assert_rejection(changed, '取货地址')

    def test_workbook_preserves_hidden_parts_and_replaces_only_data(self):
        raw = io.BytesIO()
        sheet = '<worksheet><sheetData><row r="1"><c r="A1"/></row><row r="2"><c r="A2"/></row><row r="3"><c r="A3"/></row></sheetData><tail/></worksheet>'
        with zipfile.ZipFile(raw, 'w') as z:
            z.writestr('xl/worksheets/sheet1.xml', sheet)
            z.writestr('xl/worksheets/sheet2.xml', '<foreign>source-options</foreign>')
            z.writestr('xl/workbook.xml', '<names>source-names</names>')
        output = probe.workbook(raw.getvalue(), ['<&'] + [''] * 29 + ['marker'])
        with zipfile.ZipFile(io.BytesIO(output)) as z:
            self.assertEqual(z.read('xl/worksheets/sheet2.xml'), b'<foreign>source-options</foreign>')
            self.assertEqual(z.read('xl/workbook.xml'), b'<names>source-names</names>')
            xml = z.read('xl/worksheets/sheet1.xml').decode()
            self.assertIn('r="AE3"', xml)
            self.assertIn('&lt;&amp;', xml)
            self.assertEqual(xml.count('<row r="3">'), 1)
            self.assertIn('<tail/>', xml)


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