Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test(self):
nb_test1_fname = get_notebook_path('simple_execute.ipynb')
nb_test1_executed_fname = os.path.join(self.test_dir, 'test1_executed.ipynb')
execute_notebook(nb_test1_fname, nb_test1_executed_fname, {'msg': 'Hello'})
test_nb = read_notebook(nb_test1_executed_fname)
self.assertEqual(test_nb.node.cells[0].get('source'), u'# Parameters\nmsg = "Hello"\n')
self.assertEqual(test_nb.parameters, {'msg': 'Hello'})
def test(self):
path = get_notebook_path('broken.ipynb')
result_path = os.path.join(self.test_dir, 'broken.ipynb')
execute_notebook(path, result_path)
nb = read_notebook(result_path)
self.assertEqual(nb.node.cells[0].execution_count, 1)
self.assertEqual(nb.node.cells[1].execution_count, 2)
self.assertEqual(nb.node.cells[1].outputs[0].output_type, 'error')
self.assertEqual(nb.node.cells[2].execution_count, None)
if inject_input_path or inject_paths:
parameters_final['PAPERMILL_INPUT_PATH'] = notebook_path
if inject_output_path or inject_paths:
parameters_final['PAPERMILL_OUTPUT_PATH'] = output_path
for params in parameters_base64 or []:
parameters_final.update(yaml.load(base64.b64decode(params), Loader=NoDatesSafeLoader))
for files in parameters_file or []:
parameters_final.update(read_yaml_file(files))
for params in parameters_yaml or []:
parameters_final.update(yaml.load(params, Loader=NoDatesSafeLoader))
for name, value in parameters or []:
parameters_final[name] = _resolve_type(value)
for name, value in parameters_raw or []:
parameters_final[name] = value
execute_notebook(
input_path=notebook_path,
output_path=output_path,
parameters=parameters_final,
engine_name=engine,
request_save_on_cell_execute=request_save_on_cell_execute,
prepare_only=prepare_only,
kernel_name=kernel,
progress_bar=progress_bar,
log_output=log_output,
stdout_file=stdout_file,
stderr_file=stderr_file,
start_timeout=start_timeout,
report_mode=report_mode,
cwd=cwd,
)
dir=os.getcwd(),
suffix='.ipynb',
delete=False).name
with open(input_file, 'w') as notebook_file:
nbformat.write(input_notebook, notebook_file, 4)
if output_notebook is None:
output_file = tempfile.NamedTemporaryFile(
prefix='__tmp_output_nb',
dir=os.getcwd(),
suffix='.ipynb',
delete=False).name
else:
output_file = output_notebook
execute_notebook(
input_path=input_file,
output_path=output_file,
engine_name='sos',
kernel_name='sos')
if os.path.basename(input_file).startswith('__tmp_input_nb'):
try:
os.remove(input_file)
except Exception as e:
env.logger.warning(
f'Failed to remove temporary input file {input_file}: {e}')
if os.path.basename(output_file).startswith(
'__tmp_output_nb') and return_content:
new_nb = nbformat.read(output_file, nbformat.NO_CONVERT)
try: