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

import contextlib
import json
from pathlib import Path
import shutil
import sqlite3
import tempfile
import unittest
import test_voc_db as fixture
import retrieval as r


class RetrievalTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        fixture.VocLibraryTests.setUpClass()
        cls.db=fixture.VocLibraryTests.db

    @classmethod
    def tearDownClass(cls):fixture.VocLibraryTests.tearDownClass()

    def search(self,**request):
        with fixture.v.readonly(self.db) as c:return r.search(c,request,True)

    def test_no_compulsory_three_dimensions(self):
        result=self.search(entities=['products'],filters={'brand':'康师傅'})
        self.assertTrue(any(x['category'] is None for x in result['records']))
        self.assertTrue(all(x['brand']=='ksf' for x in result['records']))
        self.assertNotIn('platform',result['effectiveQuery']['filters'])

    def test_category_only_can_find_unknown_brand(self):
        result=self.search(entities=['products'],filters={'category':'拌面'})
        self.assertIn('jd:unknown',{x['record_id'] for x in result['records']})
        self.assertTrue(all(x['category']=='banmian' for x in result['records']))

    def test_product_id_is_sufficient_and_cross_platform_safe(self):
        result=self.search(entities=['reviews'],filters={'productId':'jd:same'})
        self.assertEqual({x['product_id'] for x in result['records']},{'jd:same'})
        self.assertEqual(result['totalMatched'],2)  # includes unclear SKU absent a category constraint
        self.assertEqual(self.search(entities=['reviews'],filters={'productId':'taobao:same'})['totalMatched'],0)

    def test_actual_constraints_are_preserved(self):
        result=self.search(entities=['products'],filters={'platform':'京东','brand':'白象','category':'拌面'})
        self.assertEqual([x['record_id'] for x in result['records']],['jd:other-brand'])

    def test_explicit_scope_and_provenance_across_product_relations(self):
        for entity in ('products','reviews','prices','specs'):
            with self.subTest(entity=entity):
                result=self.search(entities=[entity],filters={'platform':'京东','brand':'康师傅','category':'拌面'},limit=100)
                self.assertGreater(result['totalMatched'],0)
                for row in result['records']:
                    self.assertEqual((row['platform'],row['brand'],row['category']),('jd','ksf','banmian'))
                    self.assertEqual(row['product_id'],'jd:same')
                    self.assertTrue(all(row['citation'][key] for key in ('id','asset_path','line_number','payload_sha256','assetSha256','archiveSha256')))
                    with fixture.v.readonly(self.db) as c:
                        source=fixture.v.get_source(c,row['source_record_id'])
                    self.assertEqual(source['citation']['id'],row['source_record_id'])
                if entity=='reviews':
                    self.assertTrue(all(row['category_state']=='consistent' and row['identity_state']=='matched' and not row['is_default'] for row in result['records']))

    def test_review_sort_keeps_all_products_and_paginates_equal_counts(self):
        for sort in ('reviews_desc','reviews_asc'):
            query={'entities':['products'],'sort':sort,'limit':2}
            rows=[];cursor=None
            while True:
                page=self.search(**query,**({'cursor':cursor} if cursor else {}))
                self.assertEqual(page['totalMatched'],9);rows+=page['records'];cursor=page['nextCursor']
                if not cursor:break
            self.assertEqual(len(rows),9);self.assertEqual(len({x['record_id'] for x in rows}),9)
            self.assertEqual(rows[0 if sort=='reviews_desc' else -1]['record_id'],'jd:same')
        first=self.search(entities=['products'],sort='reviews_desc',limit=2)
        with self.assertRaises(ValueError):self.search(entities=['products'],sort='reviews_asc',cursor=first['nextCursor'])
        with self.assertRaises(ValueError):self.search(entities=['products'],cursor=first['nextCursor'])
        with self.assertRaises(ValueError):self.search(entities=['reviews'],sort='reviews_desc')
        with self.assertRaises(ValueError):self.search(entities=['products'],sort='price; DROP TABLE products')

    def test_price_sort_uses_lowest_observation_and_keeps_unknown_prices_last(self):
        with tempfile.TemporaryDirectory() as tmp:
            db=Path(tmp)/'prices.sqlite';shutil.copyfile(self.db,db)
            with contextlib.closing(sqlite3.connect(db)) as c:
                c.row_factory=sqlite3.Row
                price=dict(c.execute("SELECT * FROM prices WHERE product_id='jd:same' LIMIT 1").fetchone())
                price.update(id='second-observation',amount_minor=9990,raw_amount='99.90')
                c.execute('INSERT INTO prices('+','.join(price)+') VALUES('+','.join('?' for _ in price)+')',list(price.values()))
                c.execute("UPDATE prices SET amount_minor=NULL WHERE product_id IN ('jd:other-brand','jd:sauce')")
                c.execute("UPDATE prices SET amount_minor=4900 WHERE product_id='jd:soup'")
                c.commit()
            with fixture.v.readonly(db) as c:
                for sort in ('price_asc','price_desc'):
                    ids=[];cursor=None
                    while True:
                        page=r.search(c,{'entities':['products'],'sort':sort,'limit':1,**({'cursor':cursor} if cursor else {})})
                        self.assertEqual(page['totalMatched'],9)
                        ids += [x['record_id'] for x in page['records']];cursor=page['nextCursor']
                        if not cursor:break
                    self.assertEqual(len(ids),9);self.assertEqual(len(set(ids)),9)
                    self.assertEqual(set(ids[-2:]),{'jd:other-brand','jd:sauce'})
                    self.assertEqual(ids[6 if sort=='price_asc' else 0],'jd:soup')
                    self.assertEqual(ids.index('jd:same')<ids.index('jd:soup'),sort=='price_asc')

    def test_literal_review_body_not_product_title(self):
        result=self.search(entities=['reviews'],text={'all':['面量']})
        self.assertGreater(result['totalMatched'],0)
        self.assertTrue(all('面量' in x['match']['text'] for x in result['records']))
        self.assertEqual(self.search(entities=['reviews'],text={'all':['110g']})['totalMatched'],0)

    def test_cjk_two_character_fulltext_and_fts_plan(self):
        result=self.search(entities=['reviews'],text={'all':['面量']})
        self.assertTrue(any('VIRTUAL TABLE INDEX' in step for step in result['queryPlan']))
        self.assertTrue(all('面量' in x['match']['matchedTerms'] for x in result['records']))

    def test_synonyms_visible_and_not_a_new_fact(self):
        result=self.search(entities=['reviews'],text={'concepts':['portion_size']})
        self.assertIn('面量',result['effectiveQuery']['text']['conceptExpansion']['portion_size'])
        self.assertTrue(result['records'][0]['match']['literalMatchOnly'])

    def test_or_within_group_and_between_groups(self):
        result=self.search(entities=['products'],text={'all':[{'any':['康师傅','白象']},'干拌面']})
        self.assertTrue(all('干拌面' in x['match']['text'] and ('康师傅' in x['match']['text'] or '白象' in x['match']['text']) for x in result['records']))
        result=self.search(entities=['products'],text={'all':['干拌面'],'exclude':['白象']})
        self.assertTrue(all('白象' not in x['match']['text'] for x in result['records']))

    def test_multientity_counts_and_social_mention_relationship(self):
        result=self.search(entities=['products','social'],filters={'brand':'康师傅'})
        self.assertIn('social',result['countsByEntity'])
        social=next(x for x in result['records'] if x['entity']=='social')
        self.assertEqual(social['brandRelationship'],'text_mention_only_not_product_ownership')

    def test_nonapplicable_filter_never_ignored(self):
        result=self.search(entities=['reviews','social'],filters={'productId':'jd:same'})
        self.assertNotIn('social',result['countsByEntity'])
        with self.assertRaises(ValueError):self.search(entities=['social'],filters={'productId':'jd:same'})

    def test_unknown_dimension_audit_and_not_forced_drop(self):
        result=self.search(entities=['products'],filters={'brand':{'exists':False}})
        self.assertGreater(result['totalMatched'],0)
        self.assertTrue(all(x['brand'] is None for x in result['records']))
        with self.assertRaises(ValueError):self.search(entities=['reviews'],filters={'isDefault':True})
        audit=self.search(entities=['reviews'],filters={'isDefault':True},quality='audit')
        self.assertEqual(audit['totalMatched'],1)

    def test_price_and_spec_range_units(self):
        result=self.search(entities=['prices'],filters={'priceMinor':{'min':1900,'max':2000},'productId':'jd:same'})
        self.assertEqual(result['totalMatched'],1)
        specs=self.search(entities=['specs'],filters={'netWeightG':{'min':100,'max':120},'specVerification':'title_candidate'})
        self.assertGreater(specs['totalMatched'],0)
        self.assertTrue(all(x['spec_verification']=='title_candidate' for x in specs['records']))

    def test_dates_filters_are_real_fields_not_import_time(self):
        result=self.search(entities=['reviews'],filters={'reviewDate':{'min':'2026-11-04','max':'2026-11-04'}})
        self.assertEqual(result['totalMatched'],2)
        with self.assertRaises(ValueError):self.search(entities=['reviews'],filters={'reviewDate':{'min':'2026-02-31'}})

    def test_aggregate_distinguishes_records_products(self):
        with fixture.v.readonly(self.db) as c:
            result=r.aggregate(c,dict(entities=['reviews'],groupBy=['platform']))
            row=result['groups']['platform']['rows'][0]
            self.assertEqual(row['recordCount'],2)
            self.assertEqual(row['linkedProductCount'],1)
            self.assertEqual(row['distinctTextCount'],1)

    def test_facets_in_scope(self):
        with fixture.v.readonly(self.db) as c:
            result=r.aggregate(c,dict(entities=['products'],filters={'platform':'jd'},groupBy=['brand']),True)
            self.assertTrue(all(x['recordCount']>0 for x in result['groups']['brand']['rows']))

    def test_pagination_scope_generation(self):
        first=self.search(entities=['products'],limit=2)
        second=self.search(entities=['products'],limit=2,cursor=first['nextCursor'])
        self.assertFalse({x['record_id'] for x in first['records']}&{x['record_id'] for x in second['records']})
        with self.assertRaises(ValueError):self.search(entities=['products'],filters={'brand':'ksf'},cursor=first['nextCursor'])

    def test_no_sql_or_fts_query_language_execution(self):
        self.assertEqual(self.search(entities=['reviews'],text={'all':['面量 OR 1=1']})['totalMatched'],0)
        self.assertEqual(self.search(entities=['products'],filters={'itemId':"same' OR 1=1 --"})['totalMatched'],0)
        with self.assertRaises(ValueError):self.search(entities=['products'],filters={'customSql':'1=1'})

    def test_catalog_and_detail_path_for_new_entities(self):
        with fixture.v.readonly(self.db) as c:
            catalog=r.catalog(c)
            self.assertIn('specs',catalog['entities']);self.assertIn('rankings',catalog['entities'])
            result=r.search(c,dict(entities=['specs'],filters={'productId':'jd:same'}))
            row=fixture.v.get_record(c,'specs',result['records'][0]['record_id'])
            self.assertEqual(row['record']['verification'],'title_candidate')

    def test_rating_requires_platform_and_original_scale(self):
        with self.assertRaises(ValueError):self.search(entities=['reviews'],filters={'ratingValue':{'max':2}})
        result=self.search(entities=['reviews'],filters={'ratingValue':{'min':5},'ratingScale':'jd原始星级','platform':'jd'})
        self.assertEqual(result['totalMatched'],2)

    def test_policy_analyst_note_is_labelled(self):
        result=self.search(entities=['clauses'],text={'all':['应用建议']})
        self.assertEqual(result['totalMatched'],1)
        match=result['records'][0]['match']
        self.assertEqual(match['matchedFields'],['application_note'])
        self.assertFalse(match['applicationNoteIsRegulation'])

    def test_social_brand_existence_is_mentions_not_null_projection(self):
        self.assertEqual(self.search(entities=['social'],filters={'brand':{'exists':True}})['totalMatched'],1)
        self.assertEqual(self.search(entities=['social'],filters={'brand':{'exists':False}})['totalMatched'],0)
        with fixture.v.readonly(self.db) as c:
            with self.assertRaises(ValueError):r.aggregate(c,dict(entities=['social'],groupBy=['brand']))


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