# # Copyright 2018-2022 Elyra Authors # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import hashlib import json import logging import os from pathlib import Path import subprocess from subprocess import CalledProcessError from subprocess import CompletedProcess from subprocess import run import sys from tempfile import TemporaryFile import time from typing import Optional import minio import mock import nbformat import papermill import pytest from elyra.kfp import bootstrapper # To run this test from an IDE: # 1. set PYTHONPATH='`path-to-repo`/etc/docker-scripts' and working directory to `path-to-repo` # 2. Manually launch test_minio container: docker run --name test_minio -d -p 9000:9000 minio/minio server /data # (this is located in Makefile) # # NOTE: Any changes to elyra/tests/kfp/resources/test-notebookA.ipynb require an # update of elyra/tests/kfp/resources/test-archive.tgz using the command below: # tar -cvzf test-archive.tgz test-notebookA.ipynb MINIO_HOST_PORT = os.getenv("MINIO_HOST_PORT", "127.0.0.1:9000") @pytest.fixture(scope="module", autouse=True) def start_minio(): """Start the minio container to simulate COS.""" # The docker run command will fail if an instance of the test_minio container is running. # We'll make a "silent" attempt to start. If that fails, assume its due to the container # conflict, force its shutdown, and try once more. If successful, yield the minio instance # but also shutdown on the flip-side of the yield (when the fixture is cleaned up). # # Although actions like SIGINT (ctrl-C) should still trigger the post-yield logic, urgent # interrupts like SIGQUIT (ctrl-\) or multiple SIGINTs can still orphan the container, so # we still need the pre-yield behavior. minio = start_minio_container(False) if minio is None: # Got a failure. Shutdown (assumed) container and try once more. stop_minio_container() minio = start_minio_container(True) time.sleep(3) # give container a chance to start yield minio stop_minio_container() def start_minio_container(raise_on_failure: bool = False) -> Optional[CompletedProcess]: minio = None try: minio = run( ["docker", "run", "--name", "test_minio", "-d", "-p", "9000:9000", "minio/minio", "server", "/data"], cwd=os.getcwd(), check=True, ) except CalledProcessError as ex: if raise_on_failure: raise RuntimeError(f"Error executing docker process: {ex}") from ex return minio def stop_minio_container(): run(["docker", "rm", "-f", "test_minio"], check=True) @pytest.fixture(scope="function") def s3_setup(): bucket_name = "test-bucket" cos_client = minio.Minio(MINIO_HOST_PORT, access_key="minioadmin", secret_key="minioadmin", secure=False) cos_client.make_bucket(bucket_name) yield cos_client cleanup_files = cos_client.list_objects(bucket_name, recursive=True) for file in cleanup_files: cos_client.remove_object(bucket_name, file.object_name) cos_client.remove_bucket(bucket_name) def main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict): """Primary body for main method testing...""" monkeypatch.setattr(bootstrapper.OpUtil, "parse_arguments", lambda x: argument_dict) monkeypatch.setattr(bootstrapper.OpUtil, "package_install", mock.Mock(return_value=True)) monkeypatch.setenv("AWS_ACCESS_KEY_ID", "minioadmin") monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "minioadmin") monkeypatch.setenv("TEST_ENV_VAR1", "test_env_var1") s3_setup.fput_object( bucket_name=argument_dict["cos-bucket"], object_name="test-directory/test-file.txt", file_path="elyra/tests/kfp/resources/test-requirements-elyra.txt", ) s3_setup.fput_object( bucket_name=argument_dict["cos-bucket"], object_name="test-directory/test,file.txt", file_path="elyra/tests/kfp/resources/test-bad-requirements-elyra.txt", ) s3_setup.fput_object( bucket_name=argument_dict["cos-bucket"], object_name="test-directory/test-archive.tgz", file_path="elyra/tests/kfp/resources/test-archive.tgz", ) with tmpdir.as_cwd(): bootstrapper.main() test_file_list = [ "test-archive.tgz", "test-file.txt", "test,file.txt", "test-file/test-file-copy.txt", "test-file/test,file/test,file-copy.txt", "test-notebookA.ipynb", "test-notebookA-output.ipynb", "test-notebookA.html", ] # Ensure working directory has all the files. for file in test_file_list: assert os.path.isfile(file) # Ensure upload directory has all the files EXCEPT the output notebook # since it was it is uploaded as the input notebook (test-notebookA.ipynb) # (which is included in the archive at start). for file in test_file_list: if file != "test-notebookA-output.ipynb": assert s3_setup.stat_object( bucket_name=argument_dict["cos-bucket"], object_name="test-directory/" + file ) if file == "test-notebookA.html": with open("test-notebookA.html") as html_file: assert "TEST_ENV_VAR1: test_env_var1" in html_file.read() def _get_operation_instance(monkeypatch, s3_setup): config = { "cos-endpoint": "http://" + MINIO_HOST_PORT, "cos-user": "minioadmin", "cos-password": "minioadmin", "cos-bucket": "test-bucket", "filepath": "untitled.ipynb", } op = bootstrapper.FileOpBase.get_instance(**config) # use the same minio instance used by the test # to avoid access denied errors when two minio # instances exist monkeypatch.setattr(op, "cos_client", s3_setup) return op def test_main_method(monkeypatch, s3_setup, tmpdir): argument_dict = { "cos-endpoint": "http://" + MINIO_HOST_PORT, "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-archive.tgz", "filepath": "elyra/tests/kfp/resources/test-notebookA.ipynb", "inputs": "test-file.txt;test,file.txt", "outputs": "test-file/test-file-copy.txt;test-file/test,file/test,file-copy.txt", "user-volume-path": None, } main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict) def test_main_method_with_wildcard_outputs(monkeypatch, s3_setup, tmpdir): argument_dict = { "cos-endpoint": "http://" + MINIO_HOST_PORT, "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-archive.tgz", "filepath": "elyra/tests/kfp/resources/test-notebookA.ipynb", "inputs": "test-file.txt;test,file.txt", "outputs": "test-file/*", "user-volume-path": None, } main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict) def test_main_method_with_dir_outputs(monkeypatch, s3_setup, tmpdir): argument_dict = { "cos-endpoint": "http://" + MINIO_HOST_PORT, "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-archive.tgz", "filepath": "elyra/tests/kfp/resources/test-notebookA.ipynb", "inputs": "test-file.txt;test,file.txt", "outputs": "test-file", # this is the directory that contains the outputs "user-volume-path": None, } main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict) def is_writable_dir(path): """Helper method determines whether 'path' is a writable directory""" try: with TemporaryFile(mode="w", dir=path) as t: t.write("1") return True except Exception: return False def remove_file(filename, fail_ok=True): """Removes filename. If fail_ok is False an assert is raised if removal failed for any reason, e.g. filenotfound """ try: os.remove(filename) except OSError as ose: if fail_ok is False: raise AssertionError(f"Cannot remove {filename}: {str(ose)} {ose}") def test_process_metrics_method_not_writable_dir(monkeypatch, s3_setup, tmpdir): """Test for process_metrics_and_metadata Validates that the method can handle output directory that is not writable """ # remove "default" output file if it already exists output_metadata_file = Path("/tmp") / "mlpipeline-ui-metadata.json" remove_file(output_metadata_file) try: monkeypatch.setenv("ELYRA_WRITABLE_CONTAINER_DIR", "/good/time/to/fail") argument_dict = { "cos-endpoint": f"http://{MINIO_HOST_PORT}", "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-archive.tgz", "filepath": "elyra/tests/kfp/resources/test-notebookA.ipynb", "inputs": "test-file.txt;test,file.txt", "outputs": "test-file/test-file-copy.txt;test-file/test,file/test,file-copy.txt", "user-volume-path": None, } main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict) except Exception as ex: print(f"Writable dir test failed: {str(ex)} {ex}") assert False assert output_metadata_file.exists() is False def test_process_metrics_method_no_metadata_file(monkeypatch, s3_setup, tmpdir): """Test for process_metrics_and_metadata Verifies that the method produces a valid KFP UI metadata file if the node's script | notebook did not generate this metadata file. """ argument_dict = { "cos-endpoint": "http://" + MINIO_HOST_PORT, "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-archive.tgz", "filepath": "elyra/tests/kfp/resources/test-notebookA.ipynb", "inputs": "test-file.txt;test,file.txt", "outputs": "test-file/test-file-copy.txt;test-file/test,file/test,file-copy.txt", "user-volume-path": None, } output_path = Path(tmpdir) # metadata file name and location metadata_file = output_path / "mlpipeline-ui-metadata.json" # remove file if it already exists remove_file(metadata_file) # override the default output directory to make this test platform # independent monkeypatch.setenv("ELYRA_WRITABLE_CONTAINER_DIR", str(tmpdir)) main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict) # process_metrics should have generated a file named mlpipeline-ui-metadata.json # in tmpdir try: with open(metadata_file, "r") as f: metadata = json.load(f) assert metadata.get("outputs") is not None assert isinstance(metadata["outputs"], list) assert len(metadata["outputs"]) == 1 assert metadata["outputs"][0]["storage"] == "inline" assert metadata["outputs"][0]["type"] == "markdown" assert ( f"{argument_dict['cos-endpoint']}/{argument_dict['cos-bucket']}/{argument_dict['cos-directory']}" in metadata["outputs"][0]["source"] ) assert argument_dict["cos-dependencies-archive"] in metadata["outputs"][0]["source"] except AssertionError: raise except Exception as ex: # Potential reasons for failures: # file not found, invalid JSON print(f'Validation of "{str(ex)}" failed: {ex}') assert False def test_process_metrics_method_valid_metadata_file(monkeypatch, s3_setup, tmpdir): """Test for process_metrics_and_metadata Verifies that the method produces a valid KFP UI metadata file if the node's script | notebook generated this metadata file. """ argument_dict = { "cos-endpoint": "http://" + MINIO_HOST_PORT, "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-archive.tgz", "filepath": "elyra/tests/kfp/resources/test-notebookA.ipynb", "inputs": "test-file.txt;test,file.txt", "outputs": "test-file/test-file-copy.txt;test-file/test,file/test,file-copy.txt", "user-volume-path": None, } output_path = Path(tmpdir) # metadata file name and location input_metadata_file = "mlpipeline-ui-metadata.json" output_metadata_file = output_path / input_metadata_file # remove output_metadata_file if it already exists remove_file(output_metadata_file) # # Simulate some custom metadata that the script | notebook produced # custom_metadata = { "some_property": "some property value", "outputs": [{"source": "gs://project/bucket/file.md", "type": "markdown"}], } with tmpdir.as_cwd(): with open(input_metadata_file, "w") as f: json.dump(custom_metadata, f) # override the default output directory to make this test platform # independent monkeypatch.setenv("ELYRA_WRITABLE_CONTAINER_DIR", str(tmpdir)) main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict) # output_metadata_file should now exist try: with open(output_metadata_file, "r") as f: metadata = json.load(f) assert metadata.get("some_property") is not None assert metadata["some_property"] == custom_metadata["some_property"] assert metadata.get("outputs") is not None assert isinstance(metadata["outputs"], list) assert len(metadata["outputs"]) == 2 for output in metadata["outputs"]: if output.get("storage") is not None: assert output["storage"] == "inline" assert output["type"] == "markdown" assert ( f"{argument_dict['cos-endpoint']}/{argument_dict['cos-bucket']}/{argument_dict['cos-directory']}" # noqa in output["source"] ) assert argument_dict["cos-dependencies-archive"] in output["source"] else: assert output["type"] == custom_metadata["outputs"][0]["type"] assert output["source"] == custom_metadata["outputs"][0]["source"] except AssertionError: raise except Exception as ex: # Potential reasons for failures: # file not found, invalid JSON print(f'Validation of "{str(ex)}" failed: {ex}') assert False def test_process_metrics_method_invalid_metadata_file(monkeypatch, s3_setup, tmpdir): """Test for process_metrics_and_metadata Verifies that the method produces a valid KFP UI metadata file if the node's script | notebook generated an invalid metadata file. """ argument_dict = { "cos-endpoint": f"http://{MINIO_HOST_PORT}", "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-archive.tgz", "filepath": "elyra/tests/kfp/resources/test-notebookA.ipynb", "inputs": "test-file.txt;test,file.txt", "outputs": "test-file/test-file-copy.txt;test-file/test,file/test,file-copy.txt", "user-volume-path": None, } output_path = Path(tmpdir) # metadata file name and location input_metadata_file = "mlpipeline-ui-metadata.json" output_metadata_file = output_path / input_metadata_file # remove output_metadata_file if it already exists remove_file(output_metadata_file) # # Populate the metadata file with some custom data that's not JSON # with tmpdir.as_cwd(): with open(input_metadata_file, "w") as f: f.write("I am not a valid JSON data structure") f.write("1,2,3,4,5,6,7") # override the default output directory to make this test platform # independent monkeypatch.setenv("ELYRA_WRITABLE_CONTAINER_DIR", str(tmpdir)) main_method_setup_execution(monkeypatch, s3_setup, tmpdir, argument_dict) # process_metrics replaces the existing metadata file # because its content cannot be merged try: with open(output_metadata_file, "r") as f: metadata = json.load(f) assert metadata.get("outputs") is not None assert isinstance(metadata["outputs"], list) assert len(metadata["outputs"]) == 1 assert metadata["outputs"][0]["storage"] == "inline" assert metadata["outputs"][0]["type"] == "markdown" assert ( f"{argument_dict['cos-endpoint']}/{argument_dict['cos-bucket']}/{argument_dict['cos-directory']}" in metadata["outputs"][0]["source"] ) assert argument_dict["cos-dependencies-archive"] in metadata["outputs"][0]["source"] except AssertionError: raise except Exception as ex: # Potential reasons for failures: # file not found, invalid JSON print(f'Validation of "{str(ex)}" failed: {ex}') assert False def test_fail_bad_notebook_main_method(monkeypatch, s3_setup, tmpdir): argument_dict = { "cos-endpoint": f"http://{MINIO_HOST_PORT}", "cos-bucket": "test-bucket", "cos-directory": "test-directory", "cos-dependencies-archive": "test-bad-archiveB.tgz", "filepath": "elyra/tests/kfp/resources/test-bad-notebookB.ipynb", "inputs": "test-file.txt", "outputs": "test-file/test-copy-file.txt", "user-volume-path": None, } monkeypatch.setattr(bootstrapper.OpUtil, "parse_arguments", lambda x: argument_dict) monkeypatch.setattr(bootstrapper.OpUtil, "package_install", mock.Mock(return_value=True)) mocked_func = mock.Mock( return_value="default", side_effect=[ "test-bad-archiveB.tgz", "test-file.txt", "test-bad-notebookB-output.ipynb", "test-bad-notebookB.html", "test-file.txt", ], ) monkeypatch.setattr(bootstrapper.FileOpBase, "get_object_storage_filename", mocked_func) monkeypatch.setenv("AWS_ACCESS_KEY_ID", "minioadmin") monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "minioadmin") s3_setup.fput_object(bucket_name=argument_dict["cos-bucket"], object_name="test-file.txt", file_path="README.md") s3_setup.fput_object( bucket_name=argument_dict["cos-bucket"], object_name="test-bad-archiveB.tgz", file_path="elyra/tests/kfp/resources/test-bad-archiveB.tgz", ) with tmpdir.as_cwd(): with pytest.raises(papermill.exceptions.PapermillExecutionError): bootstrapper.main() def test_package_installation(monkeypatch, virtualenv): elyra_dict = { "ipykernel": "5.3.0", "ansiwrap": "0.8.4", "packaging": "20.0", "text-extensions-for-pandas": "0.0.1-prealpha", } to_install_dict = { "bleach": "3.1.5", "ansiwrap": "0.7.0", "packaging": "20.4", "text-extensions-for-pandas": "0.0.1-prealpha", } correct_dict = { "ipykernel": "5.3.0", "ansiwrap": "0.8.4", "packaging": "20.4", "text-extensions-for-pandas": "0.0.1-prealpha", } mocked_func = mock.Mock(return_value="default", side_effect=[elyra_dict, to_install_dict]) monkeypatch.setattr(bootstrapper.OpUtil, "package_list_to_dict", mocked_func) monkeypatch.setattr(sys, "executable", virtualenv.python) virtualenv.run("python3 -m pip install bleach==3.1.5") virtualenv.run("python3 -m pip install ansiwrap==0.7.0") virtualenv.run("python3 -m pip install packaging==20.4") virtualenv.run( "python3 -m pip install git+https://github.com/akchinSTC/" "text-extensions-for-pandas@3de5ce17ab0493dcdf88b51e8727f580c08d6997" ) bootstrapper.OpUtil.package_install(user_volume_path=None) virtual_env_dict = {} output = virtualenv.run("python3 -m pip freeze", capture=True) print("This is the [pip freeze] output :\n" + output) for line in output.strip().split("\n"): if " @ " in line: package_name, package_version = line.strip("\n").split(sep=" @ ") elif "===" in line: package_name, package_version = line.strip("\n").split(sep="===") else: package_name, package_version = line.strip("\n").split(sep="==") virtual_env_dict[package_name] = package_version for package, version in correct_dict.items(): assert virtual_env_dict[package] == version def test_package_installation_with_target_path(monkeypatch, virtualenv, tmpdir): # TODO : Need to add test for direct-source e.g. ' @ ' elyra_dict = { "ipykernel": "5.3.0", "ansiwrap": "0.8.4", "packaging": "20.0", "text-extensions-for-pandas": "0.0.1-prealpha", } to_install_dict = { "bleach": "3.1.5", "ansiwrap": "0.7.0", "packaging": "21.0", "text-extensions-for-pandas": "0.0.1-prealpha", } correct_dict = { "ipykernel": "5.3.0", "ansiwrap": "0.8.4", "packaging": "21.0", "text-extensions-for-pandas": "0.0.1-prealpha", } mocked_func = mock.Mock(return_value="default", side_effect=[elyra_dict, to_install_dict]) monkeypatch.setattr(bootstrapper.OpUtil, "package_list_to_dict", mocked_func) monkeypatch.setattr(sys, "executable", virtualenv.python) virtualenv.run("python3 -m pip install --upgrade pip") virtualenv.run(f"python3 -m pip install --target={tmpdir} bleach==3.1.5") virtualenv.run(f"python3 -m pip install --target={tmpdir} ansiwrap==0.7.0") virtualenv.run(f"python3 -m pip install --target={tmpdir} packaging==20.9") virtualenv.run( f"python3 -m pip install --target={tmpdir} git+https://github.com/akchinSTC/" "text-extensions-for-pandas@3de5ce17ab0493dcdf88b51e8727f580c08d6997" ) bootstrapper.OpUtil.package_install(user_volume_path=str(tmpdir)) virtual_env_dict = {} output = virtualenv.run(f"python3 -m pip freeze --path={tmpdir}", capture=True) print("This is the [pip freeze] output :\n" + output) for line in output.strip().split("\n"): if " @ " in line: package_name, package_version = line.strip("\n").split(sep=" @ ") elif "===" in line: package_name, package_version = line.strip("\n").split(sep="===") else: package_name, package_version = line.strip("\n").split(sep="==") virtual_env_dict[package_name] = package_version for package, version in correct_dict.items(): assert virtual_env_dict[package].split(".")[0] == version.split(".")[0] def test_convert_notebook_to_html(tmpdir): notebook_file = os.getcwd() + "/elyra/tests/kfp/resources/test-notebookA.ipynb" notebook_output_html_file = "test-notebookA.html" with tmpdir.as_cwd(): bootstrapper.NotebookFileOp.convert_notebook_to_html(notebook_file, notebook_output_html_file) assert os.path.isfile(notebook_output_html_file) # Validate that an html file got generated from the notebook with open(notebook_output_html_file, "r") as html_file: html_data = html_file.read() assert html_data.startswith("") assert "TEST_ENV_VAR1" in html_data # from os.getenv("TEST_ENV_VAR1") assert html_data.endswith("\n") def test_fail_convert_notebook_to_html(tmpdir): notebook_file = os.getcwd() + "/elyra/tests/kfp/resources/test-bad-notebookA.ipynb" notebook_output_html_file = "bad-notebookA.html" with tmpdir.as_cwd(): # Recent versions raising typeError due to #1130 # https://github.com/jupyter/nbconvert/pull/1130 with pytest.raises((TypeError, nbformat.validator.NotebookValidationError)): bootstrapper.NotebookFileOp.convert_notebook_to_html(notebook_file, notebook_output_html_file) def test_get_file_object_store(monkeypatch, s3_setup, tmpdir): file_to_get = "README.md" current_directory = os.getcwd() + "/" bucket_name = "test-bucket" s3_setup.fput_object(bucket_name=bucket_name, object_name=file_to_get, file_path=file_to_get) with tmpdir.as_cwd(): op = _get_operation_instance(monkeypatch, s3_setup) op.get_file_from_object_storage(file_to_get) assert os.path.isfile(file_to_get) assert _fileChecksum(file_to_get) == _fileChecksum(current_directory + file_to_get) def test_fail_get_file_object_store(monkeypatch, s3_setup, tmpdir): file_to_get = "test-file.txt" with tmpdir.as_cwd(): with pytest.raises(minio.error.S3Error) as exc_info: op = _get_operation_instance(monkeypatch, s3_setup) op.get_file_from_object_storage(file_to_get=file_to_get) assert exc_info.value.code == "NoSuchKey" def test_put_file_object_store(monkeypatch, s3_setup, tmpdir): bucket_name = "test-bucket" file_to_put = "LICENSE" current_directory = os.getcwd() + "/" op = _get_operation_instance(monkeypatch, s3_setup) op.put_file_to_object_storage(file_to_upload=file_to_put) with tmpdir.as_cwd(): s3_setup.fget_object(bucket_name, file_to_put, file_to_put) assert os.path.isfile(file_to_put) assert _fileChecksum(file_to_put) == _fileChecksum(current_directory + file_to_put) def test_fail_invalid_filename_put_file_object_store(monkeypatch, s3_setup): file_to_put = "LICENSE_NOT_HERE" with pytest.raises(FileNotFoundError): op = _get_operation_instance(monkeypatch, s3_setup) op.put_file_to_object_storage(file_to_upload=file_to_put) def test_fail_bucket_put_file_object_store(monkeypatch, s3_setup): bucket_name = "test-bucket-not-exist" file_to_put = "LICENSE" with pytest.raises(minio.error.S3Error) as exc_info: op = _get_operation_instance(monkeypatch, s3_setup) monkeypatch.setattr(op, "cos_bucket", bucket_name) op.put_file_to_object_storage(file_to_upload=file_to_put) assert exc_info.value.code == "NoSuchBucket" def test_find_best_kernel_nb(tmpdir): source_nb_file = os.path.join(os.getcwd(), "elyra/tests/kfp/resources/test-notebookA.ipynb") nb_file = os.path.join(tmpdir, "test-notebookA.ipynb") # "Copy" nb file to destination - this test does not update the kernel or language. nb = nbformat.read(source_nb_file, 4) nbformat.write(nb, nb_file) with tmpdir.as_cwd(): kernel_name = bootstrapper.NotebookFileOp.find_best_kernel(nb_file) assert kernel_name == nb.metadata.kernelspec["name"] def test_find_best_kernel_lang(tmpdir, caplog): caplog.set_level(logging.INFO) source_nb_file = os.path.join(os.getcwd(), "elyra/tests/kfp/resources/test-notebookA.ipynb") nb_file = os.path.join(tmpdir, "test-notebookA.ipynb") # "Copy" nb file to destination after updating the kernel name - forcing a language match nb = nbformat.read(source_nb_file, 4) nb.metadata.kernelspec["name"] = "test-kernel" nb.metadata.kernelspec["language"] = "PYTHON" # test case-insensitivity nbformat.write(nb, nb_file) with tmpdir.as_cwd(): kernel_name = bootstrapper.NotebookFileOp.find_best_kernel(nb_file) assert kernel_name == "python3" assert len(caplog.records) == 1 assert caplog.records[0].message.startswith("Matched kernel by language (PYTHON)") def test_find_best_kernel_nomatch(tmpdir, caplog): source_nb_file = os.path.join(os.getcwd(), "elyra/tests/kfp/resources/test-notebookA.ipynb") nb_file = os.path.join(tmpdir, "test-notebookA.ipynb") # "Copy" nb file to destination after updating the kernel name and language - forcing use of updated name nb = nbformat.read(source_nb_file, 4) nb.metadata.kernelspec["name"] = "test-kernel" nb.metadata.kernelspec["language"] = "test-language" nbformat.write(nb, nb_file) with tmpdir.as_cwd(): kernel_name = bootstrapper.NotebookFileOp.find_best_kernel(nb_file) assert kernel_name == "test-kernel" assert len(caplog.records) == 1 assert caplog.records[0].message.startswith("Reverting back to missing notebook kernel 'test-kernel'") def test_parse_arguments(): test_args = [ "-e", "http://test.me.now", "-d", "test-directory", "-t", "test-archive.tgz", "-f", "test-notebook.ipynb", "-b", "test-bucket", "-p", "/tmp/lib", "-n", "test-pipeline", ] args_dict = bootstrapper.OpUtil.parse_arguments(test_args) assert args_dict["cos-endpoint"] == "http://test.me.now" assert args_dict["cos-directory"] == "test-directory" assert args_dict["cos-dependencies-archive"] == "test-archive.tgz" assert args_dict["cos-bucket"] == "test-bucket" assert args_dict["filepath"] == "test-notebook.ipynb" assert args_dict["user-volume-path"] == "/tmp/lib" assert args_dict["pipeline-name"] == "test-pipeline" assert not args_dict["inputs"] assert not args_dict["outputs"] def test_fail_missing_notebook_parse_arguments(): test_args = ["-e", "http://test.me.now", "-d", "test-directory", "-t", "test-archive.tgz", "-b", "test-bucket"] with pytest.raises(SystemExit): bootstrapper.OpUtil.parse_arguments(test_args) def test_fail_missing_endpoint_parse_arguments(): test_args = ["-d", "test-directory", "-t", "test-archive.tgz", "-f", "test-notebook.ipynb", "-b", "test-bucket"] with pytest.raises(SystemExit): bootstrapper.OpUtil.parse_arguments(test_args) def test_fail_missing_archive_parse_arguments(): test_args = ["-e", "http://test.me.now", "-d", "test-directory", "-f", "test-notebook.ipynb", "-b", "test-bucket"] with pytest.raises(SystemExit): bootstrapper.OpUtil.parse_arguments(test_args) def test_fail_missing_bucket_parse_arguments(): test_args = [ "-e", "http://test.me.now", "-d", "test-directory", "-t", "test-archive.tgz", "-f", "test-notebook.ipynb", ] with pytest.raises(SystemExit): bootstrapper.OpUtil.parse_arguments(test_args) def test_fail_missing_directory_parse_arguments(): test_args = ["-e", "http://test.me.now", "-t", "test-archive.tgz", "-f", "test-notebook.ipynb", "-b", "test-bucket"] with pytest.raises(SystemExit): bootstrapper.OpUtil.parse_arguments(test_args) def test_requirements_file(monkeypatch, tmpdir, caplog): elyra_requirements_file = Path(__file__).parent / "resources/test-requirements-elyra.txt" elyra_correct_number_of_packages = 19 elyra_list_dict = bootstrapper.OpUtil.package_list_to_dict(elyra_requirements_file) assert len(elyra_list_dict) == elyra_correct_number_of_packages current_requirements_file = Path(__file__).parent / "resources/test-requirements-current.txt" current_correct_number_of_packages = 15 current_list_dict = bootstrapper.OpUtil.package_list_to_dict(current_requirements_file) assert len(current_list_dict) == current_correct_number_of_packages mocked_package_list_to_dict = mock.Mock(return_value="default", side_effect=[elyra_list_dict, current_list_dict]) monkeypatch.setattr(bootstrapper.OpUtil, "package_list_to_dict", mocked_package_list_to_dict) mocked_subprocess_run = mock.Mock(return_value="default") monkeypatch.setattr(subprocess, "run", mocked_subprocess_run) bootstrapper.OpUtil.package_install(user_volume_path=str(tmpdir)) assert "WARNING: Source package 'jupyter-client' found already installed as an editable package" in caplog.text assert "WARNING: Source package 'requests' found already installed as an editable package" in caplog.text assert "WARNING: Source package 'tornado' found already installed from git" in caplog.text def test_fail_requirements_file_bad_delimiter(): bad_requirements_file = Path(__file__).parent / "resources/test-bad-requirements-elyra.txt" with open(bad_requirements_file, "r") as f: file_content = f.readlines() valid_package_list = [ line.strip("\n").split("==")[0] for line in file_content if not line.startswith("#") and "==" in line ] package_dict = bootstrapper.OpUtil.package_list_to_dict(bad_requirements_file) assert valid_package_list == list(package_dict.keys()) def _fileChecksum(filename): hasher = hashlib.sha256() with open(filename, "rb") as afile: buf = afile.read(65536) while len(buf) > 0: hasher.update(buf) buf = afile.read(65536) checksum = hasher.hexdigest() return checksum