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

import hashlib
import json
from pathlib import Path
import sqlite3
import tempfile
import unittest
import agent_batch
import board_library
import update_library
import voc_db

class AgentBatchTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name); (self.root / 'raw').mkdir()
        self.raw = self.root / 'raw/response.json'
        self.raw.write_text(json.dumps({'product': {'id': '123', 'name': '康师傅香辣拌面110g*5袋', 'price': '19.9'},
            'comments': [{'item': '123', 'id': 'r1', 'text': '一包根本吃不饱', 'default': False, 'sku': '香辣拌面'}]}, ensure_ascii=False), encoding='utf-8')
        self.batch = {'schema': agent_batch.SCHEMA, 'batchId': 'batch-test', 'producer': 'test-agent/v1',
            'sources': [{'id': 'run1', 'path': 'raw/response.json', 'sha256': agent_batch.digest(self.raw),
                         'platform': 'jd', 'marketplace': 'jd', 'requestItemId': '123', 'observedAt': '2026-11-07T10:00:00+08:00'}],
            'products': [{'sourceId': 'run1', 'pointer': '/product', 'fields': {'itemId': '/id', 'title': '/name', 'price': '/price'}}],
            'reviews': [{'sourceId': 'run1', 'pointer': '/comments/0', 'fields': {'itemId': '/item', 'reviewId': '/id',
                        'content': '/text', 'isDefault': '/default', 'purchasedSku': '/sku'}}],
            'themes': [{'reviewId': 'jd:123:r1', 'theme': 'portion', 'quote': '一包根本吃不饱', 'quoteField': 'content'}]}
        self.path = self.root / 'batch.json'; self.output = self.root / 'new.sqlite'; self.save()
    def save(self): self.path.write_text(json.dumps(self.batch, ensure_ascii=False), encoding='utf-8')
    def test_mapping_validation_is_readonly_then_build_publish_and_query(self):
        before = {str(p): (agent_batch.digest(p), p.stat().st_mtime_ns) for p in self.root.rglob('*') if p.is_file()}
        dry = agent_batch.build_batch(self.path, self.output)
        self.assertTrue(dry['readOnly']); self.assertFalse(self.output.exists())
        self.assertEqual(before, {str(p): (agent_batch.digest(p), p.stat().st_mtime_ns) for p in self.root.rglob('*') if p.is_file()})
        built = agent_batch.build_batch(self.path, self.output, True)
        self.assertTrue(built['applied']); self.assertFalse(built['activeLibraryModified']); self.assertTrue(self.output.with_suffix('.source.zip').is_file())
        registry = self.root / 'registry'; update_library.publish(self.output, registry, True)
        d = board_library.run({'registry': str(registry)}, {'op': 'detail', 'entity': 'products', 'id': 'jd:123'})
        self.assertEqual(d['record']['overview']['reviews']['eligible'], 1)
        theme = next(t for t in d['record']['overview']['themes'] if t['id'] == 'portion')
        self.assertIn('agent', theme['methods']); self.assertEqual(theme['examples'][0]['quote'], '一包根本吃不饱')
        with voc_db.readonly(self.output) as c:
            r = c.execute('SELECT source_record_id FROM reviews').fetchone()
            citation = voc_db.citation(c, r[0]); self.assertTrue(citation['archiveSha256'])
        with self.assertRaises(voc_db.LibraryError): agent_batch.build_batch(self.path, self.output, True)
    def test_wrong_product_default_type_fabricated_quote_and_missing_map_rejected(self):
        cases = [('source-mismatch', lambda: self.batch['sources'][0].update(requestItemId='999')),
                 ('fabricated-quote', lambda: self.batch['themes'][0].update(quote='味道鲜美')),
                 ('missing-content', lambda: self.batch['reviews'][0]['fields'].pop('content')),
                 ('duplicate', lambda: self.batch['reviews'].append(self.batch['reviews'][0]))]
        original = json.loads(json.dumps(self.batch))
        for name, mutate in cases:
            with self.subTest(name=name):
                self.batch = json.loads(json.dumps(original)); mutate(); self.save()
                with self.assertRaises(voc_db.LibraryError): agent_batch.build_batch(self.path, self.output)
                self.assertFalse(self.output.exists())
    def test_wrong_hash_and_source_path_escape_rejected(self):
        self.batch['sources'][0]['sha256'] = '0' * 64; self.save()
        with self.assertRaisesRegex(voc_db.LibraryError, 'SOURCE_HASH_MISMATCH'): agent_batch.build_batch(self.path, self.output)
        self.batch['sources'][0]['path'] = '../outside.json'; self.save()
        with self.assertRaisesRegex(voc_db.LibraryError, 'INVALID_SOURCE_PATH'): agent_batch.build_batch(self.path, self.output)
    def test_null_title_and_non_boolean_default_rejected(self):
        raw = json.loads(self.raw.read_text(encoding='utf-8'))
        raw['product']['name'] = None
        self.raw.write_text(json.dumps(raw), encoding='utf-8')
        self.batch['sources'][0]['sha256'] = agent_batch.digest(self.raw); self.save()
        with self.assertRaisesRegex(voc_db.LibraryError, 'PRODUCT_TITLE_REQUIRED'): agent_batch.build_batch(self.path, self.output)
        raw['product']['name'] = '康师傅拌面'
        raw['comments'][0]['default'] = 'false'
        self.raw.write_text(json.dumps(raw), encoding='utf-8')
        self.batch['sources'][0]['sha256'] = agent_batch.digest(self.raw); self.save()
        with self.assertRaisesRegex(voc_db.LibraryError, 'DEFAULT_STATE_MUST_BE_BOOLEAN_OR_NULL'): agent_batch.build_batch(self.path, self.output)
    def test_new_batch_merged_annotations_survive_old_snapshot_and_updates(self):
        agent_batch.build_batch(self.path, self.output, True)
        registry = self.root / 'registry'; update_library.publish(self.output, registry, True)
        self.batch['batchId'] = 'batch-two'; self.batch['producer'] = 'test-agent/v2'; self.save()
        second = self.root / 'second.sqlite'; agent_batch.build_batch(self.path, second, True)
        change = update_library.publish(second, registry, True)
        self.assertTrue(change['applied'])
        d = board_library.run({'registry': str(registry)}, {'op': 'detail', 'entity': 'products', 'id': 'jd:123'})
        self.assertEqual(d['record']['overview']['reviews']['collected'], 1)
        self.assertIn('agent', next(t for t in d['record']['overview']['themes'] if t['id'] == 'portion')['methods'])

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