From be4c46420944531765ecc8bae7305086d71a36d0 Mon Sep 17 00:00:00 2001 From: Marek Szwalkiewicz Date: Thu, 30 Jan 2020 13:49:18 +0000 Subject: Add Artifact Manager service. Adds a micro service that offers gRPC interface for CBA artifacts manipulation. By default the service is attached to py-executor but can be ran as a separate service if needed in the future. Issue-ID: CCSDK-1988 Change-Id: I40e20f085ae1c1e81a48f76dbea181af28d9bd0d Signed-off-by: Marek Szwalkiewicz --- ms/artifact-manager/tests/servicer_test.py | 196 +++++++++++++++++++++++++++++ 1 file changed, 196 insertions(+) create mode 100644 ms/artifact-manager/tests/servicer_test.py (limited to 'ms/artifact-manager/tests/servicer_test.py') 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" -- cgit 1.2.3-korg