69 lines
9.6 KiB
JavaScript
69 lines
9.6 KiB
JavaScript
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';
|
|
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;
|
|
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}};
|
|
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}];
|
|
const encode=rows=>new TextEncoder().encode(rows.map(row=>JSON.stringify(row)).join('\n')+'\n');
|
|
async function* source(rows){yield encode(rows);}
|
|
const collect=async iterable=>{const result=[];for await(const item of iterable)result.push(item);return result;};
|
|
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'};
|
|
|
|
test('stream decoder frames split UTF8/newlines, emits deltas and validates one final complete model',async()=>{
|
|
const bytes=encode(records());async function* split(){for(const byte of bytes)yield Uint8Array.of(byte);}
|
|
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]));
|
|
});
|
|
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())));
|
|
async function* truncated(){yield encode(records()).slice(0,-1);}await assert.rejects(()=>collect(decodeModelStream(truncated(),root,registry,signal())),{code:'incomplete_model_stream'});
|
|
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'});
|
|
});
|
|
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();}
|
|
});
|
|
test('malformed tails retain read deltas without yielding an authoritative model',async()=>{
|
|
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();}
|
|
});
|
|
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);
|
|
async function* trailing(){yield encode(rows);await gate.promise;yield new TextEncoder().encode('x\n');}
|
|
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();}
|
|
});
|
|
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);
|
|
});
|
|
|
|
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'});
|
|
});
|
|
|
|
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'});
|
|
});
|