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.py64
1 files changed, 32 insertions, 32 deletions
diff --git a/ms/artifact-manager/tests/servicer_test.py b/ms/artifact-manager/tests/servicer_test.py
index 702e1046a..131e6fb2c 100644
--- a/ms/artifact-manager/tests/servicer_test.py
+++ b/ms/artifact-manager/tests/servicer_test.py
@@ -19,17 +19,17 @@ 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,
+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
@@ -50,7 +50,7 @@ class MockZipFile(zipfile.ZipFile):
@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")
@@ -62,34 +62,34 @@ def grpc_servicer():
@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):
@@ -97,8 +97,8 @@ 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"
@@ -109,8 +109,8 @@ def test_servicer_download_handler_failure(grpc_stub):
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"
@@ -121,8 +121,8 @@ def test_servicer_remove_handler_failure(grpc_stub):
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"
@@ -146,10 +146,10 @@ def test_servicer_upload_handler_success(grpc_stub):
# 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"
@@ -168,10 +168,10 @@ def test_servicer_download_handler_success(grpc_stub):
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
@@ -190,7 +190,7 @@ def test_servicer_remove_handler_success(grpc_stub):
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"