stack/packages/bus/tests/peer-transport.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { connect } from 'node:net';
import { mkdtempSync,rmSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { serve } from '../src/server.mjs';
import { Client } from '../src/client.mjs';
import { BusError } from '../src/broker.mjs';
const request=(path,body)=>new Promise((resolve,reject)=>{const s=connect(path);let out='';s.on('error',reject);s.on('data',b=>out+=b);s.on('end',()=>resolve(JSON.parse(out)));s.end(JSON.stringify(body)+'\n');});
test('proof receives actual peer, cap traffic never invokes helper and transient cap disconnects',async t=>{
 const root=mkdtempSync(join(tmpdir(),'g196-s-')),path=join(root,'s');let peers=0,auth=0;const disconnected=[];
 const broker={request:()=>({ok:'value'}),disconnect:c=>disconnected.push(c)};
 const reader={read:async socket=>{peers++;return {pid:process.pid,uid:process.getuid(),gid:process.getgid()};},close(){}};
 const server=await serve({path,broker,peerReader:reader,authenticateHuman:(proof,peer)=>{auth++;assert.equal(peer.pid,process.pid);return 'transient';}});
 t.after(async()=>{await server.close();rmSync(root,{recursive:true,force:true});});
 assert.equal((await request(path,{human:{pid:process.pid},verb:'agents'})).ok,true);
 assert.equal(peers,1);assert.equal(auth,1);assert.ok(disconnected.includes('transient'));
 assert.equal((await request(path,{cap:'existing',verb:'agents'})).ok,true);
 assert.equal(peers,1);
});
test('proof helper error and disconnect during verification never dispatch or bind',async t=>{
 const root=mkdtempSync(join(tmpdir(),'g196-s-'));let auth=0,calls=0;
 const broker={request:()=>calls++,disconnect(){}};
 const server=await serve({path:join(root,'s'),broker,peerReader:{read:async()=>{throw Error('helper');},close(){}},authenticateHuman:()=>{auth++;return 'x';}});
 t.after(async()=>{await server.close();rmSync(root,{recursive:true,force:true});});
 assert.equal((await request(join(root,'s'),{human:{pid:process.pid},verb:'agents'})).error,'peercred-required');
 assert.equal(auth,0);assert.equal(calls,0);
});

test('Client receives explicit proof refusal codes without dispatch, not outcome-unknown',async t=>{
 for(const failure of ['human-required','peercred-required']) {
  const root=mkdtempSync(join(tmpdir(),'g196-refuse-')),path=join(root,'s');let calls=0;
  const server=await serve({path,broker:{request(){calls++;},disconnect(){}},
   ...(failure==='peercred-required'?{}:{authenticateHuman(){throw new BusError('human-required');}}),
   peerReader:failure==='peercred-required'?{read:async()=>{throw Error('helper');},close(){}}:undefined,
   ...(failure==='peercred-required'?{authenticateHuman(){throw Error('must not authenticate');}}:{})});
  try {
   const client=new Client({path,human:{pid:process.pid,business:'fixture',startTime:'0',nonce:'a'.repeat(64)}});
   await assert.rejects(client.call('agents'),e=>e.code===failure);
   assert.equal(calls,0);
  }finally{await server.close();rmSync(root,{recursive:true,force:true});}
 }
});

test('asynchronous proof calls release their transient session on success, failure and timeout',async t=>{
 const {TASK_VERBS,BusError}=await import('../src/broker.mjs');
 for(const mode of ['success','failure','timeout']) {
  const root=mkdtempSync(join(tmpdir(),'g196-async-')),path=join(root,'s'),disconnected=[];
  const broker={disconnect:cap=>disconnected.push(cap),requestTask:()=>mode==='success'?Promise.resolve('done'):mode==='failure'?Promise.reject(new BusError('fixture-failed')):new Promise(()=>{})};
  const server=await serve({path,broker,tasks:{handle(){},timeout:25},peerReader:{read:async()=>({pid:process.pid,uid:process.getuid(),gid:process.getgid()}),close(){}},authenticateHuman:()=>mode});
  try {
   const response=await request(path,{human:{pid:process.pid},verb:TASK_VERBS[0]});
   assert.equal(response.ok,mode==='success');
   if(mode==='timeout')assert.equal(response.error,'request-timeout');
   assert.deepEqual(disconnected,[mode]);
  }finally{await server.close();rmSync(root,{recursive:true,force:true});}
 }
});