in pce/validator/validation_suite.py [0:0]
def validate_cluster_definition(self, pce: PCE) -> ValidationResult:
"""
Check CPU, memory and image name values are according to the standard
"""
c = pce.pce_compute.container_definition
if not c:
return ValidationResult(
ValidationResultCode.ERROR,
ValidationErrorDescriptionTemplate.CLUSTER_DEFINITION_NOT_SET.value,
)
message_by_level = defaultdict(list)
for resource, expected_value, level in (
(
ClusterResourceType.CPU,
CONTAINER_CPU,
ValidationResultCode.ERROR,
),
(
ClusterResourceType.MEMORY,
CONTAINER_MEMORY,
ValidationResultCode.ERROR,
),
(
ClusterResourceType.IMAGE,
CONTAINER_IMAGE,
ValidationResultCode.WARNING,
),
):
value = getattr(c, resource.value)
if value != expected_value:
message_template = (
ValidationErrorDescriptionTemplate.CLUSTER_DEFINITION_WRONG_VALUE.value
if ValidationResultCode.ERROR == level
else ValidationWarningDescriptionTemplate.CLUSTER_DEFINITION_FLAGGED_VALUE.value
)
message = message_template.format(
resource_name=resource.value.title(),
value=value,
expected_value=expected_value,
)
message_by_level[level].append(message)
if message_by_level[ValidationResultCode.ERROR]:
return ValidationResult(
ValidationResultCode.ERROR,
ValidationErrorDescriptionTemplate.CLUSTER_DEFINITION_WRONG_VALUES.value.format(
error_reasons=",".join(message_by_level[ValidationResultCode.ERROR])
),
ValidationErrorSolutionHintTemplate.CLUSTER_DEFINITION_WRONG_VALUES.value,
)
elif message_by_level[ValidationResultCode.WARNING]:
return ValidationResult(
ValidationResultCode.WARNING,
ValidationWarningDescriptionTemplate.CLUSTER_DEFINITION_FLAGGED_VALUES.value.format(
warning_reasons=",".join(
message_by_level[ValidationResultCode.WARNING]
)
),
)
return ValidationResult(ValidationResultCode.SUCCESS)