import { expect , use } from 'chai' ; import * as chaiAsPromised from 'chai-as-promised' ; import { CancellationTokenSource } from 'vscode' ; import { createDeferred } from '../../../src/common/helpers' ; import { BufferDecoder } from '../../../src/common/process/decoder' ; import { ProcessService } from '../../../src/common/process/proc' ; import { PYTHON_PATH } from '../../common' ; use ( chaiAsPromised ) ; suite ( 'ProcessService' , ( ) => { const pythonPath = PYTHON_PATH ; test ( 'execObservable should stream output with new lines' , function ( done ) { this . timeout ( 10000 ) ; const procService = new ProcessService ( new BufferDecoder ( ) ) ; const pythonCode = [ 'import sys' , 'import time' , 'print("1")' , 'sys.stdout.flush()' , 'time.sleep(2)' , 'print("2")' , 'sys.stdout.flush()' , 'time.sleep(2)' , 'print("3")' , 'sys.stdout.flush()' , 'time.sleep(2)' ] ; const result = procService . execObservable ( pythonPath , [ '-c' , pythonCode . join ( ';' ) ] ) ; const outputs = [ '1' , '2' , '3' ] ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined' ) ; result . out . subscribe ( output => { if ( output . out . trim ( ) . length === 0 ) { return ; } const expectedValue = outputs . shift ( ) ; if ( expectedValue !== output . out . trim ( ) && expectedValue === output . out ) { done ( `Received value ${ output . out } is not same as the expectd value ${ expectedValue }` ) ; } if ( output . source !== 'stdout' ) { done ( `Source is not stdout. Value received is ${ output . source }` ) ; } } , done , done ) ; } ) ; test ( 'execObservable should stream output without new lines' , function ( done ) { this . timeout ( 10000 ) ; const procService = new ProcessService ( new BufferDecoder ( ) ) ; const pythonCode = [ 'import sys' , 'import time' , 'sys.stdout.write("1")' , 'sys.stdout.flush()' , 'time.sleep(2)' , 'sys.stdout.write("2")' , 'sys.stdout.flush()' , 'time.sleep(2)' , 'sys.stdout.write("3")' , 'sys.stdout.flush()' , 'time.sleep(2)' ] ; const result = procService . execObservable ( pythonPath , [ '-c' , pythonCode . join ( ';' ) ] ) ; const outputs = [ '1' , '2' , '3' ] ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined' ) ; result . out . subscribe ( output => { if ( output . out . trim ( ) . length === 0 ) { return ; } const expectedValue = outputs . shift ( ) ; if ( expectedValue !== output . out ) { done ( `Received value ${ output . out } is not same as the expectd value ${ expectedValue }` ) ; } if ( output . source !== 'stdout' ) { done ( `Source is not stdout. Value received is ${ output . source }` ) ; } } , done , done ) ; } ) ; test ( 'execObservable should end when cancellationToken is cancelled' , function ( done ) { this . timeout ( 15000 ) ; const procService = new ProcessService ( new BufferDecoder ( ) ) ; const pythonCode = [ 'import sys' , 'import time' , 'print("1")' , 'sys.stdout.flush()' , 'time.sleep(10)' , 'print("2")' , 'sys.stdout.flush()' , 'time.sleep(2)' ] ; const cancellationToken = new CancellationTokenSource ( ) ; const result = procService . execObservable ( pythonPath , [ '-c' , pythonCode . join ( ';' ) ] , { token : cancellationToken . token } ) ; const def = createDeferred ( ) ; def . promise . then ( done ) . catch ( done ) ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined' ) ; result . out . subscribe ( output => { const value = output . out . trim ( ) ; if ( value === '1' ) { cancellationToken . cancel ( ) ; } else { if ( ! def . completed ) { def . reject ( 'Output received when we shouldn\'t have.' ) ; } } } , done , ( ) => { if ( def . completed ) { return ; } if ( cancellationToken . token . isCancellationRequested ) { def . resolve ( ) ; } else { def . reject ( 'Program terminated even before cancelling it.' ) ; } } ) ; } ) ; test ( 'execObservable should end when process is killed' , function ( done ) { this . timeout ( 15000 ) ; const procService = new ProcessService ( new BufferDecoder ( ) ) ; const pythonCode = [ 'import sys' , 'import time' , 'print("1")' , 'sys.stdout.flush()' , 'time.sleep(10)' , 'print("2")' , 'sys.stdout.flush()' , 'time.sleep(2)' ] ; const cancellationToken = new CancellationTokenSource ( ) ; const result = procService . execObservable ( pythonPath , [ '-c' , pythonCode . join ( ';' ) ] , { token : cancellationToken . token } ) ; let procKilled = false ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined' ) ; result . out . subscribe ( output => { const value = output . out . trim ( ) ; if ( value . length === 0 ) { return ; } if ( value === '1' ) { procKilled = true ; result . proc . kill ( ) ; } else { done ( 'Output received when we shouldn\'t have.' ) ; } } , done , ( ) => { const errorMsg = procKilled ? undefined : 'Program terminated even before killing it.' ; done ( errorMsg ) ; } ) ; } ) ; test ( 'execObservable should stream stdout and stderr separately' , function ( done ) { this . timeout ( 20000 ) ; const procService = new ProcessService ( new BufferDecoder ( ) ) ; const pythonCode = [ 'import sys' , 'import time' , 'print("1")' , 'sys.stdout.flush()' , 'time.sleep(2)' , 'sys.stderr.write("a")' , 'sys.stderr.flush()' , 'time.sleep(2)' , 'print("2")' , 'sys.stdout.flush()' , 'time.sleep(2)' , 'sys.stderr.write("b")' , 'sys.stderr.flush()' , 'time.sleep(2)' , 'print("3")' , 'sys.stdout.flush()' , 'time.sleep(2)' , 'sys.stderr.write("c")' , 'sys.stderr.flush()' , 'time.sleep(2)' ] ; const result = procService . execObservable ( pythonPath , [ '-c' , pythonCode . join ( ';' ) ] ) ; const outputs = [ { out : '1' , source : 'stdout' } , { out : 'a' , source : 'stderr' } , { out : '2' , source : 'stdout' } , { out : 'b' , source : 'stderr' } , { out : '3' , source : 'stdout' } , { out : 'c' , source : 'stderr' } ] ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined' ) ; result . out . subscribe ( output => { const value = output . out . trim ( ) ; if ( value . length === 0 ) { return ; } const expectedOutput = outputs . shift ( ) ! ; expect ( value ) . to . be . equal ( expectedOutput . out , 'Expected output is incorrect' ) ; expect ( output . source ) . to . be . equal ( expectedOutput . source , 'Expected sopurce is incorrect' ) ; } , done , done ) ; } ) ; test ( 'execObservable should send stdout and stderr streams separately' , function ( done ) { this . timeout ( 7000 ) ; const procService = new ProcessService ( new BufferDecoder ( ) ) ; const pythonCode = [ 'import sys' , 'import time' , 'print("1")' , 'sys.stdout.flush()' , 'time.sleep(1)' , 'sys.stderr.write("a")' , 'sys.stderr.flush()' , 'time.sleep(1)' , 'print("2")' , 'sys.stdout.flush()' , 'time.sleep(1)' , 'sys.stderr.write("b")' , 'sys.stderr.flush()' , 'time.sleep(1)' , 'print("3")' , 'sys.stdout.flush()' , 'time.sleep(1)' , 'sys.stderr.write("c")' , 'sys.stderr.flush()' , 'time.sleep(1)' ] ; const result = procService . execObservable ( pythonPath , [ '-c' , pythonCode . join ( ';' ) ] , { mergeStdOutErr : true } ) ; const outputs = [ { out : '1' , source : 'stdout' } , { out : 'a' , source : 'stderr' } , { out : '2' , source : 'stdout' } , { out : 'b' , source : 'stderr' } , { out : '3' , source : 'stdout' } , { out : 'c' , source : 'stderr' } ] ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined' ) ; result . out . subscribe ( output => { const value = output . out . trim ( ) ; if ( value . length === 0 ) { return ; } const expectedOutput = outputs . shift ( ) ! ; expect ( value ) . to . be . equal ( expectedOutput . out , 'Expected output is incorrect' ) ; expect ( output . source ) . to . be . equal ( expectedOutput . source , 'Expected sopurce is incorrect' ) ; } , done , done ) ; } ) ; test ( 'execObservable should throw an error with stderr output' , ( done ) => { const procService = new ProcessService ( new BufferDecoder ( ) ) ; const pythonCode = [ 'import sys' , 'sys.stderr.write("a")' , 'sys.stderr.flush()' ] ; const result = procService . execObservable ( pythonPath , [ '-c' , pythonCode . join ( ';' ) ] , { throwOnStdErr : true } ) ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined.' ) ; result . out . subscribe ( ( ) => { done ( 'Output received, when we\'re expecting an error to be thrown.' ) ; } , ( ex ) => { expect ( ex ) . to . have . property ( 'message' , 'a' , 'Invalid error thrown' ) ; done ( ) ; } , ( ) => { done ( 'Completed, when we\'re expecting an error to be thrown.' ) ; } ) ; } ) ; test ( 'execObservable should throw an error when spawn file not found' , ( done ) => { const procService = new ProcessService ( new BufferDecoder ( ) ) ; const result = procService . execObservable ( Date . now ( ) . toString ( ) , [ ] ) ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined.' ) ; result . out . subscribe ( ( ) => { done ( 'Output received, when we\'re expecting an error to be thrown.' ) ; } , ex => { expect ( ex ) . to . have . property ( 'code' , 'ENOENT' , 'Invalid error code' ) ; done ( ) ; } , ( ) => { done ( 'Completed, when we\'re expecting an error to be thrown.' ) ; } ) ; } ) ; test ( 'execObservable should exit without no output' , ( done ) => { const procService = new ProcessService ( new BufferDecoder ( ) ) ; const result = procService . execObservable ( pythonPath , [ '-c' , 'import sys' , 'sys.exit()' ] ) ; expect ( result ) . not . to . be . an ( 'undefined' , 'result is undefined.' ) ; result . out . subscribe ( output => { done ( `Output received, when we\'re not expecting any, ${ JSON . stringify ( output ) }` ) ; } , done , done ) ; } ) ; } ) ;