kangshifu-board/scripts/test_prepare_voc_handoff.py

"""Regression checks for the exact adapter's risky mapping boundaries."""
import importlib.util
from pathlib import Path
import unittest

spec = importlib.util.spec_from_file_location('handoff', Path(__file__).with_name('prepare-voc-handoff.py'))
h = importlib.util.module_from_spec(spec)
spec.loader.exec_module(h)


class HandoffTests(unittest.TestCase):
    def test_masked_range_and_sentinel_prices_are_not_numeric_prices(self):
        for value in ('7?', '登录查看价格', '0', 0, False, None, '7-9', '-1', '19.999', 'NaN'):
            self.assertFalse(h.valid_price(value), repr(value))
        for value in ('66.90', 14.4, '29.3'):
            self.assertTrue(h.valid_price(value))

    def test_large_identifiers_are_lossless_and_float_ids_are_rejected(self):
        self.assertEqual(h.decimal_id(123456789012345678901), '123456789012345678901')
        for value in (0, False, 1.0, '', '1e9', ' 123'):
            with self.assertRaises(ValueError):
                h.decimal_id(value)

    def test_actual_originals_do_not_repeat_the_review_misreadings(self):
        root = h.ROOT / 'work/voc-handoff-20261106-r1/original'
        if not root.exists():
            self.skipTest('Private handoff originals are not installed')
        files = {p.relative_to(root).as_posix(): p.read_bytes() for p in root.rglob('*') if p.is_file()}
        m = h.Mapping(files)
        products = m.normalize()
        self.assertEqual(len(products), 100)
        self.assertEqual(m.payload['h1']['data']['data']['totalResults'], 165)
        self.assertEqual(m.payload['h1']['data']['data']['TotalPage'], 7)
        jd = [r for name in ('c1', 'c2') for r in m.docs[name]['records']['reviews']]
        self.assertEqual(len({r['values']['reviewId'] for r in jd}), 19)
        self.assertTrue(all(r['values']['itemId'] == '100005702440' for r in jd))
        tb = m.docs['tc1']['records']['reviews']
        self.assertEqual(len({r['values']['reviewId'] for r in tb}), 20)
        self.assertTrue(all(r['values']['reviewId'] != '0' and r['values']['purchasedSku'] for r in tb))
        self.assertTrue(all('isDefault' not in r['values'] for r in jd + tb))
        self.assertTrue(all('shopId' not in r['values'] for r in m.docs['t1']['records']['products']))
        self.assertEqual(len(m.docs['s1b']['records']['prices']), 40)
        self.assertEqual(len(m.docs['d1']['records']['rankings']), 1)
        self.assertEqual(m.docs['d1']['records']['rankings'][0]['values']['scope'], 'detail_page_badge_not_complete_ranking')
        for r in m.docs['d1']['records']['specs']:
            self.assertNotIn('netWeightG', r['values'])
        self.assertFalse(m.docs['d1']['records'].get('prices'))
        for name, doc in m.docs.items():
            self.assertIsNotNone(doc['sourceTimeRaw'])
            for rows in doc['records'].values():
                for row in rows:
                    for field, origin in row['fieldOrigins'].items():
                        if 'pointer' not in origin:
                            self.assertEqual(origin['kind'], 'adapter-policy-not-upstream-field')
                            continue
                        expected = h.agent_batch.pointer(m.payload[name], origin['pointer'])
                        if origin['transform'] == 'lossless-decimal-string':
                            expected = h.decimal_id(expected)
                        self.assertEqual(row['values'][field], expected)


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