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

import base64
import contextlib
import hashlib
import io
import json
import os
from pathlib import Path
import sqlite3
import subprocess
import tempfile
import unittest
from unittest.mock import patch
import agent_batch
import library_channel as client
import library_receiver as receiver
import update_library
import voc_db as v

class LibraryChannelTests(unittest.TestCase):
    def setUp(self):
        self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
        self.root=Path(self.temp.name);self.registry=self.root/'registry';self.intake=self.root/'intake';self.intake.mkdir(mode=0o700)
        self.rawroot=self.root/'input';(self.rawroot/'raw').mkdir(parents=True)
        self.raw=self.rawroot/'raw/response.json'
        self.raw.write_text(json.dumps({'product':{'id':'123','title':'康师傅香辣拌面'},'review':{'item':'123','id':'r1','text':'味道很好，一包吃不饱','default':False,'sku':'香辣拌面'}}),encoding='utf-8')
        self.batch={'schema':'ksf.voc-agent-batch.v2','batchId':'first','producer':'test-agent',
          'sources':[{'id':'s','path':'raw/response.json','sha256':agent_batch.digest(self.raw),'platform':'jd','marketplace':'jd','requestItemId':'123'}],
          'products':[{'sourceId':'s','pointer':'/product','fields':{'itemId':'/id','title':'/title'}}],
          'reviews':[{'sourceId':'s','pointer':'/review','fields':{'itemId':'/item','reviewId':'/id','content':'/text','isDefault':'/default','purchasedSku':'/sku'}}]}
        self.path=self.rawroot/'batch.json';self.save()
        stage=self.root/'initial.sqlite';agent_batch.build_batch(self.path,stage,True)
        self.initial=update_library.publish(stage,self.registry,True)['generation']
        self.config={'schema':client.VERSION,'registry':str(self.registry),'intake':str(self.intake),'canPublish':True}
    def save(self):self.path.write_text(json.dumps(self.batch),encoding='utf-8')
    def request(self,apply=False):
        raw,h=client.pack(self.path)
        return {'schema':client.VERSION,'action':'intake','archiveBase64':base64.b64encode(raw).decode(),
          'params':{'archiveSha256':h,'expectedGeneration':self.initial,'label':'真实批次测试','apply':apply}}
    def test_remote_reads_build_publish_and_same_generation_projection(self):
        self.batch['batchId']='second';self.save()
        request=self.request();before=(self.registry/'current.json').read_bytes()
        plan=receiver.dispatch(self.config,request)
        self.assertTrue(plan['readOnly']);self.assertFalse(plan['persistentFilesWritten']);self.assertEqual(list(self.intake.iterdir()),[])
        self.assertEqual(before,(self.registry/'current.json').read_bytes())
        request['params']['apply']=True;result=receiver.dispatch(self.config,request)
        self.assertTrue(result['remotePublished']);self.assertTrue(result['sourceArchivePreserved']);self.assertNotEqual(result['generation'],self.initial)
        contract=receiver.dispatch(self.config,{'schema':client.VERSION,'action':'read','command':'contract','params':{}})
        self.assertEqual(contract['generation'],result['generation']);self.assertEqual(contract['publicationBinding']['mode'],'ssh-channel')
        self.assertTrue(contract['publicationBinding']['remotePublisherEnabled'])
        schema=receiver.dispatch(self.config,{'schema':client.VERSION,'action':'read','command':'schema','params':{'table':'reviews','generation':result['generation']}})
        self.assertEqual(schema['generation'],result['generation']);self.assertTrue(schema['complete'])
        review=receiver.dispatch(self.config,{'schema':client.VERSION,'action':'read','command':'get','params':{'entity':'reviews','id':'jd:123:r1','generation':result['generation']}})
        self.assertEqual(review['record']['content'],'味道很好，一包吃不饱')
        source=receiver.dispatch(self.config,{'schema':client.VERSION,'action':'read','command':'source','params':{'id':review['record']['source_record_id'],'generation':result['generation']}})
        self.assertTrue(source['ok']);self.assertEqual(source['sourcePayload']['source_locator'],'/review')
        status=receiver.dispatch(self.config,{'schema':client.VERSION,'action':'status','params':{'archiveSha256':request['params']['archiveSha256']}})
        self.assertTrue(status['publicationFound']);self.assertEqual(status['publishedGeneration'],result['generation'])
        with self.assertRaisesRegex(v.LibraryError,'LIBRARY_GENERATION_CHANGED'):receiver.dispatch(self.config,request)
        request['params']['expectedGeneration']=result['generation']
        self.assertFalse(receiver.dispatch(self.config,request)['applied'])
    def test_receiver_rejects_arbitrary_paths_commands_permissions_and_changed_bytes(self):
        invalid=[{'schema':client.VERSION,'action':'read','command':'build','params':{}},
          {'schema':client.VERSION,'action':'read','command':'get','params':{'db':'/etc/shadow'}},
          {'schema':client.VERSION,'action':'shell','params':{}},
          {'schema':client.VERSION,'action':'read','command':'schema','params':{'table':'--db'}}]
        for request in invalid:
            with self.assertRaises(v.LibraryError):receiver.dispatch(self.config,request)
        request=self.request();request['params']['archiveSha256']='0'*64
        with self.assertRaisesRegex(v.LibraryError,'ARCHIVE_HASH_MISMATCH'):receiver.dispatch(self.config,request)
        with self.assertRaisesRegex(v.LibraryError,'PUBLICATION_DISABLED'):receiver.dispatch({**self.config,'canPublish':False},self.request())
        self.assertEqual(list(self.intake.iterdir()),[])
    def test_archive_paths_symlinks_extra_members_and_tampering_rejected(self):
        import zipfile
        for name in ('../escape','/absolute','raw/../../escape','raw/unreferenced.txt'):
            output=io.BytesIO()
            with zipfile.ZipFile(output,'w') as z:
                z.writestr('batch.json',self.path.read_bytes());z.writestr('raw/response.json',self.raw.read_bytes());z.writestr(name,'bad')
            with tempfile.TemporaryDirectory(dir=self.root) as td:
                with self.assertRaises(v.LibraryError):receiver.unpack(output.getvalue(),Path(td))
        self.assertFalse((self.root/'escape').exists())
    def test_transport_is_fixed_command_strict_host_check_and_never_retries_uncertain_publish(self):
        key=self.root/'key';hosts=self.root/'known_hosts';key.write_text('test-only-key');hosts.write_text('test-only-host')
        os.chmod(key,0o600);os.chmod(hosts,0o600)
        conf=self.root/'channel.json';conf.write_text(v.canonical({'schema':client.VERSION,'host':'127.0.0.1','port':22,'user':'publisher','identityFile':str(key),'knownHostsFile':str(hosts)}));os.chmod(conf,0o600)
        calls=[]
        def run(command,**kwargs):
            calls.append(command);self.assertEqual(command[-1],'ksf-library-v1');self.assertIn('StrictHostKeyChecking=yes',command)
            self.assertNotIn('shell',kwargs);return subprocess.CompletedProcess(command,0,b'{"ok":true}',b'')
        self.assertTrue(client.send(conf,{'action':'read','command':'contract','params':{}},run)['ok'])
        def timeout(command,**kwargs):calls.append(command);raise subprocess.TimeoutExpired(command,300)
        with self.assertRaisesRegex(v.LibraryError,'CHANNEL_RESULT_UNCERTAIN'):client.send(conf,{'action':'intake','params':{'apply':True}},timeout)
        self.assertEqual(len(calls),2)
    def test_hashing_and_publish_do_not_depend_on_file_digest(self):
        with patch.object(hashlib,'file_digest',None,create=True):
            self.assertEqual(agent_batch.digest(self.raw),hashlib.sha256(self.raw.read_bytes()).hexdigest())
            self.assertEqual(update_library.digest(self.raw),agent_batch.digest(self.raw))
            self.batch['batchId']='compat';self.save()
            new=self.root/'compat.sqlite';agent_batch.build_batch(self.path,new,True)
            self.assertTrue(update_library.publish(new,self.registry,True,expected_generation=self.initial)['applied'])

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