Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
task:
run:
echo this is "${i}"
sleep ${i}
[20]
input: for_each={'i': range(2)}
task:
run:
echo this aa is "${i}"
sleep ${i}
"""
# these should be automatically rerun by the frontend
long_execute(kc=kc, code=code)
wait_for_idle(kc)
iopub = kc.iopub_channel
# create a data frame
execute(kc=kc, code='''
import numpy as np
mat_var = np.matrix([[1,2],[3,4]])
''')
clear_channels(iopub)
execute(kc=kc, code="%use Julia")
wait_for_idle(kc)
execute(kc=kc, code="%get mat_var")
wait_for_idle(kc)
execute(kc=kc, code="size(mat_var)")
res = get_display_data(iopub)
self.assertEqual(res, '(2, 2)')
execute(kc=kc, code="%use sos")
wait_for_idle(kc)
#
def testWith(self):
'''Test magic with'''
with sos_kernel() as kc:
iopub = kc.iopub_channel
# create a data frame
execute(kc=kc, code='var = [1, 2, 3, 4]')
wait_for_idle(kc)
execute(kc=kc, code='%with R -i var -o m\nm=mean(var)')
wait_for_idle(kc)
execute(kc=kc, code="%dict m")
res = get_result(iopub)
self.assertEqual(res['m'], 2.5)
execute(kc=kc, code="%use sos")
wait_for_idle(kc)
def testCD(self):
with sos_kernel() as kc:
iopub = kc.iopub_channel
execute(kc=kc, code="%cd ..")
wait_for_idle(kc)
execute(kc=kc, code="import os\nprint(os.getcwd())")
stdout, stderr = get_std_output(iopub)
self.assertFalse(stdout.strip().endswith('jupyter'))
self.assertEqual(stderr, '')
execute(kc=kc, code="%cd jupyter")
def testRerun(self):
with sos_kernel() as kc:
iopub = kc.iopub_channel
execute(kc=kc, code='''
%run
parameter: a=10
[default]
b = a
''')
wait_for_idle(kc)
#
execute(kc=kc, code='''
%rerun --a 20
''')
wait_for_idle(kc)
execute(kc=kc, code="b")
res = get_result(iopub)
self.assertEqual(res, 20)
iopub = kc.iopub_channel
# create a data frame
execute(kc=kc, code='''
%use JavaScript
null_var = null
num_var = 123
num_arr_var = [1, 2, 3]
logic_var = true
logic_arr_var = [true, false, true]
char_var = '1\"23'
char_arr_var = ['1', '2', '3']
list_var = [1, 2, '3']
named_list_var = {a:1, b:2, c:3}
recursive_var = {a:1, b: {c:3, d:'whatever'}}
''')
wait_for_idle(kc)
execute(kc=kc, code="""
%put null_var num_var num_arr_var logic_var logic_arr_var char_var char_arr_var list_var named_list_var recursive_var
%dict null_var num_var num_arr_var logic_var logic_arr_var char_var char_arr_var list_var named_list_var recursive_var
""")
res = get_result(iopub)
self.assertEqual(res['null_var'], None)
self.assertEqual(res['num_var'], 123)
self.assertEqual(res['num_arr_var'], [1,2,3])
self.assertEqual(res['logic_var'], True)
self.assertEqual(res['logic_arr_var'], [True, False, True])
self.assertEqual(res['char_var'], '1"23')
self.assertEqual(res['char_arr_var'], ['1', '2', '3'])
self.assertEqual(res['list_var'], [1,2,'3'])
self.assertEqual(res['named_list_var'], {'a': 1, 'b': 2, 'c': 3})
self.assertEqual(res['recursive_var'], {'a': 1, 'b': {'c': 3, 'd': 'whatever'}})
execute(kc=kc, code="%use sos")
execute(kc=kc, code='''
null_var = None
num_var = 123
import numpy
num_arr_var = numpy.array([1, 2, 3])
logic_var = True
logic_arr_var = [True, False, True]
char_var = '1"23'
char_arr_var = ['1', '2', '3']
list_var = [1, 2, '3']
dict_var = dict(a=1, b=2, c='3')
set_var = {1, 2, '3'}
mat_var = numpy.matrix([[1,2],[3,4]])
recursive_var = {'a': {'b': 123}, 'c': True}
''')
wait_for_idle(kc)
execute(kc=kc, code='''
%use JavaScript
%get null_var num_var num_arr_var logic_var logic_arr_var char_var char_arr_var mat_var set_var list_var dict_var recursive_var
%dict -r
%put null_var num_var num_arr_var logic_var logic_arr_var char_var char_arr_var mat_var set_var list_var dict_var recursive_var
%use sos
%dict null_var num_var num_arr_var logic_var logic_arr_var char_var char_arr_var mat_var set_var list_var dict_var recursive_var
''')
res = get_result(iopub)
self.assertEqual(res['null_var'], None)
self.assertEqual(res['num_var'], 123)
self.assertEqual(res['num_arr_var'], [1,2,3])
self.assertEqual(res['logic_var'], True)
self.assertEqual(res['logic_arr_var'], [True, False, True])
self.assertEqual(res['char_var'], '1"23')
self.assertEqual(res['char_arr_var'], ['1', '2', '3'])
import pandas as pd
import numpy as np
arr = np.random.randn(1000)
arr[::10] = np.nan
df = pd.DataFrame({'column_{0}'.format(i): arr for i in range(10)})
''')
clear_channels(iopub)
execute(kc=kc, code="%use Julia")
wait_for_idle(kc)
execute(kc=kc, code="%get df")
wait_for_idle(kc)
execute(kc=kc, code="size(df)")
res = get_display_data(iopub)
self.assertEqual(res, '(1000, 10)')
execute(kc=kc, code="%use sos")
wait_for_idle(kc)
def testHelp(self):
'''Test help messages'''
with sos_kernel() as kc:
# create a data frame
execute(kc=kc, code='\n'.join('%{} -h'.format(magic) for magic in (
'cd', 'debug', 'dict', 'get', 'matplotlib', 'paste', 'preview',
'put', 'render', 'rerun', 'run', 'save', 'sandbox', 'set',
'sessioninfo', 'sosrun', 'sossave', 'shutdown', 'taskinfo', 'tasks',
'toc', 'use', 'with', 'pull', 'push')))
wait_for_idle(kc)
import pandas as pd
import numpy as np
arr = np.random.randn(1000)
arr[::10] = np.nan
df = pd.DataFrame({'column_{0}'.format(i): arr for i in range(10)})
''')
clear_channels(iopub)
execute(kc=kc, code="%use R")
wait_for_idle(kc)
execute(kc=kc, code="%get df")
wait_for_idle(kc)
execute(kc=kc, code="dim(df)")
res = get_display_data(iopub)
self.assertEqual(res, '[1] 1000 10')
execute(kc=kc, code="%use sos")
wait_for_idle(kc)
#