Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add ecr lifecycle policy (#4153)

Browse Source 
* Add ecr.put_lifecycle_policy

* Add ecr.get_lifecycle_policy

* Add ecr.delete_lifecycle_policy

* Add ecr lifecycle policy test for Terraform
This commit is contained in:
Anton Grübel 2021-08-09 22:55:29 +09:00 committed by GitHub
parent 4791a01935
commit 298e220122
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 966 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
moto/ecr/exceptions.py
View File

@ -2,6 +2,20 @@ from __future__ import unicode_literals
from moto.core.exceptions import JsonRESTError
class LifecyclePolicyNotFoundException(JsonRESTError):
code = 400
def __init__(self, repository_name, registry_id):
super().__init__(
error_type=__class__.__name__,
message=(
"Lifecycle policy does not exist "
f"for the repository with name '{repository_name}' "
f"in the registry with id '{registry_id}'"
),
)
class RepositoryAlreadyExistsException(JsonRESTError):
code = 400

55
moto/ecr/models.py
View File

@ -6,6 +6,7 @@ import uuid
from collections import namedtuple
from datetime import datetime
from random import random
from typing import Dict
from botocore.exceptions import ParamValidationError
@ -19,7 +20,9 @@ from moto.ecr.exceptions import (
RepositoryNotEmptyException,
InvalidParameterException,
RepositoryPolicyNotFoundException,
LifecyclePolicyNotFoundException,
)
from moto.ecr.policy_validation import EcrLifecyclePolicyValidator
from moto.iam.exceptions import MalformedPolicyDocument
from moto.iam.policy_validation import IAMPolicyDocumentValidator
from moto.utilities.tagging_service import TaggingService
@ -81,6 +84,7 @@ class Repository(BaseObject, CloudFormationModel):
encryption_config
)
self.policy = None
self.lifecycle_policy = None
self.images = []
def _determine_encryption_config(self, encryption_config):
@ -290,7 +294,7 @@ class Image(BaseObject):
class ECRBackend(BaseBackend):
def __init__(self, region_name):
self.region_name = region_name
self.repositories = {}
self.repositories: Dict[str, Repository] = {}
self.tagger = TaggingService(tagName="tags")
def reset(self):
@ -298,7 +302,7 @@ class ECRBackend(BaseBackend):
self.__dict__ = {}
self.__init__(region_name)
def _get_repository(self, name, registry_id=None):
def _get_repository(self, name, registry_id=None) -> Repository:
repo = self.repositories.get(name)
reg_id = registry_id or DEFAULT_REGISTRY_ID
@ -713,6 +717,53 @@ class ECRBackend(BaseBackend):
"policyText": policy,
}
def put_lifecycle_policy(self, registry_id, repository_name, lifecycle_policy_text):
repo = self._get_repository(repository_name, registry_id)
validator = EcrLifecyclePolicyValidator(lifecycle_policy_text)
validator.validate()
repo.lifecycle_policy = lifecycle_policy_text
return {
"registryId": repo.registry_id,
"repositoryName": repository_name,
"lifecyclePolicyText": repo.lifecycle_policy,
}
def get_lifecycle_policy(self, registry_id, repository_name):
repo = self._get_repository(repository_name, registry_id)
if not repo.lifecycle_policy:
raise LifecyclePolicyNotFoundException(repository_name, repo.registry_id)
return {
"registryId": repo.registry_id,
"repositoryName": repository_name,
"lifecyclePolicyText": repo.lifecycle_policy,
"lastEvaluatedAt": iso_8601_datetime_without_milliseconds(
datetime.utcnow()
),
}
def delete_lifecycle_policy(self, registry_id, repository_name):
repo = self._get_repository(repository_name, registry_id)
policy = repo.lifecycle_policy
if not policy:
raise LifecyclePolicyNotFoundException(repository_name, repo.registry_id)
repo.lifecycle_policy = None
return {
"registryId": repo.registry_id,
"repositoryName": repository_name,
"lifecyclePolicyText": policy,
"lastEvaluatedAt": iso_8601_datetime_without_milliseconds(
datetime.utcnow()
),
}
ecr_backends = {}
for region, ec2_backend in ec2_backends.items():

233
moto/ecr/policy_validation.py Normal file
View File

@ -0,0 +1,233 @@
import json
from moto.ecr.exceptions import InvalidParameterException
REQUIRED_RULE_PROPERTIES = {"rulePriority", "selection", "action"}
VALID_RULE_PROPERTIES = {"description", *REQUIRED_RULE_PROPERTIES}
REQUIRED_ACTION_PROPERTIES = {"type"}
VALID_ACTION_PROPERTIES = REQUIRED_ACTION_PROPERTIES
VALID_ACTION_TYPE_VALUES = {"expire"}
REQUIRED_SELECTION_PROPERTIES = {"tagStatus", "countType", "countNumber"}
VALID_SELECTION_PROPERTIES = {
"tagPrefixList",
"countUnit",
*REQUIRED_SELECTION_PROPERTIES,
}
VALID_SELECTION_TAG_STATUS_VALUES = {"tagged", "untagged", "any"}
VALID_SELECTION_COUNT_TYPE_VALUES = {"imageCountMoreThan", "sinceImagePushed"}
VALID_SELECTION_COUNT_UNIT_VALUES = {"days"}
class EcrLifecyclePolicyValidator:
INVALID_PARAMETER_ERROR_MESSAGE = (
"Invalid parameter at 'LifecyclePolicyText' failed to satisfy constraint: "
"'Lifecycle policy validation failure: "
)
def __init__(self, policy_text):
self._policy_text = policy_text
self._policy_json = {}
self._rules = []
def validate(self):
try:
self._parse_policy()
except Exception:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
"Could not map policyString into LifecyclePolicy.'",
]
)
)
try:
self._extract_rules()
except Exception:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
'object has missing required properties (["rules"])\'',
]
)
)
self._validate_rule_type()
self._validate_rule_top_properties()
def _parse_policy(self):
self._policy_json = json.loads(self._policy_text)
assert isinstance(self._policy_json, dict)
def _extract_rules(self):
assert "rules" in self._policy_json
assert isinstance(self._policy_json["rules"], list)
self._rules = self._policy_json["rules"]
def _validate_rule_type(self):
for rule in self._rules:
if not isinstance(rule, dict):
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f'instance type ({type(rule)}) does not match any allowed primitive type (allowed: ["object"])\'',
]
)
)
def _validate_rule_top_properties(self):
for rule in self._rules:
rule_properties = set(rule.keys())
missing_properties = REQUIRED_RULE_PROPERTIES - rule_properties
if missing_properties:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f"object has missing required properties ({json.dumps(sorted(missing_properties))})'",
]
)
)
for rule_property in rule_properties:
if rule_property not in VALID_RULE_PROPERTIES:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f'object instance has properties which are not allowed by the schema: (["{rule_property}"])\'',
]
)
)
self._validate_action(rule["action"])
self._validate_selection(rule["selection"])
def _validate_action(self, action):
given_properties = set(action.keys())
missing_properties = REQUIRED_ACTION_PROPERTIES - given_properties
if missing_properties:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f"object has missing required properties ({json.dumps(sorted(missing_properties))})'",
]
)
)
for given_property in given_properties:
if given_property not in VALID_ACTION_PROPERTIES:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
"object instance has properties "
f'which are not allowed by the schema: (["{given_property}"])\'',
]
)
)
self._validate_action_type(action["type"])
def _validate_action_type(self, action_type):
if action_type not in VALID_ACTION_TYPE_VALUES:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f"instance value ({action_type}) not found in enum "
f":(possible values: {json.dumps(sorted(VALID_ACTION_TYPE_VALUES))})'",
]
)
)
def _validate_selection(self, selection):
given_properties = set(selection.keys())
missing_properties = REQUIRED_SELECTION_PROPERTIES - given_properties
if missing_properties:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f"object has missing required properties ({json.dumps(sorted(missing_properties))})'",
]
)
)
for given_property in given_properties:
if given_property not in VALID_SELECTION_PROPERTIES:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
"object instance has properties "
f'which are not allowed by the schema: (["{given_property}"])\'',
]
)
)
self._validate_selection_tag_status(selection["tagStatus"])
self._validate_selection_count_type(selection["countType"])
self._validate_selection_count_unit(selection.get("countUnit"))
self._validate_selection_count_number(selection["countNumber"])
def _validate_selection_tag_status(self, tag_status):
if tag_status not in VALID_SELECTION_TAG_STATUS_VALUES:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f"instance value ({tag_status}) not found in enum "
f":(possible values: {json.dumps(sorted(VALID_SELECTION_TAG_STATUS_VALUES))})'",
]
)
)
def _validate_selection_count_type(self, count_type):
if count_type not in VALID_SELECTION_COUNT_TYPE_VALUES:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
"instance failed to match exactly one schema (matched 0 out of 2)",
]
)
)
def _validate_selection_count_unit(self, count_unit):
if not count_unit:
return None
if count_unit not in VALID_SELECTION_COUNT_UNIT_VALUES:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
f"instance value ({count_unit}) not found in enum "
f":(possible values: {json.dumps(sorted(VALID_SELECTION_COUNT_UNIT_VALUES))})'",
]
)
)
def _validate_selection_count_number(self, count_number):
if count_number < 1:
raise InvalidParameterException(
"".join(
[
self.INVALID_PARAMETER_ERROR_MESSAGE,
"numeric instance is lower than the required minimum "
f"(minimum: 1, found: {count_number})",
]
)
)

33
moto/ecr/responses.py
View File

@ -245,3 +245,36 @@ class ECRResponse(BaseResponse):
image_scan_config=image_scan_config,
)
)
def put_lifecycle_policy(self):
registry_id = self._get_param("registryId")
repository_name = self._get_param("repositoryName")
lifecycle_policy_text = self._get_param("lifecyclePolicyText")
return json.dumps(
self.ecr_backend.put_lifecycle_policy(
registry_id=registry_id,
repository_name=repository_name,
lifecycle_policy_text=lifecycle_policy_text,
)
)
def get_lifecycle_policy(self):
registry_id = self._get_param("registryId")
repository_name = self._get_param("repositoryName")
return json.dumps(
self.ecr_backend.get_lifecycle_policy(
registry_id=registry_id, repository_name=repository_name,
)
)
def delete_lifecycle_policy(self):
registry_id = self._get_param("registryId")
repository_name = self._get_param("repositoryName")
return json.dumps(
self.ecr_backend.delete_lifecycle_policy(
registry_id=registry_id, repository_name=repository_name,
)
)

1
tests/terraform-tests.success.txt
View File

@ -43,6 +43,7 @@ TestAccAWSEc2TransitGatewayVpcAttachmentDataSource
TestAccAWSEc2TransitGatewayVpnAttachmentDataSource
TestAccAWSEc2TransitGatewayPeeringAttachment
TestAccAWSEc2TransitGatewayPeeringAttachmentDataSource
TestAccAWSEcrLifecyclePolicy
TestAccAWSEcrRepository
TestAccAWSEcrRepositoryDataSource
TestAccAWSEcrRepositoryPolicy

235
tests/test_ecr/test_ecr_boto3.py
View File

@ -1720,3 +1720,238 @@ def test_delete_repository_policy_error_policy_not_exists():
f"for the repository with name '{repo_name}' "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_put_lifecycle_policy():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
policy = {
"rules": [
{
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
}
]
}
# when
response = client.put_lifecycle_policy(
repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy),
)
# then
response["registryId"].should.equal(ACCOUNT_ID)
response["repositoryName"].should.equal(repo_name)
json.loads(response["lifecyclePolicyText"]).should.equal(policy)
@mock_ecr
def test_put_lifecycle_policy_error_repo_not_exists():
# given
region_name = "eu-central-1"
client = boto3.client("ecr", region_name=region_name)
repo_name = "not-exists"
policy = {
"rules": [
{
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
}
]
}
# when
with pytest.raises(ClientError) as e:
client.put_lifecycle_policy(
repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy)
)
# then
ex = e.value
ex.operation_name.should.equal("PutLifecyclePolicy")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The repository with name '{repo_name}' does not exist "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_get_lifecycle_policy():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
policy = {
"rules": [
{
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
}
]
}
client.put_lifecycle_policy(
repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy),
)
# when
response = client.get_lifecycle_policy(repositoryName=repo_name)
# then
response["registryId"].should.equal(ACCOUNT_ID)
response["repositoryName"].should.equal(repo_name)
json.loads(response["lifecyclePolicyText"]).should.equal(policy)
response["lastEvaluatedAt"].should.be.a(datetime)
@mock_ecr
def test_get_lifecycle_policy_error_repo_not_exists():
# given
region_name = "eu-central-1"
client = boto3.client("ecr", region_name=region_name)
repo_name = "not-exists"
# when
with pytest.raises(ClientError) as e:
client.get_lifecycle_policy(repositoryName=repo_name)
# then
ex = e.value
ex.operation_name.should.equal("GetLifecyclePolicy")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The repository with name '{repo_name}' does not exist "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_get_lifecycle_policy_error_policy_not_exists():
# given
region_name = "eu-central-1"
client = boto3.client("ecr", region_name=region_name)
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
# when
with pytest.raises(ClientError) as e:
client.get_lifecycle_policy(repositoryName=repo_name)
# then
ex = e.value
ex.operation_name.should.equal("GetLifecyclePolicy")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("LifecyclePolicyNotFoundException")
ex.response["Error"]["Message"].should.equal(
"Lifecycle policy does not exist "
f"for the repository with name '{repo_name}' "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_delete_lifecycle_policy():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
policy = {
"rules": [
{
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
}
]
}
client.put_lifecycle_policy(
repositoryName=repo_name, lifecyclePolicyText=json.dumps(policy),
)
# when
response = client.delete_lifecycle_policy(repositoryName=repo_name)
# then
response["registryId"].should.equal(ACCOUNT_ID)
response["repositoryName"].should.equal(repo_name)
json.loads(response["lifecyclePolicyText"]).should.equal(policy)
response["lastEvaluatedAt"].should.be.a(datetime)
with pytest.raises(ClientError) as e:
client.get_lifecycle_policy(repositoryName=repo_name)
e.value.response["Error"]["Code"].should.contain("LifecyclePolicyNotFoundException")
@mock_ecr
def test_delete_lifecycle_policy_error_repo_not_exists():
# given
region_name = "eu-central-1"
client = boto3.client("ecr", region_name=region_name)
repo_name = "not-exists"
# when
with pytest.raises(ClientError) as e:
client.delete_lifecycle_policy(repositoryName=repo_name)
# then
ex = e.value
ex.operation_name.should.equal("DeleteLifecyclePolicy")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The repository with name '{repo_name}' does not exist "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_delete_lifecycle_policy_error_policy_not_exists():
# given
region_name = "eu-central-1"
client = boto3.client("ecr", region_name=region_name)
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
# when
with pytest.raises(ClientError) as e:
client.delete_lifecycle_policy(repositoryName=repo_name)
# then
ex = e.value
ex.operation_name.should.equal("DeleteLifecyclePolicy")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("LifecyclePolicyNotFoundException")
ex.response["Error"]["Message"].should.equal(
"Lifecycle policy does not exist "
f"for the repository with name '{repo_name}' "
f"in the registry with id '{ACCOUNT_ID}'"
)

397
tests/test_ecr/test_ecr_policy_validation.py Normal file
View File

@ -0,0 +1,397 @@
import json
import pytest
import sure # noqa
from moto.ecr.exceptions import InvalidParameterException
from moto.ecr.policy_validation import EcrLifecyclePolicyValidator
def test_validate():
# given
policy = {
"rules": [
{
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
},
{
"rulePriority": 2,
"description": "test policy",
"selection": {
"tagStatus": "tagged",
"tagPrefixList": ["3.9"],
"countType": "sinceImagePushed",
"countUnit": "days",
"countNumber": 30,
},
"action": {"type": "expire"},
},
]
}
# when/then
validator = EcrLifecyclePolicyValidator(json.dumps(policy))
validator.validate()
@pytest.mark.parametrize(
"policy",
[
"some invalid input",
[
{
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
}
],
],
ids=["not_json", "not_dict"],
)
def test_validate_error_parse(policy):
# given
# when
with pytest.raises(InvalidParameterException) as e:
validator = EcrLifecyclePolicyValidator(json.dumps(policy))
validator.validate()
# then
ex = e.value
ex.code.should.equal(400)
ex.error_type.should.equal("InvalidParameterException")
ex.message.should.equal(
"Invalid parameter at 'LifecyclePolicyText' failed to satisfy constraint: "
"'Lifecycle policy validation failure: "
"Could not map policyString into LifecyclePolicy.'"
)
@pytest.mark.parametrize(
"policy",
[
{"no_rules": "test"},
{
"rules": {
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
},
},
],
ids=["no_rules", "not_list"],
)
def test_validate_error_extract_rules(policy):
# given
# when
with pytest.raises(InvalidParameterException) as e:
validator = EcrLifecyclePolicyValidator(json.dumps(policy))
validator.validate()
# then
ex = e.value
ex.code.should.equal(400)
ex.error_type.should.equal("InvalidParameterException")
ex.message.should.equal(
"Invalid parameter at 'LifecyclePolicyText' failed to satisfy constraint: "
"'Lifecycle policy validation failure: "
'object has missing required properties (["rules"])\''
)
@pytest.mark.parametrize(
["rule", "rule_type"], [["not_dict", "string"]], ids=["not_dict"],
)
def test_validate_error_rule_type(rule, rule_type):
# given
# when
with pytest.raises(InvalidParameterException) as e:
validator = EcrLifecyclePolicyValidator(json.dumps({"rules": [rule]}))
validator.validate()
# then
ex = e.value
ex.code.should.equal(400)
ex.error_type.should.equal("InvalidParameterException")
ex.message.should.equal(
"Invalid parameter at 'LifecyclePolicyText' failed to satisfy constraint: "
"'Lifecycle policy validation failure: "
f'instance type ({type(rule)}) does not match any allowed primitive type (allowed: ["object"])\''
)
@pytest.mark.parametrize(
["rule", "error_msg"],
[
[
{
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
},
'object has missing required properties (["rulePriority"])\'',
],
[
{
"rulePriority": 1,
"description": "test policy",
"action": {"type": "expire"},
},
'object has missing required properties (["selection"])\'',
],
[
{
"rulePriority": 1,
"description": "test policy",
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
},
'object has missing required properties (["action"])\'',
],
[
{
"rulePriority": 1,
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": {"type": "expire"},
"unknown": 123,
},
'object instance has properties which are not allowed by the schema: (["unknown"])\'',
],
],
ids=["missing_rulePriority", "missing_selection", "missing_action", "unknown"],
)
def test_validate_error_rule_properties(rule, error_msg):
# given
# when
with pytest.raises(InvalidParameterException) as e:
validator = EcrLifecyclePolicyValidator(json.dumps({"rules": [rule]}))
validator.validate()
# then
ex = e.value
ex.code.should.equal(400)
ex.error_type.should.equal("InvalidParameterException")
ex.message.should.equal(
"".join(
[
"Invalid parameter at 'LifecyclePolicyText' failed to satisfy constraint: "
"'Lifecycle policy validation failure: ",
error_msg,
]
)
)
@pytest.mark.parametrize(
["action", "error_msg"],
[
[{}, 'object has missing required properties (["type"])\'',],
[
{"type": "expire", "unknown": 123},
(
"object instance has properties "
'which are not allowed by the schema: (["unknown"])\''
),
],
[
{"type": "keep"},
(
"instance value (keep) not found in enum "
':(possible values: ["expire"])\''
),
],
],
ids=["missing_type", "unknown", "unknown_type_value"],
)
def test_validate_error_action_properties(action, error_msg):
# given
# when
with pytest.raises(InvalidParameterException) as e:
validator = EcrLifecyclePolicyValidator(
json.dumps(
{
"rules": [
{
"rulePriority": 1,
"selection": {
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
"action": action,
}
]
}
)
)
validator.validate()
# then
ex = e.value
ex.code.should.equal(400)
ex.error_type.should.equal("InvalidParameterException")
ex.message.should.equal(
"".join(
[
"Invalid parameter at 'LifecyclePolicyText' failed to satisfy constraint: "
"'Lifecycle policy validation failure: ",
error_msg,
]
)
)
@pytest.mark.parametrize(
["selection", "error_msg"],
[
[
{"countType": "imageCountMoreThan", "countNumber": 30,},
'object has missing required properties (["tagStatus"])\'',
],
[
{"tagStatus": "untagged", "countNumber": 30,},
'object has missing required properties (["countType"])\'',
],
[
{"tagStatus": "untagged", "countType": "imageCountMoreThan",},
'object has missing required properties (["countNumber"])\'',
],
[
{
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 30,
"unknown": 123,
},
(
"object instance has properties "
'which are not allowed by the schema: (["unknown"])\''
),
],
[
{
"tagStatus": "unknown",
"countType": "imageCountMoreThan",
"countNumber": 30,
},
(
"instance value (unknown) not found in enum "
':(possible values: ["any", "tagged", "untagged"])\''
),
],
[
{"tagStatus": "untagged", "countType": "unknown", "countNumber": 30},
"instance failed to match exactly one schema (matched 0 out of 2)",
],
[
{
"tagStatus": "untagged",
"countType": "sinceImagePushed",
"countUnit": "unknown",
"countNumber": 30,
},
(
"instance value (unknown) not found in enum "
':(possible values: ["days"])\''
),
],
[
{
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": 0,
},
(
"numeric instance is lower than the required minimum "
f"(minimum: 1, found: 0)"
),
],
[
{
"tagStatus": "untagged",
"countType": "imageCountMoreThan",
"countNumber": -1,
},
(
"numeric instance is lower than the required minimum "
f"(minimum: 1, found: -1)"
),
],
],
ids=[
"missing_tagStatus",
"missing_countType",
"missing_countNumber",
"unknown",
"unknown_tagStatus_value",
"unknown_countType_value",
"unknown_countUnit_value",
"zero_countNumber_value",
"negative_countNumber_value",
],
)
def test_validate_error_selection_properties(selection, error_msg):
# given
# when
with pytest.raises(InvalidParameterException) as e:
validator = EcrLifecyclePolicyValidator(
json.dumps(
{
"rules": [
{
"rulePriority": 1,
"selection": selection,
"action": {"type": "expire"},
}
]
}
)
)
validator.validate()
# then
ex = e.value
ex.code.should.equal(400)
ex.error_type.should.equal("InvalidParameterException")
ex.message.should.equal(
"".join(
[
"Invalid parameter at 'LifecyclePolicyText' failed to satisfy constraint: "
"'Lifecycle policy validation failure: ",
error_msg,
]
)
)