aboutsummaryrefslogtreecommitdiffstats
path: root/ms/artifact-manager/tests
diff options
context:
space:
mode:
Diffstat (limited to 'ms/artifact-manager/tests')
-rw-r--r--ms/artifact-manager/tests/base_test.py19
-rw-r--r--ms/artifact-manager/tests/configuration-test.ini4
-rw-r--r--ms/artifact-manager/tests/configuration_test.py47
-rw-r--r--ms/artifact-manager/tests/servicer_test.py196
-rw-r--r--ms/artifact-manager/tests/utils_test.py59
5 files changed, 325 insertions, 0 deletions
diff --git a/ms/artifact-manager/tests/base_test.py b/ms/artifact-manager/tests/base_test.py
new file mode 100644
index 000000000..4466b33dc
--- /dev/null
+++ b/ms/artifact-manager/tests/base_test.py
@@ -0,0 +1,19 @@
+"""Copyright 2019 Deutsche Telekom.
+
+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.
+"""
+
+
+def test_mock_function():
+ """Basic mock test that allways work."""
+ assert True
diff --git a/ms/artifact-manager/tests/configuration-test.ini b/ms/artifact-manager/tests/configuration-test.ini
new file mode 100644
index 000000000..e28e6f1c4
--- /dev/null
+++ b/ms/artifact-manager/tests/configuration-test.ini
@@ -0,0 +1,4 @@
+[testSection]
+testValue: 123
+[artifactManagerServer]
+artifactManagerValue=123
diff --git a/ms/artifact-manager/tests/configuration_test.py b/ms/artifact-manager/tests/configuration_test.py
new file mode 100644
index 000000000..219908325
--- /dev/null
+++ b/ms/artifact-manager/tests/configuration_test.py
@@ -0,0 +1,47 @@
+"""Copyright 2019 Deutsche Telekom.
+
+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.
+"""
+from configparser import NoOptionError
+from pathlib import Path, PurePath
+from typing import NoReturn
+
+from pytest import raises
+
+from manager.configuration import ArtifactManagerConfiguration
+
+
+TEST_CONFIGURATION_FILE_PATH = str(
+ PurePath(Path(__file__).parent.absolute(), "configuration-test.ini")
+)
+
+
+def test_server_configuration_configuration_file_path() -> NoReturn:
+ """Test ArtifactManagerConfiguration class.
+
+ Test checks if configuration file is loaded properly and returns valid values.
+ If invalid section or option is provided it should raises KeyError or configparser.NoOptionError exceptions.
+ :return: NoReturn
+ """
+ configuration: ArtifactManagerConfiguration = ArtifactManagerConfiguration(
+ TEST_CONFIGURATION_FILE_PATH
+ )
+ assert configuration.get_section("testSection")
+ with raises(KeyError):
+ configuration.get_section("invalidSection")
+ assert configuration.get_property("testSection", "testValue") == "123"
+ with raises(NoOptionError):
+ configuration.get_property("testSection", "invalidValue")
+ assert configuration.artifact_manager_property("artifactManagerValue") == "123"
+ with raises(NoOptionError):
+ configuration.artifact_manager_property("invalidValue")
diff --git a/ms/artifact-manager/tests/servicer_test.py b/ms/artifact-manager/tests/servicer_test.py
new file mode 100644
index 000000000..131e6fb2c
--- /dev/null
+++ b/ms/artifact-manager/tests/servicer_test.py
@@ -0,0 +1,196 @@
+"""Copyright 2019 Deutsche Telekom.
+
+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 os
+import shutil
+import zipfile
+from unittest.mock import patch
+
+import manager.utils
+from manager.servicer import ArtifactManagerServicer
+from proto.BluePrintCommon_pb2 import ActionIdentifiers, CommonHeader
+from proto.BluePrintManagement_pb2 import (
+ BluePrintDownloadInput,
+ BluePrintManagementOutput,
+ BluePrintRemoveInput,
+ BluePrintUploadInput,
+ FileChunk,
+)
+from proto.BluePrintManagement_pb2_grpc import (
+ BluePrintManagementServiceStub,
+ add_BluePrintManagementServiceServicer_to_server,
+)
+from pytest import fixture
+
+ZIP_FILE_BINARY = b"PK\x05\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+
+
+class MockZipFile(zipfile.ZipFile):
+ def __init__(self, *args, **kwargs):
+ pass
+
+ def extractall(self, path: str) -> None:
+ pass
+
+ def write(self, *arg, **kwargs) -> None:
+ pass
+
+
+@fixture(scope="module")
+def grpc_add_to_server():
+ """pytest-grpcio required function."""
+ return add_BluePrintManagementServiceServicer_to_server
+
+
+@fixture(scope="module")
+def grpc_servicer():
+ """pytest-grpcio required function."""
+ return ArtifactManagerServicer()
+
+
+@fixture(scope="module") # noqa
+def grpc_stub_cls(grpc_channel):
+ """pytest-grpcio required function."""
+ return BluePrintManagementServiceStub
+
+
+def test_servicer_upload_handler_header_failure(grpc_stub):
+ """Test servicer upload handler."""
+ request: BluePrintUploadInput = BluePrintUploadInput()
+ output: BluePrintManagementOutput = grpc_stub.uploadBlueprint(request)
+ assert output.status.code == 500
+ assert output.status.message == "failure"
+ assert output.status.errorMessage == "Request has to have set both BluePrint name and version"
+
+
+def test_servicer_download_handler_header_failure(grpc_stub):
+ """Test servicer download handler."""
+ request: BluePrintDownloadInput = BluePrintDownloadInput()
+ output: BluePrintManagementOutput = grpc_stub.downloadBlueprint(request)
+ assert output.status.code == 500
+ assert output.status.message == "failure"
+ assert output.status.errorMessage == "Request has to have set both BluePrint name and version"
+
+
+def test_servicer_remove_handler_header_failure(grpc_stub):
+ """Test servicer remove handler."""
+ request: BluePrintRemoveInput = BluePrintRemoveInput()
+ output: BluePrintManagementOutput = grpc_stub.removeBlueprint(request)
+ assert output.status.code == 500
+ assert output.status.message == "failure"
+ assert output.status.errorMessage == "Request has to have set both BluePrint name and version"
+
+
+def test_servicer_upload_handler_failure(grpc_stub):
+ """Test servicer upload handler."""
+ action_identifiers: ActionIdentifiers = ActionIdentifiers()
+ action_identifiers.blueprintName = "sample-cba"
+ action_identifiers.blueprintVersion = "1.0.0"
+ request: BluePrintUploadInput = BluePrintUploadInput(actionIdentifiers=action_identifiers)
+ output: BluePrintManagementOutput = grpc_stub.uploadBlueprint(request)
+ assert output.status.code == 500
+ assert output.status.message == "failure"
+ assert output.status.errorMessage == "Invalid request"
+
+
+def test_servicer_download_handler_failure(grpc_stub):
+ """Test servicer download handler."""
+ action_identifiers: ActionIdentifiers = ActionIdentifiers()
+ action_identifiers.blueprintName = "sample-cba"
+ action_identifiers.blueprintVersion = "2.0.0"
+ request: BluePrintDownloadInput = BluePrintDownloadInput(actionIdentifiers=action_identifiers)
+ output: BluePrintManagementOutput = grpc_stub.downloadBlueprint(request)
+ assert output.status.code == 500
+ assert output.status.message == "failure"
+ assert output.status.errorMessage == "Artifact not found"
+
+
+def test_servicer_remove_handler_failure(grpc_stub):
+ """Test servicer remove handler."""
+ action_identifiers: ActionIdentifiers = ActionIdentifiers()
+ action_identifiers.blueprintName = "sample-cba"
+ action_identifiers.blueprintVersion = "1.0.0"
+ request: BluePrintRemoveInput = BluePrintRemoveInput(actionIdentifiers=action_identifiers)
+ output: BluePrintManagementOutput = grpc_stub.removeBlueprint(request)
+ assert output.status.code == 500
+ assert output.status.message == "failure"
+ assert output.status.errorMessage == "Artifact not found"
+
+
+def test_servicer_upload_handler_success(grpc_stub):
+ """Test servicer upload handler."""
+ header: CommonHeader = CommonHeader()
+ header.requestId = "1234"
+ header.subRequestId = "1234-1"
+ header.originatorId = "CDS"
+
+ action_identifiers: ActionIdentifiers = ActionIdentifiers()
+ action_identifiers.blueprintName = "sample-cba"
+ action_identifiers.blueprintVersion = "1.0.0"
+ action_identifiers.actionName = "SampleScript"
+
+ file_chunk = FileChunk()
+ file_chunk.chunk = ZIP_FILE_BINARY
+
+ # fmt: off
+ with patch.object(os, "makedirs", return_value=None), \
+ patch.object(manager.utils, 'ZipFile', return_value=MockZipFile()):
+ request: BluePrintUploadInput = BluePrintUploadInput(
+ commonHeader=header, fileChunk=file_chunk, actionIdentifiers=action_identifiers
+ )
+ output: BluePrintManagementOutput = grpc_stub.uploadBlueprint(request)
+ # fmt: on
+ assert output.status.code == 200
+ assert output.status.message == "success"
+
+
+def test_servicer_download_handler_success(grpc_stub):
+ """Test servicer download handler."""
+ header: CommonHeader = CommonHeader()
+ header.requestId = "1234"
+ header.subRequestId = "1234-1"
+ header.originatorId = "CDS"
+
+ action_identifiers: ActionIdentifiers = ActionIdentifiers()
+ action_identifiers.blueprintName = "sample-cba"
+ action_identifiers.blueprintVersion = "1.0.0"
+ action_identifiers.actionName = "SampleScript"
+
+ with patch.object(os.path, "exists", return_value=True):
+ request: BluePrintDownloadInput = BluePrintDownloadInput(
+ commonHeader=header, actionIdentifiers=action_identifiers
+ )
+ output: BluePrintManagementOutput = grpc_stub.downloadBlueprint(request)
+ assert output.status.code == 200
+ assert output.status.message == "success"
+ assert output.fileChunk.chunk == ZIP_FILE_BINARY
+
+
+def test_servicer_remove_handler_success(grpc_stub):
+ """Test servicer remove handler."""
+ header: CommonHeader = CommonHeader()
+ header.requestId = "1234"
+ header.subRequestId = "1234-1"
+ header.originatorId = "CDS"
+
+ action_identifiers: ActionIdentifiers = ActionIdentifiers()
+ action_identifiers.blueprintName = "sample-cba"
+ action_identifiers.blueprintVersion = "1.0.0"
+ action_identifiers.actionName = "SampleScript"
+
+ with patch.object(shutil, "rmtree", return_value=None) as mock_rmtree:
+ request: BluePrintRemoveInput = BluePrintRemoveInput(commonHeader=header, actionIdentifiers=action_identifiers)
+ output: BluePrintManagementOutput = grpc_stub.removeBlueprint(request)
+ assert output.status.code == 200
+ assert output.status.message == "success"
diff --git a/ms/artifact-manager/tests/utils_test.py b/ms/artifact-manager/tests/utils_test.py
new file mode 100644
index 000000000..75d8b4c19
--- /dev/null
+++ b/ms/artifact-manager/tests/utils_test.py
@@ -0,0 +1,59 @@
+import os
+import shutil
+import zipfile
+from unittest.mock import patch
+
+import manager.utils
+from manager.utils import FileRepository, Repository, RepositoryStrategy
+
+
+class MockZipFile(zipfile.ZipFile):
+ def __init__(self, *args, **kwargs):
+ pass
+
+ def extractall(self, path: str) -> None:
+ pass
+
+ def write(self, *arg, **kwargs) -> None:
+ pass
+
+
+def test_fetch_proper_repository():
+ repo: Repository = RepositoryStrategy.get_reporitory()
+ assert repo.__class__ is FileRepository
+
+
+def test_blueprint_upload():
+ repo: Repository = RepositoryStrategy.get_reporitory()
+ # fmt: off
+ with patch.object(manager.utils, "is_zipfile", return_value=True) as mock_is_zip, \
+ patch.object(os, "makedirs", return_value=None) as mock_mkdirs, \
+ patch.object(manager.utils, 'ZipFile', return_value=MockZipFile()
+ ):
+ repo.upload_blueprint(b"abcd", "test_cba", "1.0.a")
+ mock_is_zip.assert_called_once()
+ mock_mkdirs.assert_called_once_with('/tmp/test_cba/1.0.a', mode=0o744)
+ # fmt: on
+
+
+def test_blueprint_download():
+ repo: Repository = RepositoryStrategy.get_reporitory()
+ mock_path = [
+ ("test_cba", ["1.0.a"], []),
+ ("test_cba/1.0.a", [], ["file.txt"]),
+ ]
+ # fmt: off
+ with patch.object(os, "walk", return_value=mock_path) as mock_walk, \
+ patch.object(manager.utils, 'ZipFile', return_value=MockZipFile()), \
+ patch.object(os.path, 'exists', return_value=True
+ ):
+ repo.download_blueprint("test_cba", "1.0.a")
+ mock_walk.assert_called_once_with('/tmp/test_cba/1.0.a')
+ # fmt: on
+
+
+def test_remove_blueprint():
+ repo: Repository = RepositoryStrategy.get_reporitory()
+ with patch.object(shutil, "rmtree", return_value=None) as mock_rmtree:
+ repo.remove_blueprint("cba", "1.0a")
+ mock_rmtree.assert_called_once()