import manager.utils
from manager.servicer import ArtifactManagerServicer
-from proto.BluePrintCommon_pb2 import ActionIdentifiers, CommonHeader
-from proto.BluePrintManagement_pb2 import (
- BluePrintDownloadInput,
- BluePrintManagementOutput,
- BluePrintRemoveInput,
- BluePrintUploadInput,
+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 proto.BlueprintManagement_pb2_grpc import (
+ BlueprintManagementServiceStub,
+ add_BlueprintManagementServiceServicer_to_server,
)
from pytest import fixture
@fixture(scope="module")
def grpc_add_to_server():
"""pytest-grpcio required function."""
- return add_BluePrintManagementServiceServicer_to_server
+ return add_BlueprintManagementServiceServicer_to_server
@fixture(scope="module")
@fixture(scope="module") # noqa
def grpc_stub_cls(grpc_channel):
"""pytest-grpcio required function."""
- return BluePrintManagementServiceStub
+ return BlueprintManagementServiceStub
def test_servicer_upload_handler_header_failure(grpc_stub):
"""Test servicer upload handler."""
- request: BluePrintUploadInput = BluePrintUploadInput()
- output: BluePrintManagementOutput = grpc_stub.uploadBlueprint(request)
+ 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"
+ 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)
+ 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"
+ 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)
+ 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"
+ assert output.status.errorMessage == "Request has to have set both Blueprint name and version"
def test_servicer_upload_handler_failure(grpc_stub):
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)
+ 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"
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)
+ 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"
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)
+ 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"
# fmt: off
with patch.object(os, "makedirs", return_value=None), \
patch.object(manager.utils, 'ZipFile', return_value=MockZipFile()):
- request: BluePrintUploadInput = BluePrintUploadInput(
+ request: BlueprintUploadInput = BlueprintUploadInput(
commonHeader=header, fileChunk=file_chunk, actionIdentifiers=action_identifiers
)
- output: BluePrintManagementOutput = grpc_stub.uploadBlueprint(request)
+ output: BlueprintManagementOutput = grpc_stub.uploadBlueprint(request)
# fmt: on
assert output.status.code == 200
assert output.status.message == "success"
action_identifiers.actionName = "SampleScript"
with patch.object(os.path, "exists", return_value=True):
- request: BluePrintDownloadInput = BluePrintDownloadInput(
+ request: BlueprintDownloadInput = BlueprintDownloadInput(
commonHeader=header, actionIdentifiers=action_identifiers
)
- output: BluePrintManagementOutput = grpc_stub.downloadBlueprint(request)
+ output: BlueprintManagementOutput = grpc_stub.downloadBlueprint(request)
assert output.status.code == 200
assert output.status.message == "success"
assert output.fileChunk.chunk == ZIP_FILE_BINARY
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)
+ request: BlueprintRemoveInput = BlueprintRemoveInput(commonHeader=header, actionIdentifiers=action_identifiers)
+ output: BlueprintManagementOutput = grpc_stub.removeBlueprint(request)
assert output.status.code == 200
assert output.status.message == "success"