ms/artifact-manager/tests/servicer_test.py

   1 """Copyright 2019 Deutsche Telekom.
   2
   3 Licensed under the Apache License, Version 2.0 (the "License");
   4 you may not use this file except in compliance with the License.
   5 You may obtain a copy of the License at
   6
   7     http://www.apache.org/licenses/LICENSE-2.0
   8
   9 Unless required by applicable law or agreed to in writing, software
  10 distributed under the License is distributed on an "AS IS" BASIS,
  11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 See the License for the specific language governing permissions and
  13 limitations under the License.
  14 """
  15 import os
  16 import shutil
  17 import zipfile
  18 from unittest.mock import patch
  19
  20 import manager.utils
  21 from manager.servicer import ArtifactManagerServicer
  22 from proto.BlueprintCommon_pb2 import ActionIdentifiers, CommonHeader
  23 from proto.BlueprintManagement_pb2 import (
  24     BlueprintDownloadInput,
  25     BlueprintManagementOutput,
  26     BlueprintRemoveInput,
  27     BlueprintUploadInput,
  28     FileChunk,
  29 )
  30 from proto.BlueprintManagement_pb2_grpc import (
  31     BlueprintManagementServiceStub,
  32     add_BlueprintManagementServiceServicer_to_server,
  33 )
  34 from pytest import fixture
  35
  36 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"
  37
  38
  39 class MockZipFile(zipfile.ZipFile):
  40     def __init__(self, *args, **kwargs):
  41         pass
  42
  43     def extractall(self, path: str) -> None:
  44         pass
  45
  46     def write(self, *arg, **kwargs) -> None:
  47         pass
  48
  49
  50 @fixture(scope="module")
  51 def grpc_add_to_server():
  52     """pytest-grpcio required function."""
  53     return add_BlueprintManagementServiceServicer_to_server
  54
  55
  56 @fixture(scope="module")
  57 def grpc_servicer():
  58     """pytest-grpcio required function."""
  59     return ArtifactManagerServicer()
  60
  61
  62 @fixture(scope="module")  # noqa
  63 def grpc_stub_cls(grpc_channel):
  64     """pytest-grpcio required function."""
  65     return BlueprintManagementServiceStub
  66
  67
  68 def test_servicer_upload_handler_header_failure(grpc_stub):
  69     """Test servicer upload handler."""
  70     request: BlueprintUploadInput = BlueprintUploadInput()
  71     output: BlueprintManagementOutput = grpc_stub.uploadBlueprint(request)
  72     assert output.status.code == 500
  73     assert output.status.message == "failure"
  74     assert output.status.errorMessage == "Request has to have set both Blueprint name and version"
  75
  76
  77 def test_servicer_download_handler_header_failure(grpc_stub):
  78     """Test servicer download handler."""
  79     request: BlueprintDownloadInput = BlueprintDownloadInput()
  80     output: BlueprintManagementOutput = grpc_stub.downloadBlueprint(request)
  81     assert output.status.code == 500
  82     assert output.status.message == "failure"
  83     assert output.status.errorMessage == "Request has to have set both Blueprint name and version"
  84
  85
  86 def test_servicer_remove_handler_header_failure(grpc_stub):
  87     """Test servicer remove handler."""
  88     request: BlueprintRemoveInput = BlueprintRemoveInput()
  89     output: BlueprintManagementOutput = grpc_stub.removeBlueprint(request)
  90     assert output.status.code == 500
  91     assert output.status.message == "failure"
  92     assert output.status.errorMessage == "Request has to have set both Blueprint name and version"
  93
  94
  95 def test_servicer_upload_handler_failure(grpc_stub):
  96     """Test servicer upload handler."""
  97     action_identifiers: ActionIdentifiers = ActionIdentifiers()
  98     action_identifiers.blueprintName = "sample-cba"
  99     action_identifiers.blueprintVersion = "1.0.0"
 100     request: BlueprintUploadInput = BlueprintUploadInput(actionIdentifiers=action_identifiers)
 101     output: BlueprintManagementOutput = grpc_stub.uploadBlueprint(request)
 102     assert output.status.code == 500
 103     assert output.status.message == "failure"
 104     assert output.status.errorMessage == "Invalid request"
 105
 106
 107 def test_servicer_download_handler_failure(grpc_stub):
 108     """Test servicer download handler."""
 109     action_identifiers: ActionIdentifiers = ActionIdentifiers()
 110     action_identifiers.blueprintName = "sample-cba"
 111     action_identifiers.blueprintVersion = "2.0.0"
 112     request: BlueprintDownloadInput = BlueprintDownloadInput(actionIdentifiers=action_identifiers)
 113     output: BlueprintManagementOutput = grpc_stub.downloadBlueprint(request)
 114     assert output.status.code == 500
 115     assert output.status.message == "failure"
 116     assert output.status.errorMessage == "Artifact not found"
 117
 118
 119 def test_servicer_remove_handler_failure(grpc_stub):
 120     """Test servicer remove handler."""
 121     action_identifiers: ActionIdentifiers = ActionIdentifiers()
 122     action_identifiers.blueprintName = "sample-cba"
 123     action_identifiers.blueprintVersion = "1.0.0"
 124     request: BlueprintRemoveInput = BlueprintRemoveInput(actionIdentifiers=action_identifiers)
 125     output: BlueprintManagementOutput = grpc_stub.removeBlueprint(request)
 126     assert output.status.code == 500
 127     assert output.status.message == "failure"
 128     assert output.status.errorMessage == "Artifact not found"
 129
 130
 131 def test_servicer_upload_handler_success(grpc_stub):
 132     """Test servicer upload handler."""
 133     header: CommonHeader = CommonHeader()
 134     header.requestId = "1234"
 135     header.subRequestId = "1234-1"
 136     header.originatorId = "CDS"
 137
 138     action_identifiers: ActionIdentifiers = ActionIdentifiers()
 139     action_identifiers.blueprintName = "sample-cba"
 140     action_identifiers.blueprintVersion = "1.0.0"
 141     action_identifiers.actionName = "SampleScript"
 142
 143     file_chunk = FileChunk()
 144     file_chunk.chunk = ZIP_FILE_BINARY
 145
 146     # fmt: off
 147     with patch.object(os, "makedirs", return_value=None), \
 148             patch.object(manager.utils, 'ZipFile', return_value=MockZipFile()):
 149         request: BlueprintUploadInput = BlueprintUploadInput(
 150             commonHeader=header, fileChunk=file_chunk, actionIdentifiers=action_identifiers
 151         )
 152         output: BlueprintManagementOutput = grpc_stub.uploadBlueprint(request)
 153     # fmt: on
 154     assert output.status.code == 200
 155     assert output.status.message == "success"
 156
 157
 158 def test_servicer_download_handler_success(grpc_stub):
 159     """Test servicer download handler."""
 160     header: CommonHeader = CommonHeader()
 161     header.requestId = "1234"
 162     header.subRequestId = "1234-1"
 163     header.originatorId = "CDS"
 164
 165     action_identifiers: ActionIdentifiers = ActionIdentifiers()
 166     action_identifiers.blueprintName = "sample-cba"
 167     action_identifiers.blueprintVersion = "1.0.0"
 168     action_identifiers.actionName = "SampleScript"
 169
 170     with patch.object(os.path, "exists", return_value=True):
 171         request: BlueprintDownloadInput = BlueprintDownloadInput(
 172             commonHeader=header, actionIdentifiers=action_identifiers
 173         )
 174         output: BlueprintManagementOutput = grpc_stub.downloadBlueprint(request)
 175     assert output.status.code == 200
 176     assert output.status.message == "success"
 177     assert output.fileChunk.chunk == ZIP_FILE_BINARY
 178
 179
 180 def test_servicer_remove_handler_success(grpc_stub):
 181     """Test servicer remove handler."""
 182     header: CommonHeader = CommonHeader()
 183     header.requestId = "1234"
 184     header.subRequestId = "1234-1"
 185     header.originatorId = "CDS"
 186
 187     action_identifiers: ActionIdentifiers = ActionIdentifiers()
 188     action_identifiers.blueprintName = "sample-cba"
 189     action_identifiers.blueprintVersion = "1.0.0"
 190     action_identifiers.actionName = "SampleScript"
 191
 192     with patch.object(shutil, "rmtree", return_value=None) as mock_rmtree:
 193         request: BlueprintRemoveInput = BlueprintRemoveInput(commonHeader=header, actionIdentifiers=action_identifiers)
 194         output: BlueprintManagementOutput = grpc_stub.removeBlueprint(request)
 195     assert output.status.code == 200
 196     assert output.status.message == "success"