aboutsummaryrefslogtreecommitdiffstats
path: root/ms/artifact-manager/tests/servicer_test.py
diff options
context:
space:
mode:
Diffstat (limited to 'ms/artifact-manager/tests/servicer_test.py')
-rw-r--r--ms/artifact-manager/tests/servicer_test.py196
1 files changed, 196 insertions, 0 deletions
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"