kangshifu-board/server/voc-library/test_board_library.py

import sys
sys.dont_write_bytecode=True
import hashlib
import json
import os
from pathlib import Path
import shutil
import sqlite3
import tempfile
import unittest
import board_library as board
import retrieval
import update_library as updater
import voc_db

ROOT=Path(__file__).resolve().parents[2]
DB=Path(os.environ.get('KSF_TEST_VOC_DB',str(ROOT/'artifacts/康师傅VOC结构化入库-20261105/ksf-voc-library-20261105-r2/library/library.sqlite')))

@unittest.skipUnless(DB.is_file(),'requires private VOC corpus; set KSF_TEST_VOC_DB')
class BoardLibraryTests(unittest.TestCase):
    def setUp(self):self.config={'db':str(DB)}
    def test_real_totals_and_read_only(self):
        before=updater.digest(DB)
        r=board.run(self.config,{'op':'summary'})
        self.assertEqual(r['counts']['products'],4639);self.assertEqual(r['counts']['policy_documents'],25)
        self.assertEqual(r['counts']['policy_clauses'],31);self.assertEqual(r['analysisCounts']['reviews'],771)
        self.assertEqual(updater.digest(DB),before);self.assertEqual(r['databaseWrites'],0)
    def test_exact_scope_facets_and_text(self):
        p={'op':'browse','entity':'products','filters':{'platform':'jd','brand':'ksf','category':'banmian'}}
        r=board.run(self.config,p);self.assertGreater(r['totalMatched'],0)
        self.assertTrue(all(x['platform']=='jd' and x['brand']=='ksf' and x['category']=='banmian' for x in r['records']))
        p['keyword']='不可能命中的独特测试文字81743';self.assertEqual(board.run(self.config,p)['totalMatched'],0)
    def test_cursor_bound_to_query_and_version(self):
        r=board.run(self.config,{'op':'browse','entity':'products'})
        with self.assertRaises(voc_db.LibraryError):board.run(self.config,{'op':'browse','entity':'products','filters':{'brand':'ksf'},'cursor':r['nextCursor']})
        with self.assertRaises(voc_db.LibraryError):board.run(self.config,{'op':'summary','generation':'stale'})
    def test_every_entity_and_source_projection(self):
        for entity in board.LABELS:
            with self.subTest(entity=entity):
                r=board.run(self.config,{'op':'browse','entity':entity})
                self.assertGreater(r['totalMatched'],0);self.assertNotIn('citation',r['records'][0])
                d=board.run(self.config,{'op':'detail','entity':entity,'id':r['records'][0]['record_id']})
                self.assertNotIn('sourcePayload',d);self.assertNotIn('upstreamSources',d['record']['source'])
                self.assertNotIn('/',d['record']['source']['asset_path'])
    def test_reader_product_preview_and_clause_source(self):
        r=board.run(self.config,{'op':'browse','entity':'products'})
        self.assertTrue(all('分类依据' not in x['preview'] and '分类依据' not in (x.get('shop_name') or '') for x in r['records']))
        clause=board.run(self.config,{'op':'browse','entity':'clauses'})['records'][0]
        detail=board.run(self.config,{'op':'detail','entity':'clauses','id':clause['record_id']})['record']
        self.assertEqual(detail['parentDocument']['id'],detail['document_id'])
        self.assertTrue(detail['parentDocument']['title']);self.assertTrue(detail['parentDocument']['source_url'])
    def test_review_quality_and_unsupported_conditions(self):
        a=board.run(self.config,{'op':'browse','entity':'reviews'})
        raw=board.run(self.config,{'op':'browse','entity':'reviews','quality':'audit'})
        self.assertEqual(raw['totalMatched'],2360);self.assertEqual(a['pendingCount'],2360-771)
        with self.assertRaises(voc_db.LibraryError):board.run(self.config,{'op':'browse','entity':'documents','filters':{'brand':'ksf'}})
        with self.assertRaises(voc_db.LibraryError):board.run(self.config,{'op':'source','id':'anything'})
    def test_related_product_counts_are_exact(self):
        r=board.run(self.config,{'op':'browse','entity':'products','filters':{'brand':'ksf'}})['records'][0]
        d=board.run(self.config,{'op':'detail','entity':'products','id':r['record_id']})
        for relation in d['relations']:
            rows=board.run(self.config,{'op':'browse','entity':relation['entity'],'filters':{'productId':r['record_id']},'quality':'audit'})
            self.assertEqual(relation['count'],rows['totalMatched'])
    def test_product_sort_matches_full_database_across_page_boundary(self):
        before=updater.digest(DB)
        with voc_db.readonly(DB) as c:
            counts=dict(c.execute('SELECT product_id,count(*) FROM reviews GROUP BY product_id'))
            prices=dict(c.execute('SELECT product_id,min(amount_minor) FROM prices GROUP BY product_id'))
            products=[(r['record_id'],r['doc_id']) for r in c.execute("SELECT record_id,doc_id FROM retrieval_documents WHERE entity='products'")]
        for sort in ('reviews_asc','reviews_desc','price_asc','price_desc'):
            metric=counts if sort.startswith('reviews_') else prices
            def key(row):
                val=metric.get(row[0],0 if sort.startswith('reviews_') else None)
                return (val is None,(-val if sort.endswith('desc') else val) if val is not None else 0,row[1])
            expected=[x[0] for x in sorted(products,key=key)][:40]
            request={'op':'browse','entity':'products','sort':sort}
            first=board.run(self.config,request)
            second=board.run(self.config,{**request,'cursor':first['nextCursor']})
            self.assertEqual(first['totalMatched'],4639);self.assertEqual(second['totalMatched'],4639)
            self.assertEqual([r['record_id'] for r in first['records']+second['records']],expected)
        scoped=board.run(self.config,{'op':'browse','entity':'products','sort':'reviews_desc','filters':{'platform':'jd','brand':'ksf','category':'banmian'}})
        self.assertTrue(all((r['platform'],r['brand'],r['category'])==('jd','ksf','banmian') for r in scoped['records']))
        self.assertEqual(updater.digest(DB),before)
    def test_product_cards_counts_and_inline_details_keep_exact_product_scope(self):
        before=updater.digest(DB)
        cards=board.run(self.config,{'op':'browse','entity':'products','filters':{'brand':'ksf','category':'banmian'}})
        for card in cards['records'][:3]:
            detail=board.run(self.config,{'op':'detail','entity':'products','id':card['record_id']})
            for relation in detail['relations']:
                self.assertEqual(card['relatedCounts'].get(relation['entity'],0),relation['count'])
        with voc_db.readonly(DB) as c:
            pid=c.execute('SELECT product_id FROM reviews WHERE product_id IS NOT NULL GROUP BY product_id ORDER BY count(*) DESC LIMIT 1').fetchone()[0]
        for entity in ('reviews','specs','prices','rankings'):
            p={'op':'browse','entity':entity,'filters':{'productId':pid},'quality':'audit'}
            result=board.run(self.config,p)
            self.assertTrue(all(row['product_id']==pid for row in result['records']))
            if entity=='reviews':
                self.assertGreater(result['totalMatched'],20)
                row=result['records'][0]
                detail=board.run(self.config,{'op':'detail','entity':entity,'id':row['record_id']})['record']
                self.assertEqual(row['content'],detail['content'])
                self.assertIn('purchased_sku_text',row)
                if result['nextCursor']:
                    second=board.run(self.config,{**p,'cursor':result['nextCursor']})
                    self.assertTrue(all(row['product_id']==pid for row in second['records']))
                    self.assertFalse({r['record_id'] for r in result['records']} & {r['record_id'] for r in second['records']})
                    with self.assertRaises(voc_db.LibraryError):board.run(self.config,{**p,'filters':{'productId':'jd:100000283767'},'cursor':result['nextCursor']})
        self.assertEqual(updater.digest(DB),before)

@unittest.skipUnless(DB.is_file(),'requires private VOC corpus; set KSF_TEST_VOC_DB')
class UpdateTests(unittest.TestCase):
    def setUp(self):self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name);self.registry=self.root/'registry'
    def tearDown(self):self.tmp.cleanup()
    def incoming(self):
        p=self.root/'incoming.sqlite';shutil.copyfile(DB,p)
        c=sqlite3.connect(p);c.row_factory=sqlite3.Row
        c.execute("UPDATE meta SET value='new-generation-test' WHERE key='generation'")
        c.execute("UPDATE meta SET value=? WHERE key='archive_sha256'",('a'*64,))
        c.execute("DELETE FROM batches")
        c.execute('INSERT INTO batches VALUES(?,?,?,?,?)',('a'*64,'新增测试批次.zip','a'*64,'2026-11-06T01:00:00Z','test'))
        product=dict(c.execute('SELECT * FROM products LIMIT 1').fetchone());product.update(id='jd:test-added',item_id='test-added',title='新增拌面测试商品',marketplace='jd')
        c.execute('INSERT INTO products('+','.join(product)+') VALUES('+','.join('?' for _ in product)+')',list(product.values()))
        c.execute("UPDATE reviews SET content='更新后的评价正文' WHERE id=(SELECT id FROM reviews LIMIT 1)")
        price=dict(c.execute('SELECT * FROM prices WHERE source_time_raw IS NOT NULL LIMIT 1').fetchone())
        for pid,observed in [('test-reexport-price',price['source_time_raw']),('test-new-price','2026-11-06T10:00:00+08:00')]:
            row={**price,'id':pid,'source_time_raw':observed}
            if pid=='test-new-price':row.update(source_time_utc='2026-11-06T02:00:00Z',source_calendar_date='2026-11-06')
            c.execute('INSERT INTO prices('+','.join(row)+') VALUES('+','.join('?' for _ in row)+')',list(row.values()))
        for table in ('retrieval_fts','retrieval_documents'):c.execute('DROP TABLE '+table)
        retrieval.create_index(c);c.commit();c.close();return p
    def test_dry_run_persistence_merge_provenance_and_rollback(self):
        updater.publish(DB,self.registry)
        self.assertFalse(self.registry.exists())
        first=updater.publish(DB,self.registry,True,'首批');before=(self.registry/first['file']).read_bytes()
        repeat=updater.publish(DB,self.registry,True);self.assertEqual(repeat['operation'],'already-imported')
        incoming=self.incoming();second=updater.publish(incoming,self.registry,True,'新增一批')
        self.assertNotEqual(first['generation'],second['generation']);self.assertEqual((self.registry/first['file']).read_bytes(),before)
        cfg={'registry':str(self.registry)};s=board.run(cfg,{'op':'summary'})
        self.assertEqual(voc_db.registry_database(self.registry),self.registry/second['file'])
        self.assertEqual(s['counts']['products'],4640);self.assertEqual(s['counts']['reviews'],2360);self.assertEqual(len(s['updates']),2)
        self.assertEqual(s['counts']['prices'],4071)
        self.assertEqual(second['delta']['added']['products'],1);self.assertEqual(second['delta']['changed']['reviews'],1)
        with voc_db.readonly(self.registry/second['file']) as c:
            self.assertGreater(c.execute('SELECT count(*) FROM record_versions').fetchone()[0],0)
            origin=voc_db.citation(c,c.execute("SELECT source_record_id FROM products WHERE id='jd:test-added'").fetchone()[0])
            self.assertTrue(origin['archiveSha256s'])
        updater.publish(None,self.registry,True,rollback=first['generation'])
        self.assertEqual(board.run(cfg,{'op':'summary'})['counts']['products'],4639)
    def test_invalid_batch_keeps_current(self):
        first=updater.publish(DB,self.registry,True)
        pointer=(self.registry/'current.json').read_bytes()
        bad=self.root/'bad.sqlite';bad.write_text('bad',encoding='utf-8')
        with self.assertRaises(sqlite3.DatabaseError):updater.publish(bad,self.registry,True)
        self.assertEqual((self.registry/'current.json').read_bytes(),pointer)
        self.assertFalse((self.registry/'.update.lock').exists())
        self.assertEqual(board.run({'registry':str(self.registry)},{'op':'version'})['generation'],first['generation'])

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