Merge pull request #5405 from snyk/fix/stream-dep-graphs

fix: write dep-graph payloads to stdout stream

src/lib/ecosystems/test.ts

@@ -1,3 +1,5 @@
+import { Writable } from 'stream';
+
 import config from '../config';
 import { isCI } from '../is-ci';
 import { makeRequest } from '../request/promise';
@@ -9,7 +11,8 @@ import { getPlugin } from './plugins';
 import { TestDependenciesResponse } from '../snyk-test/legacy';
 import {
   assembleQueryString,
-  depGraphToOutputString,
+  printDepGraph,
+  shouldPrintDepGraph,
 } from '../snyk-test/common';
 import { getAuthHeader } from '../api-token';
 import { resolveAndTestFacts } from './resolve-test-facts';
@@ -46,9 +49,9 @@ export async function testEcosystem(
   }
   spinner.clearAll();
 
-  if (isUnmanagedEcosystem(ecosystem) && options['print-graph']) {
+  if (isUnmanagedEcosystem(ecosystem) && shouldPrintDepGraph(options)) {
     const [target] = paths;
-    return formatUnmanagedResults(results, target);
+    return printUnmanagedDepGraph(results, target, process.stdout);
   }
 
   const [testResults, errors] = await selectAndExecuteTestStrategy(
@@ -87,16 +90,17 @@ export async function selectAndExecuteTestStrategy(
     : await testDependencies(scanResultsByPath, options);
 }
 
-export async function formatUnmanagedResults(
+export async function printUnmanagedDepGraph(
   results: ScanResultsByPath,
   target: string,
+  destination: Writable,
 ): Promise<TestCommandResult> {
   const [result] = await getUnmanagedDepGraph(results);
   const depGraph = convertDepGraph(result);
 
-  return TestCommandResult.createJsonTestCommandResult(
-    depGraphToOutputString(depGraph, target),
-  );
+  await printDepGraph(depGraph, target, destination);
+
+  return TestCommandResult.createJsonTestCommandResult('');
 }
 
 async function testDependencies(

src/lib/snyk-test/assemble-payloads.ts

@@ -1,16 +1,21 @@
 import * as path from 'path';
+import { DepGraph } from '@snyk/dep-graph';
+
 import config from '../config';
 import { isCI } from '../is-ci';
 import { getPlugin } from '../ecosystems';
 import { Ecosystem, ContainerTarget, ScanResult } from '../ecosystems/types';
 import { Options, PolicyOptions, TestOptions } from '../types';
 import { Payload } from './types';
-import { assembleQueryString, depGraphToOutputString } from './common';
+import {
+  assembleQueryString,
+  printDepGraph,
+  shouldPrintDepGraph,
+} from './common';
 import { spinner } from '../spinner';
 import { findAndLoadPolicyForScanResult } from '../ecosystems/policy';
 import { getAuthHeader } from '../../lib/api-token';
 import { DockerImageNotFoundError } from '../errors';
-import { DepGraph } from '@snyk/dep-graph';
 
 export async function assembleEcosystemPayloads(
   ecosystem: Ecosystem,
@@ -53,17 +58,18 @@ export async function assembleEcosystemPayloads(
       scanResult.name =
         options['project-name'] || config.PROJECT_NAME || scanResult.name;
 
-      if (options['print-graph'] && !options['print-deps']) {
+      if (shouldPrintDepGraph(options)) {
+        spinner.clear<void>(spinnerLbl)();
+
         // not every scanResult has a 'depGraph' fact, for example the JAR
         // fingerprints. I don't think we have another option than to skip
         // those.
         const dg = scanResult.facts.find((dg) => dg.type === 'depGraph');
         if (dg) {
-          console.log(
-            depGraphToOutputString(
-              dg.data.toJSON(),
-              constructProjectName(scanResult),
-            ),
+          await printDepGraph(
+            dg.data.toJSON(),
+            constructProjectName(scanResult),
+            process.stdout,
           );
         }
       }

src/lib/snyk-test/common.ts

@@ -1,7 +1,11 @@
+import { Readable, Writable } from 'stream';
+import { JsonStreamStringify } from 'json-stream-stringify';
+import { DepGraphData } from '@snyk/dep-graph';
+
 import config from '../config';
 import { color } from '../theme';
-import { DepGraphData } from '@snyk/dep-graph';
-import { jsonStringifyLargeObject } from '../json';
+import { Options } from '../types';
+import { ConcatStream } from '../stream';
 
 export function assembleQueryString(options) {
   const org = options.org || config.org || null;
@@ -71,15 +75,27 @@ export type FailOn = 'all' | 'upgradable' | 'patchable';
 export const RETRY_ATTEMPTS = 3;
 export const RETRY_DELAY = 500;
 
-// depGraphData formats the given depGrahData with the targetName as expected by
-// the `depgraph` CLI workflow.
-export function depGraphToOutputString(
-  dg: DepGraphData,
+/**
+ * printDepGraph writes the given dep-graph and target name to the destination
+ * stream as expected by the `depgraph` CLI workflow.
+ */
+export async function printDepGraph(
+  depGraph: DepGraphData,
   targetName: string,
-): string {
-  return `DepGraph data:
-${jsonStringifyLargeObject(dg)}
-DepGraph target:
-${targetName}
-DepGraph end`;
+  destination: Writable,
+): Promise<void> {
+  return new Promise((res, rej) => {
+    new ConcatStream(
+      Readable.from('DepGraph data:\n'),
+      new JsonStreamStringify(depGraph),
+      Readable.from(`\nDepGraph target:\n${targetName}\nDepGraph end\n\n`),
+    )
+      .on('end', res)
+      .on('error', rej)
+      .pipe(destination);
+  });
+}
+
+export function shouldPrintDepGraph(opts: Options): boolean {
+  return opts['print-graph'] && !opts['print-deps'];
 }

src/lib/snyk-test/run-test.ts

@@ -35,8 +35,13 @@ import {
 } from '../errors';
 import * as snyk from '../';
 import { isCI } from '../is-ci';
-import * as common from './common';
-import { RETRY_ATTEMPTS, RETRY_DELAY } from './common';
+import {
+  RETRY_ATTEMPTS,
+  RETRY_DELAY,
+  printDepGraph,
+  assembleQueryString,
+  shouldPrintDepGraph,
+} from './common';
 import config from '../config';
 import * as analytics from '../analytics';
 import { maybePrintDepGraph, maybePrintDepTree } from '../print-deps';
@@ -341,7 +346,7 @@ export async function runTest(
   try {
     const payloads = await assemblePayloads(root, options, featureFlags);
 
-    if (options['print-graph'] && !options['print-deps']) {
+    if (shouldPrintDepGraph(options)) {
       const results: TestResult[] = [];
       return results;
     }
@@ -754,8 +759,7 @@ async function assembleLocalPayloads(
         ? (pkg as depGraphLib.DepGraph).rootPkg.name
         : (pkg as DepTree).name;
 
-      // print dep graph if `--print-graph` is set
-      if (options['print-graph'] && !options['print-deps']) {
+      if (shouldPrintDepGraph(options)) {
         spinner.clear<void>(spinnerLbl)();
         let root: depGraphLib.DepGraph;
         if (scannedProject.depGraph) {
@@ -768,9 +772,7 @@ async function assembleLocalPayloads(
           );
         }
 
-        console.log(
-          common.depGraphToOutputString(root.toJSON(), targetFile || ''),
-        );
+        await printDepGraph(root.toJSON(), targetFile || '', process.stdout);
       }
 
       const body: PayloadBody = {
@@ -829,7 +831,7 @@ async function assembleLocalPayloads(
           'x-is-ci': isCI(),
           authorization: getAuthHeader(),
         },
-        qs: common.assembleQueryString(options),
+        qs: assembleQueryString(options),
         body,
       };
 
@@ -871,7 +873,7 @@ async function assembleRemotePayloads(root, options): Promise<Payload[]> {
     {
       method: 'GET',
       url,
-      qs: common.assembleQueryString(options),
+      qs: assembleQueryString(options),
       json: true,
       headers: {
         'x-is-ci': isCI(),

src/lib/stream.ts

@@ -0,0 +1,37 @@
+import { Readable } from 'stream';
+
+export class ConcatStream extends Readable {
+  private current: Readable | undefined;
+  private queue: Readable[] = [];
+
+  constructor(...streams: Readable[]) {
+    super({ objectMode: false }); // Adjust objectMode if needed
+    this.queue.push(...streams);
+  }
+
+  append(...streams: Readable[]): void {
+    this.queue.push(...streams);
+    if (!this.current) {
+      this._read();
+    }
+  }
+
+  _read(size?: number): void {
+    if (this.current) {
+      return;
+    }
+
+    this.current = this.queue.shift();
+    if (!this.current) {
+      this.push(null);
+      return;
+    }
+
+    this.current.on('data', (chunk) => this.push(chunk));
+    this.current.on('end', () => {
+      this.current = undefined;
+      this._read(size);
+    });
+    this.current.on('error', (err) => this.emit('error', err));
+  }
+}

test/jest/unit/lib/ecosystems/__snapshots__/common.spec.ts.snap

@@ -0,0 +1,11 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`printUnmanagedDepGraph fn should print the dep-graph 1`] = `
+"DepGraph data:
+{"schemaVersion":"1.2.0","pkgManager":{"name":"cpp"},"pkgs":[{"id":"[email protected]","info":{"name":"root-node","version":"0.0.0"}},{"id":"https://ftp.gnu.org|[email protected]","info":{"name":"https://ftp.gnu.org|cpio","version":"2.12"}}],"graph":{"rootNodeId":"root-node","nodes":[{"nodeId":"root-node","pkgId":"[email protected]","deps":[{"nodeId":"https://ftp.gnu.org|[email protected]"}]},{"nodeId":"https://ftp.gnu.org|[email protected]","pkgId":"https://ftp.gnu.org|[email protected]","deps":[]}]}}
+DepGraph target:
+foo/bar
+DepGraph end
+
+"
+`;

test/jest/unit/lib/ecosystems/common.spec.ts

@@ -1,7 +1,9 @@
+import { Writable } from 'stream';
+
 import { isUnmanagedEcosystem } from '../../../../../src/lib/ecosystems/common';
 import { handleProcessingStatus } from '../../../../../src/lib/polling/common';
 import { FailedToRunTestError } from '../../../../../src/lib/errors';
-import { formatUnmanagedResults } from '../../../../../src/lib/ecosystems/test';
+import { printUnmanagedDepGraph } from '../../../../../src/lib/ecosystems/test';
 import * as utils from '../../../../../src/lib/ecosystems/unmanaged/utils';
 import { DepGraphDataOpenAPI } from '../../../../../src/lib/ecosystems/unmanaged/types';
 
@@ -34,8 +36,8 @@ describe('handleProcessingStatus fn', () => {
   );
 });
 
-describe('formatUnmanagedResults fn', () => {
-  it('should return formatted results', async () => {
+describe('printUnmanagedDepGraph fn', () => {
+  it('should print the dep-graph', async () => {
     const mockedUnmanagedDepGraph: DepGraphDataOpenAPI = {
       schema_version: '1.2.0',
       pkg_manager: {
@@ -80,12 +82,17 @@ describe('formatUnmanagedResults fn', () => {
     jest
       .spyOn(utils, 'getUnmanagedDepGraph')
       .mockImplementation(() => Promise.resolve([mockedUnmanagedDepGraph]));
+    let buffer = Buffer.alloc(0);
+    const mockDest = new Writable({
+      write(chunk, _, next) {
+        buffer = Buffer.concat([buffer, chunk]);
+        next();
+      },
+    });
 
-    const { result } = await formatUnmanagedResults({}, 'foo/bar');
+    const { result } = await printUnmanagedDepGraph({}, 'foo/bar', mockDest);
 
-    expect(result.includes('DepGraph data:')).toBeTruthy();
-    expect(result.includes('DepGraph target:')).toBeTruthy();
-    expect(result.includes('foo/bar')).toBeTruthy();
-    expect(result.includes('DepGraph end')).toBeTruthy();
+    expect(result).toBe('');
+    expect(buffer.toString()).toMatchSnapshot();
   });
 });

test/jest/unit/lib/stream.spec.ts

@@ -0,0 +1,31 @@
+import { Readable, Writable } from 'stream';
+
+import { ConcatStream } from '../../../../src/lib/stream';
+
+describe('ConcatStream', () => {
+  it('should create a readable stream', () => {
+    const stream = new ConcatStream();
+
+    expect(stream).toBeInstanceOf(Readable);
+  });
+
+  it('should concatenate readable streams', async () => {
+    const stream = new ConcatStream();
+    const chunks = jest.fn();
+    const out = new Writable({
+      write: (chunk, enc, done) => {
+        chunks(chunk.toString());
+        done();
+      },
+    });
+
+    stream.append(Readable.from('foo'), Readable.from('bar'));
+
+    await new Promise((res) => {
+      stream.pipe(out).on('finish', res);
+    });
+
+    expect(chunks).toHaveBeenCalledWith('foo');
+    expect(chunks).toHaveBeenCalledWith('bar');
+  });
+});