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()