133 lines
4.4 KiB
Python
133 lines
4.4 KiB
Python
#!/usr/bin/env/python3
|
|
# Copyright (c) Facebook, Inc. and its affiliates.
|
|
# 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.
|
|
"""Unit Tests for Minio Component."""
|
|
import tempfile
|
|
import os
|
|
import mock
|
|
import pytest
|
|
from pytorch_kfp_components.components.minio.component import MinIO
|
|
from pytorch_kfp_components.components.minio.executor import Executor
|
|
|
|
tmpdir = tempfile.mkdtemp()
|
|
|
|
with open(os.path.join(str(tmpdir), "dummy.txt"), "w") as fp:
|
|
fp.write("dummy")
|
|
|
|
#pylint: disable=redefined-outer-name
|
|
#pylint: disable=invalid-name
|
|
|
|
@pytest.fixture(scope="class")
|
|
def minio_inputs():
|
|
"""Sets the minio inputs.
|
|
|
|
Returns:
|
|
minio_inputs : dict of inputs for minio uploads.
|
|
"""
|
|
minio_input = {
|
|
"bucket_name": "dummy",
|
|
"source": f"{tmpdir}/dummy.txt",
|
|
"destination": "dummy.txt",
|
|
"endpoint": "localhost:9000",
|
|
}
|
|
return minio_input
|
|
|
|
|
|
def upload_to_minio(minio_inputs):
|
|
"""Uoloads the artifact to minio."""
|
|
MinIO(
|
|
source=minio_inputs["source"],
|
|
bucket_name=minio_inputs["bucket_name"],
|
|
destination=minio_inputs["destination"],
|
|
endpoint=minio_inputs["endpoint"],
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"key",
|
|
["source", "bucket_name", "destination", "endpoint"],
|
|
)
|
|
def test_minio_variables_invalid_type(minio_inputs, key):
|
|
"""Testing for invalid variable types."""
|
|
minio_inputs[key] = ["test"]
|
|
expected_exception_msg = f"{key} must be of type <class 'str'>" \
|
|
f" but received as {type(minio_inputs[key])}"
|
|
with pytest.raises(TypeError, match=expected_exception_msg):
|
|
upload_to_minio(minio_inputs)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"key",
|
|
["source", "bucket_name", "destination", "endpoint"],
|
|
)
|
|
def test_minio_mandatory_param(minio_inputs, key):
|
|
"""Testing for invalid minio mandatory parameters."""
|
|
minio_inputs[key] = None
|
|
expected_exception_msg = (
|
|
f"{key} is not optional. Received value: {minio_inputs[key]}")
|
|
with pytest.raises(ValueError, match=expected_exception_msg):
|
|
upload_to_minio(minio_inputs)
|
|
|
|
|
|
def test_missing_access_key(minio_inputs):
|
|
"""Test upload if minio access key is missing."""
|
|
os.environ["MINIO_SECRET_KEY"] = "dummy"
|
|
expected_exception_msg = "Environment variable MINIO_ACCESS_KEY not found"
|
|
with pytest.raises(ValueError, match=expected_exception_msg):
|
|
upload_to_minio(minio_inputs)
|
|
|
|
os.environ.pop("MINIO_SECRET_KEY")
|
|
|
|
|
|
def test_missing_secret_key(minio_inputs):
|
|
"""Test upload if minio secret key is missing."""
|
|
os.environ["MINIO_ACCESS_KEY"] = "dummy"
|
|
expected_exception_msg = "Environment variable MINIO_SECRET_KEY not found"
|
|
with pytest.raises(ValueError, match=expected_exception_msg):
|
|
upload_to_minio(minio_inputs)
|
|
os.environ.pop("MINIO_ACCESS_KEY")
|
|
|
|
|
|
def test_unreachable_endpoint(minio_inputs):
|
|
"""Testing unreachable minio endpoint with invalid minio creds."""
|
|
os.environ["MINIO_ACCESS_KEY"] = "dummy"
|
|
os.environ["MINIO_SECRET_KEY"] = "dummy"
|
|
with pytest.raises(Exception, match="Max retries exceeded with url: "):
|
|
upload_to_minio(minio_inputs)
|
|
|
|
|
|
def test_invalid_file_path(minio_inputs):
|
|
"""Test invalid source file path."""
|
|
minio_inputs["source"] = "dummy"
|
|
expected_exception_msg = (
|
|
f"Input path - {minio_inputs['source']} does not exists")
|
|
with pytest.raises(ValueError, match=expected_exception_msg):
|
|
upload_to_minio(minio_inputs)
|
|
|
|
|
|
def test_minio_upload_file(minio_inputs):
|
|
"""Testing upload of files to minio."""
|
|
with mock.patch.object(Executor, "upload_artifacts_to_minio") as client:
|
|
client.return_value = []
|
|
upload_to_minio(minio_inputs)
|
|
client.asser_called_once()
|
|
|
|
|
|
def test_minio_upload_folder(minio_inputs):
|
|
"""Testing upload of folder to minio."""
|
|
minio_inputs["source"] = tmpdir
|
|
with mock.patch.object(Executor, "upload_artifacts_to_minio") as client:
|
|
client.return_value = []
|
|
upload_to_minio(minio_inputs)
|
|
client.asser_called_once()
|