1 """Copyright 2019 Deutsche Telekom.
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
7 http://www.apache.org/licenses/LICENSE-2.0
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.
18 from unittest.mock import patch
21 from manager.servicer import ArtifactManagerServicer
22 from proto.BluePrintCommon_pb2 import ActionIdentifiers, CommonHeader
23 from proto.BluePrintManagement_pb2 import (
24 BluePrintDownloadInput,
25 BluePrintManagementOutput,
30 from proto.BluePrintManagement_pb2_grpc import (
31 BluePrintManagementServiceStub,
32 add_BluePrintManagementServiceServicer_to_server,
34 from pytest import fixture
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"
39 class MockZipFile(zipfile.ZipFile):
40 def __init__(self, *args, **kwargs):
43 def extractall(self, path: str) -> None:
46 def write(self, *arg, **kwargs) -> None:
50 @fixture(scope="module")
51 def grpc_add_to_server():
52 """pytest-grpcio required function."""
53 return add_BluePrintManagementServiceServicer_to_server
56 @fixture(scope="module")
58 """pytest-grpcio required function."""
59 return ArtifactManagerServicer()
62 @fixture(scope="module") # noqa
63 def grpc_stub_cls(grpc_channel):
64 """pytest-grpcio required function."""
65 return BluePrintManagementServiceStub
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"
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"
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"
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"
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"
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"
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"
138 action_identifiers: ActionIdentifiers = ActionIdentifiers()
139 action_identifiers.blueprintName = "sample-cba"
140 action_identifiers.blueprintVersion = "1.0.0"
141 action_identifiers.actionName = "SampleScript"
143 file_chunk = FileChunk()
144 file_chunk.chunk = ZIP_FILE_BINARY
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
152 output: BluePrintManagementOutput = grpc_stub.uploadBlueprint(request)
154 assert output.status.code == 200
155 assert output.status.message == "success"
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"
165 action_identifiers: ActionIdentifiers = ActionIdentifiers()
166 action_identifiers.blueprintName = "sample-cba"
167 action_identifiers.blueprintVersion = "1.0.0"
168 action_identifiers.actionName = "SampleScript"
170 with patch.object(os.path, "exists", return_value=True):
171 request: BluePrintDownloadInput = BluePrintDownloadInput(
172 commonHeader=header, actionIdentifiers=action_identifiers
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
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"
187 action_identifiers: ActionIdentifiers = ActionIdentifiers()
188 action_identifiers.blueprintName = "sample-cba"
189 action_identifiers.blueprintVersion = "1.0.0"
190 action_identifiers.actionName = "SampleScript"
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"