91 lines
12 KiB
JavaScript
91 lines
12 KiB
JavaScript
/**
|
||
* Foundry 模型流 framing、完整校验及 transport/loader 协调测试。流可以先交付只读 delta，完整文档必须等终态记录与正常 EOF。
|
||
* 快照/序号/覆盖/身份清单绑定所有后续记录，错误尾部保留可读前缀但不能打开为权威基线。
|
||
*/
|
||
import {test} from 'node:test';
|
||
import assert from 'node:assert/strict';
|
||
import {createRegistry,builtinComponents} from '../../src/lib/canvas-core/index.ts';
|
||
import {decodeModelStream,modelStreamType} from '../../src/lib/canvas-service/model-stream.ts';
|
||
import {createFoundryTransport} from '../../src/lib/canvas-service/foundry.ts';
|
||
import {createFoundryModelLoader} from '../../src/lib/canvas-service/model-loader.ts';
|
||
import {createScopedLoader} from '../../src/lib/canvas-runtime/loader.ts';
|
||
// 固定 UUID 作用域，满足 transport 与模型解析的身份约束。
|
||
const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000002',account='10000000-0000-4000-8000-000000000003';
|
||
// 内置定义和独立信号工厂，避免取消跨用例复用。
|
||
const registry=createRegistry(builtinComponents()),signal=()=>new AbortController().signal;
|
||
// 含中文及 emoji 的合法节点样本，验证逐字节传输时 UTF-8 边界处理。
|
||
const node={id:'a',incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',data:{title:'中文',text:'🙂'},geometry:{x:0,y:0,width:100,height:80,rotation:0}};
|
||
// 每次新建 header/elements/complete 记录，快照 token 相同、序号连续且声明一个元素；便于逐字段篡改。
|
||
const records=()=>[{streamVersion:1,snapshotToken:'snapshot-a',sequence:0,kind:'header',documentFormatVersion:1,projectId:root,canvases:[{id:root,parent:null}],versions:{[root]:3},tombstones:{old:2},removedCanvasIds:[],elementCount:1},{streamVersion:1,snapshotToken:'snapshot-a',sequence:1,kind:'elements',elements:[structuredClone(node)]},{streamVersion:1,snapshotToken:'snapshot-a',sequence:2,kind:'complete',elementCount:1}];
|
||
// 编码为带末尾换行的 UTF-8 NDJSON；byteLength 断言以真实传输字节为准。
|
||
const encode=rows=>new TextEncoder().encode(rows.map(row=>JSON.stringify(row)).join('\n')+'\n');
|
||
// 单批字节异步源，不模拟网络切片，逐字节场景使用另一个 split generator。
|
||
async function* source(rows){yield encode(rows);}
|
||
// 排空异步迭代器到数组，让尾部或 EOF 校验错误也传播到测试断言。
|
||
const collect=async iterable=>{const result=[];for await(const item of iterable)result.push(item);return result;};
|
||
// canonical 完整项目查询作用域，交给通用 scoped loader 检查权限和查询绑定。
|
||
const scope={accountId:account,workspaceId:workspace,projectId:null,canvasId:root,authEpoch:1,navigationGeneration:1,viewKey:'canonical-project',format:'canonical-model-v1',queryFingerprint:'full-project',source:'foundry'};
|
||
|
||
// 逐字节传入 Unicode/换行，依次输出接收头、元素 delta、最终 ready；正文/墓碑及总 byteLength 完整，记录冻结。
|
||
test('stream decoder frames split UTF8/newlines, emits deltas and validates one final complete model',async()=>{
|
||
const bytes=encode(records());/* 每批只发一个字节，主动拆开多字节 UTF-8 与记录边界。 */ async function* split(){for(const byte of bytes)yield Uint8Array.of(byte);}
|
||
// collect 会读到生成器 EOF：看到 complete 记录还不够，尾部 framing 校验也必须成功。
|
||
const chunks=await collect(decodeModelStream(split(),root,registry,signal()));
|
||
assert.deepEqual(chunks.map(chunk=>[chunk.sequence,chunk.complete,chunk.value.status]),[[0,false,'receiving'],[1,false,'receiving'],[2,true,'ready']]);
|
||
assert.equal(chunks[1].value.elements[0].data.text,'🙂');assert.equal(chunks[2].value.document.elements.get('a').data.title,'中文');assert.equal(chunks[2].value.document.tombstones.get('old'),2);assert.equal(chunks.reduce((sum,chunk)=>sum+chunk.byteLength,0),bytes.length);assert.ok(Object.isFrozen(chunks[1].value.elements[0]));
|
||
});
|
||
// 快照切换、序号缺口、计数不符、重复 ID、错误 owner/几何/版本/拓扑、缺少结束及缺末换行都不能完成，超大行按流预算拒绝。
|
||
test('snapshot switches, sequence gaps, duplicate IDs, wrong counts, late records and invalid topology cannot finish',async()=>{
|
||
const cases=[rows=>rows[1].snapshotToken='other',rows=>rows[1].sequence=2,rows=>rows[0].elementCount=2,rows=>rows[2].elementCount=2,rows=>rows.push({...rows[2],sequence:3}),rows=>rows[1].elements.push(rows[1].elements[0]),rows=>rows[1].elements[0].ownerCanvasId=workspace,rows=>rows[1].elements[0].geometry.width=-1,rows=>rows[0].versions={},rows=>rows[0].canvases[0].parent={canvasId:root,entryElementId:'a'}];
|
||
for(const mutate of cases){const rows=records();mutate(rows);await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal())));}
|
||
for(const rows of [[],records().slice(0,1),records().slice(0,2)])await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal())));
|
||
/* 删除文件末换行，验证终态记录也不能绕过 framing 完整性。 */ async function* truncated(){yield encode(records()).slice(0,-1);}await assert.rejects(()=>collect(decodeModelStream(truncated(),root,registry,signal())),{code:'incomplete_model_stream'});
|
||
/* 单行超过 8 MiB，没有换行，必须在无限积累前触发预算。 */ async function* huge(){yield new Uint8Array(8*1024*1024+1).fill(65);}await assert.rejects(()=>collect(decodeModelStream(huge(),root,registry,signal())),{code:'model_stream_budget'});
|
||
});
|
||
// transport 以 Accept 协商流并交出 ReadableStream；尾部 gate 前 loader 已看到首个元素 delta，放行后才 complete。
|
||
test('Foundry transport negotiates streaming without buffering; coordinator retains first delta before gated tail',async()=>{
|
||
let release,opened;const gate=new Promise(resolve=>release=resolve),started=new Promise(resolve=>opened=resolve);let requested;
|
||
const rows=records(),stream=new ReadableStream({async start(controller){controller.enqueue(encode(rows.slice(0,2)));opened();await gate;controller.enqueue(encode(rows.slice(2)));controller.close();}});
|
||
const transport=createFoundryTransport({accountId:account,isCurrentIdentity:()=>true,fetch:async(_url,input)=>{requested=input;return new Response(stream,{headers:{'Content-Type':modelStreamType}});}});
|
||
const seen=[],loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport}),current:()=>true});
|
||
const handle=loader.acquire(scope,{onUpdate:snapshot=>seen.push(snapshot)});try{
|
||
await started;while(!seen.some(snapshot=>snapshot.chunks.length===2))await new Promise(resolve=>setImmediate(resolve));
|
||
assert.equal(requested.headers.Accept,modelStreamType);assert.equal(handle.getSnapshot().status,'partial');assert.equal(handle.getSnapshot().chunks.at(-1).value.status,'receiving');
|
||
release();const done=await handle.ready;assert.equal(done.status,'complete');assert.equal(done.chunks.at(-1).value.document.elements.size,1);
|
||
}finally{release();handle.release();loader.close();}
|
||
});
|
||
// 坏 JSON 尾部使 loader failed-with-partial，保留两个 receiving 块，不能输出权威模型。
|
||
test('malformed tails retain read deltas without yielding an authoritative model',async()=>{
|
||
/* 合法头/元素后追加坏 JSON，保留可读前缀并拒绝最终模型。 */ async function* broken(){yield encode(records().slice(0,2));yield new TextEncoder().encode('{broken}\n');}
|
||
const states=[],loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:broken()})}),current:()=>true});const handle=loader.acquire(scope,{onUpdate:value=>states.push(value)});
|
||
try{await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');assert.equal(handle.getSnapshot().chunks.length,2);assert.equal(handle.getSnapshot().chunks.at(-1).value.status,'receiving');}finally{handle.release();loader.close();}
|
||
});
|
||
// complete 记录已到但流未结束时 ready 仍未解决；随后垃圾尾行使读取失败，不能先打开基线再反悔。
|
||
test('terminal record still requires EOF; trailing garbage rejects rather than opening a baseline',async()=>{
|
||
const gate={},rows=records();let release;gate.promise=new Promise(resolve=>release=resolve);
|
||
/* complete 之后仍阻塞迭代器 EOF，之后追加垃圾，检查 ready 不会过早成功。 */ async function* trailing(){yield encode(rows);await gate.promise;yield new TextEncoder().encode('x\n');}
|
||
// trailing 先交出终态记录再等待，检测 ready 是否在底层迭代结束前过早解决；放行后追加垃圾使它最终拒绝。
|
||
const loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:trailing()})}),current:()=>true}),handle=loader.acquire(scope);let ready=false;handle.ready.then(()=>ready=true,()=>{});
|
||
try{while(handle.getSnapshot().chunks.length<2)await new Promise(resolve=>setImmediate(resolve));assert.equal(ready,false);release();await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');}finally{release();handle.release();loader.close();}
|
||
});
|
||
// abort 取消阻塞的 body reader；身份失效也拒绝下一次读取并取消底层流，分别核对 cancel 次数。
|
||
test('stream transport fences changed identity and abort cancels a blocked body reader',async()=>{
|
||
let current=true,cancelled=0;const abort=new AbortController();const transport=createFoundryTransport({accountId:account,isCurrentIdentity:()=>current,fetch:async()=>new Response(new ReadableStream({cancel(){cancelled++;}}),{headers:{'Content-Type':modelStreamType}})});
|
||
const reply=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1',signal:abort.signal});const iterator=reply.stream[Symbol.asyncIterator]();const pending=iterator.next();await Promise.resolve();abort.abort();await assert.rejects(pending);assert.equal(cancelled,1);
|
||
const other=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1'});current=false;await assert.rejects(()=>other.stream[Symbol.asyncIterator]().next(),{code:'identity_changed'});assert.equal(cancelled,2);
|
||
});
|
||
|
||
// 可选 canvasElementCounts 保持旧流兼容；非法清单或后续元素超过单 canvas 声明在发布前拒绝。
|
||
test('optional snapshot scope counts preserve legacy streams and reject invalid or exceeded coverage before publication',async()=>{
|
||
const rows=records();rows[0].canvasElementCounts={[root]:1};const seen=await collect(decodeModelStream(source(rows),root,registry,signal()));assert.deepEqual(seen[0].value.canvasElementCounts,{[root]:1});assert.equal(Object.isFrozen(seen[0].value.canvasElementCounts),true);
|
||
for(const manifest of [null,[],{}, {[root]:-1},{[root]:.5},{[root]:2},{[root]:1,[workspace]:0}]){const broken=records();broken[0].canvasElementCounts=manifest;await assert.rejects(()=>collect(decodeModelStream(source(broken),root,registry,signal())),{code:'invalid_model_coverage'});}
|
||
const broken=records();broken[0].canvases.push({id:workspace,parent:{canvasId:root,entryElementId:'frame'}});broken[0].versions[workspace]=1;broken[0].canvasElementCounts={[root]:0,[workspace]:1};const iterator=decodeModelStream(source(broken),root,registry,signal())[Symbol.asyncIterator]();assert.equal((await iterator.next()).value.value.receivedElements,0);await assert.rejects(iterator.next(),{code:'invalid_model_coverage'});
|
||
});
|
||
|
||
// elementOwners 声明尚未加载的 ID 与 owner；非法清单或后续未捕获 ID 在发布前拒绝，清单自身冻结。
|
||
test('identity manifest covers unloaded IDs and binds each received record to its captured owner',async()=>{
|
||
const rows=records();rows[0].canvasElementCounts={[root]:1};rows[0].elementOwners={a:root};const chunks=await collect(decodeModelStream(source(rows),root,registry,signal()));assert.deepEqual(chunks[0].value.elementOwners,{a:root});assert.ok(Object.isFrozen(chunks[0].value.elementOwners));
|
||
for(const manifest of [null,[],{}, {other:workspace},{a:root,other:root}]){const broken=structuredClone(rows);broken[0].elementOwners=manifest;await assert.rejects(()=>collect(decodeModelStream(source(broken),root,registry,signal())),{code:'invalid_model_identity'});}
|
||
const swapped=structuredClone(rows);swapped[0].elementOwners={unloaded:root};const iterator=decodeModelStream(source(swapped),root,registry,signal())[Symbol.asyncIterator]();assert.equal((await iterator.next()).value.value.receivedElements,0);await assert.rejects(iterator.next(),{code:'invalid_model_identity'});
|
||
});
|