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

Add support for AWS::ElasticLoadBalancingV2::ListenerRule (#3969)

Browse Source 
* Add ssm parsing support for cloudformation stacks

* Start adding unit tests for ssm parameter parsing

* Add tests for code update

* Add tests to parse ssm parameters code

* Fix black lint errors

* Fix bug.

* Need to specify region_name

* region needs to be same

* Use ssm_backends[region] instead of ssm_backend

* StringList -> string

* Linting

* check if servermode tests are on

* Typo

* Added support for ListenerRule. Will remove cruft

* Pushing latest

* Something works

* Put back ripped out code

* Save point. Incase I need more validations

* Revert "Save point. Incase I need more validations"

This reverts commit dac4953335dd9335eddb7a91a63667bc3c17104c.

* Fixed validations and some refactor

* Fix formatting

* Linting

* Cannot refactor if I have to fix all tests

* Remove exceptions for now. Will do in another PR

* Remove validations. Will add in next PR

* Fix broken tests. Almost.:

* Fix all tests. Some sneaky for now.

* Python2 making me write bad code

* OrderedDict.move_to_end() does not work in python2

* Linting

* Add more checks to field in conditions later.

* Unwnated change in FakeListener

* Revert "Unwnated change in FakeListener"

This reverts commit 962c2fdfd76fce999de9feccf1dd1c3ec48c459f.

* Add back default listener rule

* Linting fix

* Fix priority sorting

* Add cloudformation test for edge case

Co-authored-by: Bert Blommers <bblommers@users.noreply.github.com>
This commit is contained in:
Sahil Shah 2021-06-05 05:04:04 -04:00 committed by GitHub
parent a281f9d4a6
commit b82096ba37
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 295 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

230
moto/elbv2/models.py
View File

@ -1,4 +1,5 @@
from __future__ import unicode_literals
import random
import datetime
import re
@ -214,8 +215,9 @@ class FakeListener(CloudFormationModel):
self.certificate = certificate
self.certificates = [certificate] if certificate is not None else []
self.default_actions = default_actions
self._non_default_rules = []
self._default_rule = FakeRule(
self._non_default_rules = OrderedDict()
self._default_rule = OrderedDict()
self._default_rule[0] = FakeRule(
listener_arn=self.arn,
conditions=[],
priority="default",
@ -229,17 +231,17 @@ class FakeListener(CloudFormationModel):
@property
def rules(self):
return self._non_default_rules + [self._default_rule]
def remove_rule(self, rule):
self._non_default_rules.remove(rule)
def register(self, rule):
self._non_default_rules.append(rule)
self._non_default_rules = sorted(
self._non_default_rules, key=lambda x: x.priority
return OrderedDict(
list(self._non_default_rules.items()) + list(self._default_rule.items())
)
def remove_rule(self, arn):
self._non_default_rules.pop(arn)
def register(self, arn, rule):
self._non_default_rules[arn] = rule
sorted(self._non_default_rules.values(), key=lambda x: x.priority)
@staticmethod
def cloudformation_name_type():
return None
@ -262,43 +264,33 @@ class FakeListener(CloudFormationModel):
ssl_policy = properties.get("SslPolicy")
certificates = properties.get("Certificates")
# transform default actions to confirm with the rest of the code and XML templates
if "DefaultActions" in properties:
default_actions = []
for i, action in enumerate(properties["DefaultActions"]):
action_type = action["Type"]
if action_type == "forward":
default_actions.append(
{
"type": action_type,
"target_group_arn": action["TargetGroupArn"],
}
)
elif action_type in [
"redirect",
"authenticate-cognito",
"fixed-response",
]:
redirect_action = {"type": action_type}
key = (
underscores_to_camelcase(
action_type.capitalize().replace("-", "_")
)
+ "Config"
)
for redirect_config_key, redirect_config_value in action[
key
].items():
# need to match the output of _get_list_prefix
redirect_action[
camelcase_to_underscores(key)
+ "._"
+ camelcase_to_underscores(redirect_config_key)
] = redirect_config_value
default_actions.append(redirect_action)
else:
raise InvalidActionTypeError(action_type, i + 1)
else:
default_actions = None
default_actions = []
for i, action in enumerate(properties["DefaultActions"]):
action_type = action["Type"]
if action_type == "forward":
default_actions.append(
{"type": action_type, "target_group_arn": action["TargetGroupArn"],}
)
elif action_type in [
"redirect",
"authenticate-cognito",
"fixed-response",
]:
redirect_action = {"type": action_type}
key = (
underscores_to_camelcase(action_type.capitalize().replace("-", "_"))
+ "Config"
)
for redirect_config_key, redirect_config_value in action[key].items():
# need to match the output of _get_list_prefix
redirect_action[
camelcase_to_underscores(key)
+ "._"
+ camelcase_to_underscores(redirect_config_key)
] = redirect_config_value
default_actions.append(redirect_action)
else:
raise InvalidActionTypeError(action_type, i + 1)
listener = elbv2_backend.create_listener(
load_balancer_arn, protocol, port, ssl_policy, certificates, default_actions
@ -306,6 +298,81 @@ class FakeListener(CloudFormationModel):
return listener
class FakeListenerRule(CloudFormationModel):
def __init__(
self, listener_arn, arn, conditions, priority, actions,
):
self.listener_arn = listener_arn
self.arn = arn
self.conditions = conditions
self.actions = actions
self.priority = priority
@property
def physical_resource_id(self):
return self.arn
@staticmethod
def cloudformation_type():
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-elasticloadbalancingv2-listenerrule.html
return "AWS::ElasticLoadBalancingV2::ListenerRule"
@classmethod
def create_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name
):
properties = cloudformation_json["Properties"]
elbv2_backend = elbv2_backends[region_name]
listener_arn = properties.get("ListenerArn")
priority = properties.get("Priority")
conditions = properties.get("Conditions")
# transform Actions to confirm with the rest of the code and XML templates
default_actions = []
for i, action in enumerate(properties["Actions"]):
action_type = action["Type"]
if action_type == "forward":
default_actions.append(
{"type": action_type, "target_group_arn": action["TargetGroupArn"],}
)
elif action_type in [
"redirect",
"authenticate-cognito",
"fixed-response",
]:
redirect_action = {"type": action_type}
key = (
underscores_to_camelcase(action_type.capitalize().replace("-", "_"))
+ "Config"
)
for redirect_config_key, redirect_config_value in action[key].items():
# need to match the output of _get_list_prefix
redirect_action[
camelcase_to_underscores(key)
+ "._"
+ camelcase_to_underscores(redirect_config_key)
] = redirect_config_value
default_actions.append(redirect_action)
else:
raise InvalidActionTypeError(action_type, i + 1)
listener_rule = elbv2_backend.create_rule(
listener_arn, conditions, priority, default_actions
)
return listener_rule
class FakeRule(BaseModel):
def __init__(self, listener_arn, conditions, priority, actions, is_default):
self.listener_arn = listener_arn
self.arn = listener_arn.replace(":listener/", ":listener-rule/") + "/%s" % (
id(self)
)
self.conditions = conditions
self.priority = priority # int or 'default'
self.actions = actions
self.is_default = is_default
class FakeAction(BaseModel):
def __init__(self, data):
self.data = data
@ -340,18 +407,6 @@ class FakeAction(BaseModel):
return template.render(action=self)
class FakeRule(BaseModel):
def __init__(self, listener_arn, conditions, priority, actions, is_default):
self.listener_arn = listener_arn
self.arn = listener_arn.replace(":listener/", ":listener-rule/") + "/%s" % (
id(self)
)
self.conditions = conditions
self.priority = priority # int or 'default'
self.actions = actions
self.is_default = is_default
class FakeBackend(BaseModel):
def __init__(self, instance_port):
self.instance_port = instance_port
@ -564,35 +619,44 @@ class ELBv2Backend(BaseBackend):
# validate conditions
for condition in conditions:
field = condition["field"]
if field not in ["path-pattern", "host-header"]:
raise InvalidConditionFieldError(field)
if "field" in condition:
field = condition["field"]
if field not in ["path-pattern", "host-header"]:
raise InvalidConditionFieldError(field)
values = condition["values"]
if len(values) == 0:
raise InvalidConditionValueError("A condition value must be specified")
if len(values) > 1:
raise InvalidConditionValueError(
"The '%s' field contains too many values; the limit is '1'" % field
)
values = condition["values"]
if len(values) == 0:
raise InvalidConditionValueError(
"A condition value must be specified"
)
if len(values) > 1:
raise InvalidConditionValueError(
"The '%s' field contains too many values; the limit is '1'"
% field
)
# TODO: check pattern of value for 'host-header'
# TODO: check pattern of value for 'path-pattern'
# TODO: check pattern of value for 'host-header'
# TODO: check pattern of value for 'path-pattern'
# validate Priority
for rule in listener.rules:
for rule in listener.rules.values():
if rule.priority == priority:
raise PriorityInUseError()
self._validate_actions(actions)
arn = (
listener_arn.replace(":listener/", ":listener-rule/")
+ "{}".format(random.randint(0, 50))
+ "/%s" % (id(self))
)
# TODO: check for error 'TooManyRegistrationsForTargetId'
# TODO: check for error 'TooManyRules'
# create rule
rule = FakeRule(listener.arn, conditions, priority, actions, is_default=False)
listener.register(rule)
return [rule]
rule = FakeListenerRule(listener.arn, arn, conditions, priority, actions,)
listener.register(arn, rule)
return rule
def _validate_actions(self, actions):
# validate Actions
@ -789,14 +853,14 @@ Member must satisfy regular expression pattern: {}".format(
)
if listener_arn:
listener = self.describe_listeners(None, [listener_arn])[0]
return listener.rules
return listener.rules.values()
# search for rule arns
matched_rules = []
for load_balancer_arn in self.load_balancers:
listeners = self.load_balancers.get(load_balancer_arn).listeners.values()
for listener in listeners:
for rule in listener.rules:
for rule in listener.rules.values():
if rule.arn in rule_arns:
matched_rules.append(rule)
return matched_rules
@ -853,9 +917,9 @@ Member must satisfy regular expression pattern: {}".format(
for load_balancer_arn in self.load_balancers:
listeners = self.load_balancers.get(load_balancer_arn).listeners.values()
for listener in listeners:
for rule in listener.rules:
for rule in listener.rules.values():
if rule.arn == arn:
listener.remove_rule(rule)
listener.remove_rule(rule.arn)
return
# should raise RuleNotFound Error according to the AWS API doc
@ -923,7 +987,7 @@ Member must satisfy regular expression pattern: {}".format(
rule.conditions = conditions
if actions:
rule.actions = actions
return [rule]
return rule
def register_targets(self, target_group_arn, instances):
target_group = self.target_groups.get(target_group_arn)
@ -965,7 +1029,7 @@ Member must satisfy regular expression pattern: {}".format(
given_rule = _given_rules[0]
listeners = self.describe_listeners(None, [given_rule.listener_arn])
listener = listeners[0]
for rule_in_listener in listener.rules:
for rule_in_listener in listener.rules.values():
if rule_in_listener.priority == priority:
raise PriorityInUseError()
# modify
@ -1201,7 +1265,7 @@ Member must satisfy regular expression pattern: {}".format(
def _any_listener_using(self, target_group_arn):
for load_balancer in self.load_balancers.values():
for listener in load_balancer.listeners.values():
for rule in listener.rules:
for rule in listener.rules.values():
for action in rule.actions:
if action.data.get("target_group_arn") == target_group_arn:
return True

32
moto/elbv2/responses.py
View File

@ -151,7 +151,7 @@ class ELBV2Response(BaseResponse):
@amzn_request_id
def create_rule(self):
lister_arn = self._get_param("ListenerArn")
listener_arn = self._get_param("ListenerArn")
_conditions = self._get_list_prefix("Conditions.member")
conditions = []
for _condition in _conditions:
@ -166,7 +166,7 @@ class ELBV2Response(BaseResponse):
priority = self._get_int_param("Priority")
actions = self._get_list_prefix("Actions.member")
rules = self.elbv2_backend.create_rule(
listener_arn=lister_arn,
listener_arn=listener_arn,
conditions=conditions,
priority=priority,
actions=actions,
@ -268,7 +268,7 @@ class ELBV2Response(BaseResponse):
)
else None
)
all_rules = self.elbv2_backend.describe_rules(listener_arn, rule_arns)
all_rules = list(self.elbv2_backend.describe_rules(listener_arn, rule_arns))
all_arns = [rule.arn for rule in all_rules]
page_size = self._get_int_param("PageSize", 50) # set 50 for temporary
@ -720,11 +720,10 @@ CREATE_LOAD_BALANCER_TEMPLATE = """<CreateLoadBalancerResponse xmlns="http://ela
CREATE_RULE_TEMPLATE = """<CreateRuleResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<CreateRuleResult>
<Rules>
{% for rule in rules %}
<member>
<IsDefault>{{ "true" if rule.is_default else "false" }}</IsDefault>
<IsDefault>{{ "true" if rules.is_default else "false" }}</IsDefault>
<Conditions>
{% for condition in rule.conditions %}
{% for condition in rules.conditions %}
<member>
<Field>{{ condition["field"] }}</Field>
<Values>
@ -735,9 +734,9 @@ CREATE_RULE_TEMPLATE = """<CreateRuleResponse xmlns="http://elasticloadbalancing
</member>
{% endfor %}
</Conditions>
<Priority>{{ rule.priority }}</Priority>
<Priority>{{ rules.priority }}</Priority>
<Actions>
{% for action in rule.actions %}
{% for action in rules.actions %}
<member>
<Type>{{ action["type"] }}</Type>
{% if action["type"] == "forward" %}
@ -748,9 +747,8 @@ CREATE_RULE_TEMPLATE = """<CreateRuleResponse xmlns="http://elasticloadbalancing
</member>
{% endfor %}
</Actions>
<RuleArn>{{ rule.arn }}</RuleArn>
<RuleArn>{{ rules.arn }}</RuleArn>
</member>
{% endfor %}
</Rules>
</CreateRuleResult>
<ResponseMetadata>
@ -898,7 +896,7 @@ DESCRIBE_RULES_TEMPLATE = """<DescribeRulesResponse xmlns="http://elasticloadbal
<Rules>
{% for rule in rules %}
<member>
<IsDefault>{{ "true" if rule.is_default else "false" }}</IsDefault>
<IsDefault>{{ "true" if rules.is_default else "false" }}</IsDefault>
<Conditions>
{% for condition in rule.conditions %}
<member>
@ -1038,11 +1036,10 @@ CONFIGURE_HEALTH_CHECK_TEMPLATE = """<ConfigureHealthCheckResponse xmlns="http:/
MODIFY_RULE_TEMPLATE = """<ModifyRuleResponse xmlns="http://elasticloadbalancing.amazonaws.com/doc/2015-12-01/">
<ModifyRuleResult>
<Rules>
{% for rule in rules %}
<member>
<IsDefault>{{ "true" if rule.is_default else "false" }}</IsDefault>
<IsDefault>{{ "true" if rules.is_default else "false" }}</IsDefault>
<Conditions>
{% for condition in rule.conditions %}
{% for condition in rules.conditions %}
<member>
<Field>{{ condition["field"] }}</Field>
<Values>
@ -1053,17 +1050,16 @@ MODIFY_RULE_TEMPLATE = """<ModifyRuleResponse xmlns="http://elasticloadbalancing
</member>
{% endfor %}
</Conditions>
<Priority>{{ rule.priority }}</Priority>
<Priority>{{ rules.priority }}</Priority>
<Actions>
{% for action in rule.actions %}
{% for action in rules.actions %}
<member>
{{ action.to_xml() }}
</member>
{% endfor %}
</Actions>
<RuleArn>{{ rule.arn }}</RuleArn>
<RuleArn>{{ rules.arn }}</RuleArn>
</member>
{% endfor %}
</Rules>
</ModifyRuleResult>
<ResponseMetadata>

92
tests/test_cloudformation/test_cloudformation_stack_integration.py
View File

@ -12,6 +12,7 @@ import boto.ec2
import boto.ec2.autoscale
import boto.ec2.elb
from boto.exception import BotoServerError
from botocore.exceptions import ClientError
import boto.iam
import boto.rds
import boto.redshift
@ -2154,6 +2155,74 @@ def test_stack_spot_fleet_should_figure_out_default_price():
assert "SpotPrice" not in launch_spec2
@mock_ec2
@mock_elbv2
@mock_cloudformation
def test_invalid_action_type_listener_rule():
invalid_listener_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"alb": {
"Type": "AWS::ElasticLoadBalancingV2::LoadBalancer",
"Properties": {
"Name": "myelbv2",
"Scheme": "internet-facing",
"Subnets": [{"Ref": "mysubnet"}],
},
},
"mytargetgroup1": {
"Type": "AWS::ElasticLoadBalancingV2::TargetGroup",
"Properties": {"Name": "mytargetgroup1",},
},
"mytargetgroup2": {
"Type": "AWS::ElasticLoadBalancingV2::TargetGroup",
"Properties": {"Name": "mytargetgroup2",},
},
"listener": {
"Type": "AWS::ElasticLoadBalancingV2::Listener",
"Properties": {
"DefaultActions": [
{"Type": "forward", "TargetGroupArn": {"Ref": "mytargetgroup1"}}
],
"LoadBalancerArn": {"Ref": "alb"},
"Port": "80",
"Protocol": "HTTP",
},
},
"rule": {
"Type": "AWS::ElasticLoadBalancingV2::ListenerRule",
"Properties": {
"Actions": [
{
"Type": "forward2",
"TargetGroupArn": {"Ref": "mytargetgroup2"},
}
],
"Conditions": [{"field": "path-pattern", "values": ["/*"]}],
"ListenerArn": {"Ref": "listener"},
"Priority": 2,
},
},
"myvpc": {
"Type": "AWS::EC2::VPC",
"Properties": {"CidrBlock": "10.0.0.0/16"},
},
"mysubnet": {
"Type": "AWS::EC2::Subnet",
"Properties": {"CidrBlock": "10.0.0.0/27", "VpcId": {"Ref": "myvpc"}},
},
},
}
listener_template_json = json.dumps(invalid_listener_template)
cfn_conn = boto3.client("cloudformation", "us-west-1")
cfn_conn.create_stack.when.called_with(
StackName="listener_stack", TemplateBody=listener_template_json
).should.throw(ClientError)
@mock_ec2
@mock_elbv2
@mock_cloudformation
@ -2234,6 +2303,17 @@ def test_stack_elbv2_resources_integration():
"Protocol": "HTTP",
},
},
"rule": {
"Type": "AWS::ElasticLoadBalancingV2::ListenerRule",
"Properties": {
"Actions": [
{"Type": "forward", "TargetGroupArn": {"Ref": "mytargetgroup2"}}
],
"Conditions": [{"field": "path-pattern", "values": ["/*"]}],
"ListenerArn": {"Ref": "listener"},
"Priority": 2,
},
},
"myvpc": {
"Type": "AWS::EC2::VPC",
"Properties": {"CidrBlock": "10.0.0.0/16"},
@ -2312,6 +2392,18 @@ def test_stack_elbv2_resources_integration():
[{"Type": "forward", "TargetGroupArn": target_groups[0]["TargetGroupArn"]}]
)
listener_rule = elbv2_conn.describe_rules(ListenerArn=listeners[0]["ListenerArn"])[
"Rules"
]
len(listener_rule).should.equal(2)
listener_rule[0]["Priority"].should.equal("2")
listener_rule[0]["Actions"].should.equal(
[{"Type": "forward", "TargetGroupArn": target_groups[1]["TargetGroupArn"]}]
)
listener_rule[0]["Conditions"].should.equal(
[{"Field": "path-pattern", "Values": ["/*"]}]
)
# test outputs
stacks = cfn_conn.describe_stacks(StackName="elb_stack")["Stacks"]
len(stacks).should.equal(1)

62
tests/test_elbv2/test_elbv2.py
View File

@ -345,6 +345,7 @@ def test_create_target_group_and_listeners():
response.get("TargetGroups").should.have.length_of(1)
# And another with SSL
actions = {"Type": "forward", "TargetGroupArn": target_group.get("TargetGroupArn")}
response = conn.create_listener(
LoadBalancerArn=load_balancer_arn,
Protocol="HTTPS",
@ -356,9 +357,7 @@ def test_create_target_group_and_listeners():
)
}
],
DefaultActions=[
{"Type": "forward", "TargetGroupArn": target_group.get("TargetGroupArn")}
],
DefaultActions=[actions],
)
listener = response.get("Listeners")[0]
listener.get("Port").should.equal(443)
@ -391,6 +390,12 @@ def test_create_target_group_and_listeners():
)
response.get("Listeners").should.have.length_of(2)
listener_response = conn.create_rule(
ListenerArn=http_listener_arn,
Conditions=[{"Field": "path-pattern", "Values": ["/*"]},],
Priority=3,
Actions=[actions],
)
# Try to delete the target group and it fails because there's a
# listener referencing it
with pytest.raises(ClientError) as e:
@ -1011,11 +1016,12 @@ def test_handle_listener_rules():
Actions=[
{"TargetGroupArn": target_group.get("TargetGroupArn"), "Type": "forward"}
],
)["Rules"][0]
created_rule["Priority"].should.equal("100")
)
rule = created_rule.get("Rules")[0]
rule["Priority"].should.equal("100")
# check if rules is sorted by priority
priority = 50
priority = 500
host = "yyy.example.com"
path_pattern = "foobar"
rules = conn.create_rule(
@ -1059,7 +1065,7 @@ def test_handle_listener_rules():
obtained_rules = conn.describe_rules(ListenerArn=http_listener_arn)
len(obtained_rules["Rules"]).should.equal(3)
priorities = [rule["Priority"] for rule in obtained_rules["Rules"]]
priorities.should.equal(["50", "100", "default"])
priorities.should.equal(["100", "500", "default"])
first_rule = obtained_rules["Rules"][0]
second_rule = obtained_rules["Rules"][1]
@ -1076,7 +1082,6 @@ def test_handle_listener_rules():
ListenerArn=http_listener_arn, PageSize=1, Marker=next_marker
)
len(following_rules["Rules"]).should.equal(1)
following_rules.should.have.key("NextMarker")
following_rules["Rules"][0]["RuleArn"].should_not.equal(
obtained_rules["Rules"][0]["RuleArn"]
)
@ -1105,11 +1110,10 @@ def test_handle_listener_rules():
"PathPatternConfig": {"Values": [new_pathpatternconfig_pattern]},
},
],
)["Rules"][0]
)
rules = conn.describe_rules(ListenerArn=http_listener_arn)
obtained_rule = rules["Rules"][0]
modified_rule.should.equal(obtained_rule)
obtained_rule["Conditions"][0]["Values"][0].should.equal(new_host)
obtained_rule["Conditions"][1]["Values"][0].should.equal(new_path_pattern)
obtained_rule["Conditions"][2]["Values"][0].should.equal(
@ -1696,20 +1700,20 @@ def test_redirect_action_listener_rule():
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn")
action = {
"Type": "redirect",
"RedirectConfig": {
"Protocol": "HTTPS",
"Port": "443",
"StatusCode": "HTTP_301",
},
}
response = conn.create_listener(
LoadBalancerArn=load_balancer_arn,
Protocol="HTTP",
Port=80,
DefaultActions=[
{
"Type": "redirect",
"RedirectConfig": {
"Protocol": "HTTPS",
"Port": "443",
"StatusCode": "HTTP_301",
},
}
],
DefaultActions=[action],
)
listener = response.get("Listeners")[0]
@ -1726,6 +1730,12 @@ def test_redirect_action_listener_rule():
listener.get("DefaultActions").should.equal(expected_default_actions)
listener_arn = listener.get("ListenerArn")
listener_response = conn.create_rule(
ListenerArn=listener_arn,
Conditions=[{"Field": "path-pattern", "Values": ["/*"]},],
Priority=3,
Actions=[action],
)
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
describe_rules_response["Rules"][0]["Actions"].should.equal(
expected_default_actions
@ -1789,6 +1799,12 @@ def test_cognito_action_listener_rule():
listener.get("DefaultActions")[0].should.equal(action)
listener_arn = listener.get("ListenerArn")
listener_response = conn.create_rule(
ListenerArn=listener_arn,
Conditions=[{"Field": "path-pattern", "Values": ["/*"]},],
Priority=3,
Actions=[action],
)
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
describe_rules_response["Rules"][0]["Actions"][0].should.equal(action)
@ -1844,6 +1860,12 @@ def test_fixed_response_action_listener_rule():
listener.get("DefaultActions")[0].should.equal(action)
listener_arn = listener.get("ListenerArn")
listener_response = conn.create_rule(
ListenerArn=listener_arn,
Conditions=[{"Field": "path-pattern", "Values": ["/*"]},],
Priority=3,
Actions=[action],
)
describe_rules_response = conn.describe_rules(ListenerArn=listener_arn)
describe_rules_response["Rules"][0]["Actions"][0].should.equal(action)