import { describe, it, expect } from 'vitest'; import { PipeWatcher } from '../src/pipe-watcher'; import type { HookProcessBaseInfo } from '../src/injector'; function setupWatcher(initial: { processes: HookProcessBaseInfo[]; live: Set }) { let processes = initial.processes; let live = initial.live; const watcher = new PipeWatcher({ listProcesses: () => processes, // Defensively copy so the watcher can't mutate the test's reference. listLivePipes: async () => new Set(live), intervalMs: 60_000, // disable the internal timer; tests drive ticks manually }); return { watcher, setProcesses: (p: HookProcessBaseInfo[]) => { processes = p; }, setLive: (s: Set) => { live = s; }, }; } type EventRow = [string, unknown]; function captureEvents(watcher: PipeWatcher): EventRow[] { const events: EventRow[] = []; watcher.on('process-discovered', (info) => events.push(['discovered', (info as HookProcessBaseInfo).pid]), ); watcher.on('process-gone', (pid) => events.push(['gone', pid])); watcher.on('pipe-up', (pid) => events.push(['up', pid])); watcher.on('pipe-down', (pid) => events.push(['down', pid])); return events; } describe('PipeWatcher', () => { it('passes the current process snapshot to pipe discovery', async () => { const processes = [ { pid: 1234, name: 'qq', path: '/opt/QQ/qq' }, { pid: 5678, name: 'qq', path: '/opt/QQ/qq' }, ]; let observed: readonly HookProcessBaseInfo[] | undefined; const watcher = new PipeWatcher({ listProcesses: () => processes, listLivePipes: async (current) => { observed = current; return new Set(); }, intervalMs: 60_000, }); await watcher.start(); expect(observed).toEqual(processes); watcher.stop(); }); it('start() emits process-discovered + pipe-up for live PIDs on first tick', async () => { const ctx = setupWatcher({ processes: [{ pid: 1234, name: 'QQ.exe', path: '' }], live: new Set([1234]), }); const events = captureEvents(ctx.watcher); await ctx.watcher.start(); expect(events).toEqual([ ['discovered', 1234], ['up', 1234], ]); expect(ctx.watcher.isPipeLive(1234)).toBe(true); expect(ctx.watcher.isProcessAlive(1234)).toBe(true); ctx.watcher.stop(); }); it('pipe-down emitted when a previously-live pipe disappears', async () => { const ctx = setupWatcher({ processes: [{ pid: 1234, name: 'QQ.exe', path: '' }], live: new Set([1234]), }); await ctx.watcher.start(); const events = captureEvents(ctx.watcher); ctx.setLive(new Set()); await ctx.watcher.tickNow(); expect(events).toEqual([['down', 1234]]); expect(ctx.watcher.isPipeLive(1234)).toBe(false); ctx.watcher.stop(); }); it('process-gone + pipe-down both emitted (pipe-down first) when QQ.exe is killed with a live pipe', async () => { const ctx = setupWatcher({ processes: [{ pid: 1234, name: 'QQ.exe', path: '' }], live: new Set([1234]), }); await ctx.watcher.start(); const events = captureEvents(ctx.watcher); ctx.setProcesses([]); ctx.setLive(new Set()); await ctx.watcher.tickNow(); expect(events).toEqual([ ['down', 1234], ['gone', 1234], ]); ctx.watcher.stop(); }); it('ignores live pipes whose process is not running', async () => { const ctx = setupWatcher({ processes: [], live: new Set([9999]), }); const events = captureEvents(ctx.watcher); await ctx.watcher.start(); expect(events).toEqual([]); expect(ctx.watcher.isPipeLive(9999)).toBe(false); ctx.watcher.stop(); }); it('emits pipe-up again after a pipe-down → pipe-up cycle', async () => { const ctx = setupWatcher({ processes: [{ pid: 1234, name: 'QQ.exe', path: '' }], live: new Set(), }); await ctx.watcher.start(); const events = captureEvents(ctx.watcher); ctx.setLive(new Set([1234])); await ctx.watcher.tickNow(); expect(events).toEqual([['up', 1234]]); ctx.setLive(new Set()); await ctx.watcher.tickNow(); expect(events).toEqual([ ['up', 1234], ['down', 1234], ]); ctx.setLive(new Set([1234])); await ctx.watcher.tickNow(); expect(events).toEqual([ ['up', 1234], ['down', 1234], ['up', 1234], ]); ctx.watcher.stop(); }); it('does not double-emit on idempotent ticks (no state change)', async () => { const ctx = setupWatcher({ processes: [{ pid: 1234, name: 'QQ.exe', path: '' }], live: new Set([1234]), }); await ctx.watcher.start(); const events = captureEvents(ctx.watcher); await ctx.watcher.tickNow(); await ctx.watcher.tickNow(); expect(events).toEqual([]); ctx.watcher.stop(); }); it('emits "tick" after every poll', async () => { const ctx = setupWatcher({ processes: [], live: new Set() }); let ticks = 0; ctx.watcher.on('tick', () => { ticks++; }); await ctx.watcher.start(); expect(ticks).toBe(1); await ctx.watcher.tickNow(); expect(ticks).toBe(2); ctx.watcher.stop(); }); it('wake() pulls the next tick forward', async () => { const ctx = setupWatcher({ processes: [{ pid: 1234, name: 'QQ.exe', path: '' }], live: new Set(), }); await ctx.watcher.start(); const events = captureEvents(ctx.watcher); // Wait on 'tick' rather than guessing how many event-loop turns // setTimeout(0) needs (it's clamped to ~1ms and races setImmediate). const nextTick = new Promise((resolve) => ctx.watcher.once('tick', () => resolve())); ctx.setLive(new Set([1234])); ctx.watcher.wake(); await nextTick; expect(events).toEqual([['up', 1234]]); ctx.watcher.stop(); }); it('survives a throwing listProcesses without crashing the watcher', async () => { let processes: HookProcessBaseInfo[] = []; let live = new Set(); let shouldThrow = true; const watcher = new PipeWatcher({ listProcesses: () => { if (shouldThrow) { throw new Error('boom'); } return processes; }, listLivePipes: async () => new Set(live), intervalMs: 60_000, }); await watcher.start(); expect(watcher.isProcessAlive(1234)).toBe(false); shouldThrow = false; processes = [{ pid: 1234, name: 'QQ.exe', path: '' }]; live = new Set([1234]); const events = captureEvents(watcher); await watcher.tickNow(); expect(events).toEqual([ ['discovered', 1234], ['up', 1234], ]); watcher.stop(); }); it('preserves live sessions when pipe discovery temporarily fails', async () => { let processes = [{ pid: 1234, name: 'QQ.exe', path: '' }]; let failPipeDiscovery = false; const watcher = new PipeWatcher({ listProcesses: () => processes, listLivePipes: async () => { if (failPipeDiscovery) { throw new Error('permission denied'); } return new Set([1234]); }, intervalMs: 60_000, }); await watcher.start(); expect(watcher.isPipeLive(1234)).toBe(true); const events = captureEvents(watcher); failPipeDiscovery = true; await watcher.tickNow(); expect(events).toEqual([]); expect(watcher.isProcessAlive(1234)).toBe(true); expect(watcher.isPipeLive(1234)).toBe(true); processes = []; await watcher.tickNow(); expect(events).toEqual([ ['down', 1234], ['gone', 1234], ]); expect(watcher.isProcessAlive(1234)).toBe(false); expect(watcher.isPipeLive(1234)).toBe(false); watcher.stop(); }); it('UNKNOWN (null) enumeration keeps prior state — no false process-gone (issue #158)', async () => { let result: HookProcessBaseInfo[] | null = [{ pid: 1234, name: 'qq', path: '' }]; let live = new Set([1234]); const watcher = new PipeWatcher({ listProcesses: () => result, listLivePipes: async () => new Set(live), intervalMs: 60_000, }); await watcher.start(); expect(watcher.isProcessAlive(1234)).toBe(true); expect(watcher.isPipeLive(1234)).toBe(true); const events = captureEvents(watcher); // Enumeration goes UNKNOWN (e.g. native /proc walk timed out during a QQ // hot-update). Prior process + pipe state MUST be preserved — no teardown. result = null; await watcher.tickNow(); expect(events).toEqual([]); expect(watcher.isProcessAlive(1234)).toBe(true); expect(watcher.isPipeLive(1234)).toBe(true); // Enumeration recovers with the same set → still no churn. result = [{ pid: 1234, name: 'qq', path: '' }]; await watcher.tickNow(); expect(events).toEqual([]); // A REAL disappearance (empty list, not null) still fires the teardown. result = []; live = new Set(); await watcher.tickNow(); expect(events).toEqual([ ['down', 1234], ['gone', 1234], ]); watcher.stop(); }); });