kangshifu-board/scripts/test_enrich_research_library.py

import importlib.util
from pathlib import Path
import unittest

spec = importlib.util.spec_from_file_location('enrich', Path(__file__).with_name('enrich-research-library.py'))
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)

class ProvenanceTests(unittest.TestCase):
    def doc(self, d): return {'data': {'code': 0, 'data': d, 'recordTime': '2026-11-10T12:00:00'}}
    def test_exact_review_identity_and_large_id(self):
        rows, excluded = m.reviews(self.doc({'comments': [
            {'auction': {'aucNumId': '123'}, 'rateId': 12345678901234567890, 'content': '口感有嚼劲'},
            {'auction': {'aucNumId': '124'}, 'rateId': 1, 'content': '不属于原商品'}]}), 'taobao', '123')
        self.assertEqual(rows[0]['reviewId'], '12345678901234567890')
        self.assertEqual(len(rows), 1); self.assertEqual(excluded['REVIEW_PRODUCT_ID_MISMATCH'], 1)
    def test_shipping_weight_and_masked_price_not_converted(self):
        rows, price = m.detail(self.doc({'product': {'skuId': '123', 'color': '牛肉面6桶装', 'weight': '1.12'}, 'priceFloor': {'price': '5?'}}), 'jd', '123')
        self.assertEqual(rows[0]['specText'], '牛肉面6桶装')
        self.assertNotIn('netWeightG', rows[0]); self.assertIsNone(price)
    def test_sku_choices_separated(self):
        d = {'item': {'itemId': '123'}, 'skuBase': {'props': [{'pid': '1', 'name': '口味', 'values': [{'vid': 'a', 'name': '牛肉'}, {'vid': 'b', 'name': '鸡汤'}]}], 'skus': [{'skuId': '7', 'propPath': '1:a'}, {'skuId': '8', 'propPath': '1:b'}]}}
        rows, _ = m.detail(self.doc(d), 'taobao', '123')
        self.assertEqual([r['specText'] for r in rows], ['口味:牛肉', '口味:鸡汤'])
    def test_recommendations_and_title_only_rejected(self):
        with self.assertRaisesRegex(ValueError, 'SOCIAL_ID_MISMATCH'): m.social_body(self.doc([{'id': 'other', 'desc': '错配正文'}]), 'xiaohongshu', '123')
        with self.assertRaisesRegex(ValueError, 'BODY_ONLY_REPEATS_TITLE'): m.social_body(self.doc([{'id': '123', 'title': '好吃！', 'desc': '好吃'}]), 'xiaohongshu', '123')
    def test_empty_bili_description_not_title_fallback(self):
        with self.assertRaisesRegex(ValueError, 'UPSTREAM_BODY_EMPTY'): m.social_body(self.doc({'bvid': 'BVx', 'title': '标题', 'desc': ''}), 'bilibili', 'BVx')
    def test_bili_recommendations_never_supply_primary_body(self):
        d = {'bvid': 'BVx', 'title': '标题', 'desc': '', 'related': [{'bvid': 'BVx', 'desc': '这不是本视频的作者简介'}]}
        with self.assertRaisesRegex(ValueError, 'UPSTREAM_BODY_EMPTY'): m.social_body(self.doc(d), 'bilibili', 'BVx')
        with self.assertRaisesRegex(ValueError, 'SOCIAL_ID_MISMATCH'): m.social_body(self.doc({'bvid': 'BVother', 'desc': '他人正文', 'related': [d]}), 'bilibili', 'BVx')
    def test_placeholder_description_is_not_fulltext(self):
        for text in ('-', '——', '...', '暂无简介'):
            with self.subTest(text=text), self.assertRaisesRegex(ValueError, 'UPSTREAM_BODY_PLACEHOLDER'):
                m.social_body(self.doc({'bvid': 'BVx', 'title': '标题', 'desc': text}), 'bilibili', 'BVx')
    def test_primary_web_and_app_shapes(self):
        self.assertEqual(m.social_body(self.doc({'videoData': {'bvid': 'BVx', 'title': '标题', 'desc_v2': [{'raw_text': '真实简介'}]}}), 'bilibili', 'BVx'), '真实简介')
        self.assertEqual(m.social_body(self.doc([{'note_list': [{'id': '123', 'title': '标题', 'desc': '真实笔记正文'}]}]), 'xiaohongshu', '123'), '真实笔记正文')
        self.assertEqual(m.social_body(self.doc({'note_id': '123', 'title': '标题', 'desc': '真实笔记正文'}), 'xiaohongshu', '123'), '真实笔记正文')
    def test_nested_xhs_recommendation_rejected(self):
        with self.assertRaisesRegex(ValueError, 'SOCIAL_ID_MISMATCH'):
            m.social_body(self.doc([{'id': 'other', 'desc': '推荐正文', 'related': [{'id': '123', 'desc': '不能用'}]}]), 'xiaohongshu', '123')
    def test_conflicting_xhs_primary_objects_rejected(self):
        with self.assertRaisesRegex(ValueError, 'SOCIAL_PRIMARY_OBJECT_CONFLICT'):
            m.social_body(self.doc([{'id': '123', 'desc': '原文甲'}, {'id': '123', 'desc': '原文乙'}]), 'xiaohongshu', '123')
    def test_http_success_with_no_video_is_not_a_body(self):
        with self.assertRaisesRegex(ValueError, 'UPSTREAM_DETAIL_UNAVAILABLE'):
            m.social_body(self.doc({'message': 'NO VIDEO'}), 'bilibili', 'BVx')

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