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

Techdebt: Replace sure with regular assertions in ELBv2 (#6558)

Browse Source
This commit is contained in:
Bert Blommers 2023-07-24 12:50:35 +00:00 committed by GitHub
parent 9cb38575f1
commit 6b0d022925
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 527 additions and 582 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

606
tests/test_elbv2/test_elbv2.py
View File

File diff suppressed because it is too large Load Diff

89
tests/test_elbv2/test_elbv2_cloudformation.py
View File

@ -1,6 +1,5 @@
import boto3 import boto3
import json import json
import sure # noqa # pylint: disable=unused-import
from moto import mock_elbv2, mock_ec2, mock_cloudformation from moto import mock_elbv2, mock_ec2, mock_cloudformation
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -68,32 +67,24 @@ def test_redirect_action_listener_rule_cloudformation():
template_json = json.dumps(template) template_json = json.dumps(template)
cnf_conn.create_stack(StackName="test-stack", TemplateBody=template_json) cnf_conn.create_stack(StackName="test-stack", TemplateBody=template_json)
describe_load_balancers_response = elbv2_client.describe_load_balancers( resp = elbv2_client.describe_load_balancers(Names=["my-lb"])
Names=["my-lb"] assert len(resp["LoadBalancers"]) == 1
) load_balancer_arn = resp["LoadBalancers"][0]["LoadBalancerArn"]
describe_load_balancers_response["LoadBalancers"].should.have.length_of(1)
load_balancer_arn = describe_load_balancers_response["LoadBalancers"][0][ listeners = elbv2_client.describe_listeners(LoadBalancerArn=load_balancer_arn)
"LoadBalancerArn"
assert len(listeners["Listeners"]) == 1
assert listeners["Listeners"][0]["DefaultActions"] == [
{
"Type": "redirect",
"RedirectConfig": {
"Port": "443",
"Protocol": "HTTPS",
"StatusCode": "HTTP_301",
},
}
] ]
describe_listeners_response = elbv2_client.describe_listeners(
LoadBalancerArn=load_balancer_arn
)
describe_listeners_response["Listeners"].should.have.length_of(1)
describe_listeners_response["Listeners"][0]["DefaultActions"].should.equal(
[
{
"Type": "redirect",
"RedirectConfig": {
"Port": "443",
"Protocol": "HTTPS",
"StatusCode": "HTTP_301",
},
}
]
)
@mock_elbv2 @mock_elbv2
@mock_cloudformation @mock_cloudformation
@ -167,19 +158,17 @@ def test_cognito_action_listener_rule_cloudformation():
LoadBalancerArn=load_balancer_arn LoadBalancerArn=load_balancer_arn
) )
describe_listeners_response["Listeners"].should.have.length_of(1) assert len(describe_listeners_response["Listeners"]) == 1
describe_listeners_response["Listeners"][0]["DefaultActions"].should.equal( assert describe_listeners_response["Listeners"][0]["DefaultActions"] == [
[ {
{ "Type": "authenticate-cognito",
"Type": "authenticate-cognito", "AuthenticateCognitoConfig": {
"AuthenticateCognitoConfig": { "UserPoolArn": f"arn:aws:cognito-idp:us-east-1:{ACCOUNT_ID}:userpool/us-east-1_ABCD1234",
"UserPoolArn": f"arn:aws:cognito-idp:us-east-1:{ACCOUNT_ID}:userpool/us-east-1_ABCD1234", "UserPoolClientId": "abcd1234abcd",
"UserPoolClientId": "abcd1234abcd", "UserPoolDomain": "testpool",
"UserPoolDomain": "testpool", },
}, }
} ]
]
)
@mock_ec2 @mock_ec2
@ -330,16 +319,14 @@ def test_fixed_response_action_listener_rule_cloudformation():
LoadBalancerArn=load_balancer_arn LoadBalancerArn=load_balancer_arn
) )
describe_listeners_response["Listeners"].should.have.length_of(1) assert len(describe_listeners_response["Listeners"]) == 1
describe_listeners_response["Listeners"][0]["DefaultActions"].should.equal( assert describe_listeners_response["Listeners"][0]["DefaultActions"] == [
[ {
{ "Type": "fixed-response",
"Type": "fixed-response", "FixedResponseConfig": {
"FixedResponseConfig": { "ContentType": "text/plain",
"ContentType": "text/plain", "MessageBody": "This page does not exist",
"MessageBody": "This page does not exist", "StatusCode": "404",
"StatusCode": "404", },
}, }
} ]
]
)

2
tests/test_elbv2/test_elbv2_integration.py
View File

@ -65,4 +65,4 @@ def test_modify_listener_using_iam_certificate():
Certificates=[{"CertificateArn": iam_arn}], Certificates=[{"CertificateArn": iam_arn}],
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}], DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
)["Listeners"][0] )["Listeners"][0]
listener["Certificates"].should.equal([{"CertificateArn": iam_arn}]) assert listener["Certificates"] == [{"CertificateArn": iam_arn}]

28
tests/test_elbv2/test_elbv2_listener_rule_tags.py
View File

@ -1,5 +1,3 @@
import sure # noqa # pylint: disable=unused-import
from moto import mock_elbv2, mock_ec2 from moto import mock_elbv2, mock_ec2
from .test_elbv2 import create_load_balancer from .test_elbv2 import create_load_balancer
@ -35,7 +33,7 @@ def test_create_listener_rule_with_tags():
# Ensure the tags persisted # Ensure the tags persisted
response = elbv2.describe_tags(ResourceArns=[rule_arn]) response = elbv2.describe_tags(ResourceArns=[rule_arn])
tags = {d["Key"]: d["Value"] for d in response["TagDescriptions"][0]["Tags"]} tags = {d["Key"]: d["Value"] for d in response["TagDescriptions"][0]["Tags"]}
tags.should.equal({"k1": "v1"}) assert tags == {"k1": "v1"}
@mock_elbv2 @mock_elbv2
@ -62,7 +60,7 @@ def test_listener_rule_add_remove_tags():
elbv2.add_tags(ResourceArns=[rule_arn], Tags=[{"Key": "a", "Value": "b"}]) elbv2.add_tags(ResourceArns=[rule_arn], Tags=[{"Key": "a", "Value": "b"}])
tags = elbv2.describe_tags(ResourceArns=[rule_arn])["TagDescriptions"][0]["Tags"] tags = elbv2.describe_tags(ResourceArns=[rule_arn])["TagDescriptions"][0]["Tags"]
tags.should.equal([{"Key": "a", "Value": "b"}]) assert tags == [{"Key": "a", "Value": "b"}]
elbv2.add_tags( elbv2.add_tags(
ResourceArns=[rule_arn], ResourceArns=[rule_arn],
@ -74,20 +72,16 @@ def test_listener_rule_add_remove_tags():
) )
tags = elbv2.describe_tags(ResourceArns=[rule_arn])["TagDescriptions"][0]["Tags"] tags = elbv2.describe_tags(ResourceArns=[rule_arn])["TagDescriptions"][0]["Tags"]
tags.should.equal( assert tags == [
[ {"Key": "a", "Value": "b"},
{"Key": "a", "Value": "b"}, {"Key": "b", "Value": "b"},
{"Key": "b", "Value": "b"}, {"Key": "c", "Value": "b"},
{"Key": "c", "Value": "b"}, ]
]
)
elbv2.remove_tags(ResourceArns=[rule_arn], TagKeys=["a"]) elbv2.remove_tags(ResourceArns=[rule_arn], TagKeys=["a"])
tags = elbv2.describe_tags(ResourceArns=[rule_arn])["TagDescriptions"][0]["Tags"] tags = elbv2.describe_tags(ResourceArns=[rule_arn])["TagDescriptions"][0]["Tags"]
tags.should.equal( assert tags == [
[ {"Key": "b", "Value": "b"},
{"Key": "b", "Value": "b"}, {"Key": "c", "Value": "b"},
{"Key": "c", "Value": "b"}, ]
]
)

87
tests/test_elbv2/test_elbv2_listener_rules.py
View File

@ -1,7 +1,6 @@
import boto3 import boto3
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
import pytest import pytest
import sure # noqa # pylint: disable=unused-import
from moto import mock_elbv2, mock_ec2 from moto import mock_elbv2, mock_ec2
@ -117,24 +116,24 @@ def test_create_rule_condition(condition):
) )
# assert create_rule response # assert create_rule response
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Priority"].should.equal("100") assert rule["Priority"] == "100"
rule["Conditions"].should.equal([condition]) assert rule["Conditions"] == [condition]
# assert describe_rules response # assert describe_rules response
response = conn.describe_rules(ListenerArn=http_listener_arn) response = conn.describe_rules(ListenerArn=http_listener_arn)
response["Rules"].should.have.length_of(2) # including the default rule assert len(response["Rules"]) == 2 # including the default rule
# assert describe_rules with arn filter response # assert describe_rules with arn filter response
rule = response["Rules"][0] rule = response["Rules"][0]
rule["Conditions"].should.equal([condition]) assert rule["Conditions"] == [condition]
response = conn.describe_rules(RuleArns=[rule["RuleArn"]]) response = conn.describe_rules(RuleArns=[rule["RuleArn"]])
response["Rules"].should.equal([rule]) assert response["Rules"] == [rule]
# assert describe_tags response # assert describe_tags response
response = conn.describe_tags(ResourceArns=[rule["RuleArn"]]) response = conn.describe_tags(ResourceArns=[rule["RuleArn"]])
response["TagDescriptions"].should.have.length_of(1) assert len(response["TagDescriptions"]) == 1
@mock_elbv2 @mock_elbv2
@ -188,9 +187,9 @@ def test_modify_rule_condition(create_condition, modify_condition):
# modify_rule # modify_rule
response = conn.modify_rule(RuleArn=rule["RuleArn"], Conditions=[modify_condition]) response = conn.modify_rule(RuleArn=rule["RuleArn"], Conditions=[modify_condition])
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
modified_rule = response.get("Rules")[0] modified_rule = response.get("Rules")[0]
modified_rule["Conditions"].should.equal([modify_condition]) assert modified_rule["Conditions"] == [modify_condition]
@mock_elbv2 @mock_elbv2
@ -309,8 +308,8 @@ def test_create_rule_validate_condition(condition, expected_message):
) )
err = ex.value.response["Error"] err = ex.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal(expected_message) assert err["Message"] == expected_message
@mock_elbv2 @mock_elbv2
@ -321,8 +320,8 @@ def test_describe_unknown_rule():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
conn.describe_rules(RuleArns=["unknown_arn"]) conn.describe_rules(RuleArns=["unknown_arn"])
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("RuleNotFound") assert err["Code"] == "RuleNotFound"
err["Message"].should.equal("One or more rules not found") assert err["Message"] == "One or more rules not found"
@mock_elbv2 @mock_elbv2
@ -403,17 +402,17 @@ def test_create_rule_action(action):
) )
# assert create_rule response # assert create_rule response
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Priority"].should.equal("100") assert rule["Priority"] == "100"
rule["Conditions"].should.equal([]) assert rule["Conditions"] == []
rule["Actions"].should.equal([action]) assert rule["Actions"] == [action]
# assert describe_rules response # assert describe_rules response
response = conn.describe_rules(ListenerArn=http_listener_arn) response = conn.describe_rules(ListenerArn=http_listener_arn)
response["Rules"].should.have.length_of(2) # including the default rule assert len(response["Rules"]) == 2 # including the default rule
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action
# assert set_rule_priorities response # assert set_rule_priorities response
rule_arn = response.get("Rules")[0]["RuleArn"] rule_arn = response.get("Rules")[0]["RuleArn"]
@ -421,11 +420,11 @@ def test_create_rule_action(action):
RulePriorities=[{"RuleArn": rule_arn, "Priority": 99}] RulePriorities=[{"RuleArn": rule_arn, "Priority": 99}]
) )
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Priority"].should.equal("99") assert rule["Priority"] == "99"
rule["Conditions"].should.equal([]) assert rule["Conditions"] == []
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action
@mock_elbv2 @mock_elbv2
@ -451,17 +450,17 @@ def test_create_rule_action_forward_config():
) )
# assert create_rule response # assert create_rule response
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Priority"].should.equal("100") assert rule["Priority"] == "100"
rule["Conditions"].should.equal([]) assert rule["Conditions"] == []
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action
# assert describe_rules response # assert describe_rules response
response = conn.describe_rules(ListenerArn=http_listener_arn) response = conn.describe_rules(ListenerArn=http_listener_arn)
response["Rules"].should.have.length_of(2) # including the default rule assert len(response["Rules"]) == 2 # including the default rule
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action
# assert set_rule_priorities response # assert set_rule_priorities response
rule_arn = response.get("Rules")[0]["RuleArn"] rule_arn = response.get("Rules")[0]["RuleArn"]
@ -469,11 +468,11 @@ def test_create_rule_action_forward_config():
RulePriorities=[{"RuleArn": rule_arn, "Priority": 99}] RulePriorities=[{"RuleArn": rule_arn, "Priority": 99}]
) )
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Priority"].should.equal("99") assert rule["Priority"] == "99"
rule["Conditions"].should.equal([]) assert rule["Conditions"] == []
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action
@mock_elbv2 @mock_elbv2
@ -495,17 +494,17 @@ def test_create_rule_action_forward_target_group():
) )
# assert create_rule response # assert create_rule response
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Priority"].should.equal("100") assert rule["Priority"] == "100"
rule["Conditions"].should.equal([]) assert rule["Conditions"] == []
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action
# assert describe_rules response # assert describe_rules response
response = conn.describe_rules(ListenerArn=http_listener_arn) response = conn.describe_rules(ListenerArn=http_listener_arn)
response["Rules"].should.have.length_of(2) # including the default rule assert len(response["Rules"]) == 2 # including the default rule
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action
# assert set_rule_priorities # assert set_rule_priorities
rule_arn = response.get("Rules")[0]["RuleArn"] rule_arn = response.get("Rules")[0]["RuleArn"]
@ -514,8 +513,8 @@ def test_create_rule_action_forward_target_group():
) )
# assert set_rule_priorities response # assert set_rule_priorities response
response["Rules"].should.have.length_of(1) assert len(response["Rules"]) == 1
rule = response.get("Rules")[0] rule = response.get("Rules")[0]
rule["Priority"].should.equal("99") assert rule["Priority"] == "99"
rule["Conditions"].should.equal([]) assert rule["Conditions"] == []
rule["Actions"][0].should.equal(action) assert rule["Actions"][0] == action

32
tests/test_elbv2/test_elbv2_listener_tags.py
View File

@ -1,5 +1,3 @@
import sure # noqa # pylint: disable=unused-import
from moto import mock_elbv2, mock_ec2 from moto import mock_elbv2, mock_ec2
from .test_elbv2 import create_load_balancer from .test_elbv2 import create_load_balancer
@ -23,7 +21,7 @@ def test_create_listener_with_tags():
# Ensure the tags persisted # Ensure the tags persisted
response = elbv2.describe_tags(ResourceArns=[listener_arn]) response = elbv2.describe_tags(ResourceArns=[listener_arn])
tags = {d["Key"]: d["Value"] for d in response["TagDescriptions"][0]["Tags"]} tags = {d["Key"]: d["Value"] for d in response["TagDescriptions"][0]["Tags"]}
tags.should.equal({"k1": "v1"}) assert tags == {"k1": "v1"}
@mock_elbv2 @mock_elbv2
@ -46,7 +44,7 @@ def test_listener_add_remove_tags():
tags = elbv2.describe_tags(ResourceArns=[listener_arn])["TagDescriptions"][0][ tags = elbv2.describe_tags(ResourceArns=[listener_arn])["TagDescriptions"][0][
"Tags" "Tags"
] ]
tags.should.equal([{"Key": "k1", "Value": "v1"}, {"Key": "a", "Value": "b"}]) assert tags == [{"Key": "k1", "Value": "v1"}, {"Key": "a", "Value": "b"}]
elbv2.add_tags( elbv2.add_tags(
ResourceArns=[listener_arn], ResourceArns=[listener_arn],
@ -60,24 +58,20 @@ def test_listener_add_remove_tags():
tags = elbv2.describe_tags(ResourceArns=[listener_arn])["TagDescriptions"][0][ tags = elbv2.describe_tags(ResourceArns=[listener_arn])["TagDescriptions"][0][
"Tags" "Tags"
] ]
tags.should.equal( assert tags == [
[ {"Key": "k1", "Value": "v1"},
{"Key": "k1", "Value": "v1"}, {"Key": "a", "Value": "b"},
{"Key": "a", "Value": "b"}, {"Key": "b", "Value": "b"},
{"Key": "b", "Value": "b"}, {"Key": "c", "Value": "b"},
{"Key": "c", "Value": "b"}, ]
]
)
elbv2.remove_tags(ResourceArns=[listener_arn], TagKeys=["a"]) elbv2.remove_tags(ResourceArns=[listener_arn], TagKeys=["a"])
tags = elbv2.describe_tags(ResourceArns=[listener_arn])["TagDescriptions"][0][ tags = elbv2.describe_tags(ResourceArns=[listener_arn])["TagDescriptions"][0][
"Tags" "Tags"
] ]
tags.should.equal( assert tags == [
[ {"Key": "k1", "Value": "v1"},
{"Key": "k1", "Value": "v1"}, {"Key": "b", "Value": "b"},
{"Key": "b", "Value": "b"}, {"Key": "c", "Value": "b"},
{"Key": "c", "Value": "b"}, ]
]
)

9
tests/test_elbv2/test_elbv2_set_subnets.py
View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_elbv2, mock_ec2 from moto import mock_elbv2, mock_ec2
@ -38,7 +37,7 @@ def test_set_subnets_errors():
) )
resp = client.describe_load_balancers(LoadBalancerArns=[arn]) resp = client.describe_load_balancers(LoadBalancerArns=[arn])
len(resp["LoadBalancers"][0]["AvailabilityZones"]).should.equal(3) assert len(resp["LoadBalancers"][0]["AvailabilityZones"]) == 3
@mock_elbv2 @mock_elbv2
@ -74,6 +73,6 @@ def test_set_subnets__mapping():
resp = client.describe_load_balancers(LoadBalancerArns=[arn]) resp = client.describe_load_balancers(LoadBalancerArns=[arn])
a_zones = resp["LoadBalancers"][0]["AvailabilityZones"] a_zones = resp["LoadBalancers"][0]["AvailabilityZones"]
a_zones.should.have.length_of(2) assert len(a_zones) == 2
a_zones.should.contain({"ZoneName": "us-east-1a", "SubnetId": subnet1.id}) assert {"ZoneName": "us-east-1a", "SubnetId": subnet1.id} in a_zones
a_zones.should.contain({"ZoneName": "us-east-1b", "SubnetId": subnet2.id}) assert {"ZoneName": "us-east-1b", "SubnetId": subnet2.id} in a_zones

252
tests/test_elbv2/test_elbv2_target_groups.py
View File

@ -1,7 +1,6 @@
import boto3 import boto3
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
import pytest import pytest
import sure # noqa # pylint: disable=unused-import
from moto import mock_elbv2, mock_ec2 from moto import mock_elbv2, mock_ec2
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -30,9 +29,10 @@ def test_create_target_group_with_invalid_healthcheck_protocol():
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"Value /HTTP at 'healthCheckProtocol' failed to satisfy constraint: Member must satisfy enum value set: ['HTTPS', 'HTTP', 'TCP', 'TLS', 'UDP', 'TCP_UDP', 'GENEVE']" err["Message"]
== "Value /HTTP at 'healthCheckProtocol' failed to satisfy constraint: Member must satisfy enum value set: ['HTTPS', 'HTTP', 'TCP', 'TLS', 'UDP', 'TCP_UDP', 'GENEVE']"
) )
@ -56,7 +56,7 @@ def test_create_target_group_with_tags():
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
Tags=[{"Key": "key1", "Value": "val1"}], Tags=[{"Key": "key1", "Value": "val1"}],
) )
target_group = response.get("TargetGroups")[0] target_group = response["TargetGroups"][0]
target_group_arn = target_group["TargetGroupArn"] target_group_arn = target_group["TargetGroupArn"]
# Add tags to the target group # Add tags to the target group
@ -64,15 +64,17 @@ def test_create_target_group_with_tags():
ResourceArns=[target_group_arn], ResourceArns=[target_group_arn],
Tags=[{"Key": "key2", "Value": "val2"}], Tags=[{"Key": "key2", "Value": "val2"}],
) )
conn.describe_tags(ResourceArns=[target_group_arn])["TagDescriptions"][0][ tags = conn.describe_tags(ResourceArns=[target_group_arn])["TagDescriptions"][0][
"Tags" "Tags"
].should.equal([{"Key": "key1", "Value": "val1"}, {"Key": "key2", "Value": "val2"}]) ]
assert tags == [{"Key": "key1", "Value": "val1"}, {"Key": "key2", "Value": "val2"}]
# Verify they can be removed # Verify they can be removed
conn.remove_tags(ResourceArns=[target_group_arn], TagKeys=["key1"]) conn.remove_tags(ResourceArns=[target_group_arn], TagKeys=["key1"])
conn.describe_tags(ResourceArns=[target_group_arn])["TagDescriptions"][0][ tags = conn.describe_tags(ResourceArns=[target_group_arn])["TagDescriptions"][0][
"Tags" "Tags"
].should.equal([{"Key": "key2", "Value": "val2"}]) ]
assert tags == [{"Key": "key2", "Value": "val2"}]
@mock_elbv2 @mock_elbv2
@ -80,7 +82,7 @@ def test_create_target_group_with_tags():
def test_create_target_group_and_listeners(): def test_create_target_group_and_listeners():
response, vpc, _, _, _, conn = create_load_balancer() response, vpc, _, _, _, conn = create_load_balancer()
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn") load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
response = conn.create_target_group( response = conn.create_target_group(
Name="a-target", Name="a-target",
@ -96,13 +98,13 @@ def test_create_target_group_and_listeners():
UnhealthyThresholdCount=2, UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
target_group = response.get("TargetGroups")[0] target_group = response["TargetGroups"][0]
target_group_arn = target_group.get("TargetGroupArn") target_group_arn = target_group["TargetGroupArn"]
target_group.get("HealthCheckProtocol").should.equal("HTTP") assert target_group["HealthCheckProtocol"] == "HTTP"
# Check it's in the describe_target_groups response # Check it's in the describe_target_groups response
response = conn.describe_target_groups() response = conn.describe_target_groups()
response.get("TargetGroups").should.have.length_of(1) assert len(response["TargetGroups"]) == 1
# Plain HTTP listener # Plain HTTP listener
response = conn.create_listener( response = conn.create_listener(
@ -111,18 +113,18 @@ def test_create_target_group_and_listeners():
Port=80, Port=80,
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}], DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn}],
) )
listener = response.get("Listeners")[0] listener = response["Listeners"][0]
listener.get("Port").should.equal(80) assert listener["Port"] == 80
listener.get("Protocol").should.equal("HTTP") assert listener["Protocol"] == "HTTP"
listener.get("DefaultActions").should.equal( assert listener["DefaultActions"] == [
[{"TargetGroupArn": target_group_arn, "Type": "forward"}] {"TargetGroupArn": target_group_arn, "Type": "forward"}
) ]
http_listener_arn = listener.get("ListenerArn") http_listener_arn = listener["ListenerArn"]
response = conn.describe_target_groups( response = conn.describe_target_groups(
LoadBalancerArn=load_balancer_arn, Names=["a-target"] LoadBalancerArn=load_balancer_arn, Names=["a-target"]
) )
response.get("TargetGroups").should.have.length_of(1) assert len(response["TargetGroups"]) == 1
# And another with SSL # And another with SSL
actions = {"Type": "forward", "TargetGroupArn": target_group_arn} actions = {"Type": "forward", "TargetGroupArn": target_group_arn}
@ -135,30 +137,30 @@ def test_create_target_group_and_listeners():
], ],
DefaultActions=[actions], DefaultActions=[actions],
) )
listener = response.get("Listeners")[0] listener = response["Listeners"][0]
listener.get("Port").should.equal(443) assert listener["Port"] == 443
listener.get("Protocol").should.equal("HTTPS") assert listener["Protocol"] == "HTTPS"
listener.get("Certificates").should.equal( assert listener["Certificates"] == [
[{"CertificateArn": f"arn:aws:iam:{ACCOUNT_ID}:server-certificate/test-cert"}] {"CertificateArn": f"arn:aws:iam:{ACCOUNT_ID}:server-certificate/test-cert"}
) ]
listener.get("DefaultActions").should.equal( assert listener["DefaultActions"] == [
[{"TargetGroupArn": target_group_arn, "Type": "forward"}] {"TargetGroupArn": target_group_arn, "Type": "forward"}
) ]
https_listener_arn = listener.get("ListenerArn") https_listener_arn = listener["ListenerArn"]
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn) response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
response.get("Listeners").should.have.length_of(2) assert len(response["Listeners"]) == 2
response = conn.describe_listeners(ListenerArns=[https_listener_arn]) response = conn.describe_listeners(ListenerArns=[https_listener_arn])
response.get("Listeners").should.have.length_of(1) assert len(response["Listeners"]) == 1
listener = response.get("Listeners")[0] listener = response["Listeners"][0]
listener.get("Port").should.equal(443) assert listener["Port"] == 443
listener.get("Protocol").should.equal("HTTPS") assert listener["Protocol"] == "HTTPS"
response = conn.describe_listeners( response = conn.describe_listeners(
ListenerArns=[http_listener_arn, https_listener_arn] ListenerArns=[http_listener_arn, https_listener_arn]
) )
response.get("Listeners").should.have.length_of(2) assert len(response["Listeners"]) == 2
conn.create_rule( conn.create_rule(
ListenerArn=http_listener_arn, ListenerArn=http_listener_arn,
@ -170,40 +172,39 @@ def test_create_target_group_and_listeners():
# listener referencing it # listener referencing it
with pytest.raises(ClientError) as e: with pytest.raises(ClientError) as e:
conn.delete_target_group(TargetGroupArn=target_group_arn) conn.delete_target_group(TargetGroupArn=target_group_arn)
e.value.operation_name.should.equal("DeleteTargetGroup") assert e.value.operation_name == "DeleteTargetGroup"
e.value.args.should.equal( assert (
( e.value.response["Error"]["Message"]
f"An error occurred (ResourceInUse) when calling the DeleteTargetGroup operation: The target group 'arn:aws:elasticloadbalancing:us-east-1:{ACCOUNT_ID}:targetgroup/a-target/50dc6c495c0c9188' is currently in use by a listener or a rule", == f"The target group 'arn:aws:elasticloadbalancing:us-east-1:{ACCOUNT_ID}:targetgroup/a-target/50dc6c495c0c9188' is currently in use by a listener or a rule"
) )
) # NOQA
# Delete one listener # Delete one listener
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn) response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
response.get("Listeners").should.have.length_of(2) assert len(response["Listeners"]) == 2
conn.delete_listener(ListenerArn=http_listener_arn) conn.delete_listener(ListenerArn=http_listener_arn)
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn) response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
response.get("Listeners").should.have.length_of(1) assert len(response["Listeners"]) == 1
# Then delete the load balancer # Then delete the load balancer
conn.delete_load_balancer(LoadBalancerArn=load_balancer_arn) conn.delete_load_balancer(LoadBalancerArn=load_balancer_arn)
# It's gone # It's gone
response = conn.describe_load_balancers() response = conn.describe_load_balancers()
response.get("LoadBalancers").should.have.length_of(0) assert len(response["LoadBalancers"]) == 0
# And it deleted the remaining listener # And it deleted the remaining listener
with pytest.raises(ClientError) as e: with pytest.raises(ClientError) as e:
conn.describe_listeners(ListenerArns=[http_listener_arn, https_listener_arn]) conn.describe_listeners(ListenerArns=[http_listener_arn, https_listener_arn])
e.value.response["Error"]["Code"].should.equal("ListenerNotFound") assert e.value.response["Error"]["Code"] == "ListenerNotFound"
# But not the target groups # But not the target groups
response = conn.describe_target_groups() response = conn.describe_target_groups()
response.get("TargetGroups").should.have.length_of(1) assert len(response["TargetGroups"]) == 1
# Which we'll now delete # Which we'll now delete
conn.delete_target_group(TargetGroupArn=target_group_arn) conn.delete_target_group(TargetGroupArn=target_group_arn)
response = conn.describe_target_groups() response = conn.describe_target_groups()
response.get("TargetGroups").should.have.length_of(0) assert len(response["TargetGroups"]) == 0
@mock_elbv2 @mock_elbv2
@ -221,7 +222,7 @@ def test_create_target_group_without_non_required_parameters():
HealthCheckProtocol="HTTP", HealthCheckProtocol="HTTP",
HealthCheckPort="8080", HealthCheckPort="8080",
) )
response.get("TargetGroups", []).should.have.length_of(1) assert len(response.get("TargetGroups", [])) == 1
@mock_elbv2 @mock_elbv2
@ -250,9 +251,10 @@ def test_create_invalid_target_group_long_name():
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
"Target group name 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA' cannot be longer than '32' characters" err["Message"]
== "Target group name 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA' cannot be longer than '32' characters"
) )
@ -281,9 +283,10 @@ def test_create_invalid_target_group_invalid_characters(name):
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
f"1 validation error detected: Value '{name}' at 'targetGroup.targetGroupArn.targetGroupName' failed to satisfy constraint: Member must satisfy regular expression pattern: (?!.*--)(?!^-)(?!.*-$)^[A-Za-z0-9-]+$" err["Message"]
== f"1 validation error detected: Value '{name}' at 'targetGroup.targetGroupArn.targetGroupName' failed to satisfy constraint: Member must satisfy regular expression pattern: (?!.*--)(?!^-)(?!.*-$)^[A-Za-z0-9-]+$"
) )
@ -312,9 +315,10 @@ def test_create_invalid_target_group_alphanumeric_characters(name):
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.equal( assert (
f"Target group name '{name}' can only contain characters that are alphanumeric characters or hyphens(-)" err["Message"]
== f"Target group name '{name}' can only contain characters that are alphanumeric characters or hyphens(-)"
) )
@ -362,26 +366,26 @@ def test_target_group_attributes():
UnhealthyThresholdCount=2, UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
target_group = response.get("TargetGroups")[0] target_group = response["TargetGroups"][0]
# Check it's in the describe_target_groups response # Check it's in the describe_target_groups response
response = conn.describe_target_groups() response = conn.describe_target_groups()
response.get("TargetGroups").should.have.length_of(1) assert len(response["TargetGroups"]) == 1
target_group_arn = target_group["TargetGroupArn"] target_group_arn = target_group["TargetGroupArn"]
# check if Names filter works # check if Names filter works
response = conn.describe_target_groups(Names=[]) response = conn.describe_target_groups(Names=[])
response = conn.describe_target_groups(Names=["a-target"]) response = conn.describe_target_groups(Names=["a-target"])
response.get("TargetGroups").should.have.length_of(1) assert len(response["TargetGroups"]) == 1
target_group_arn = target_group["TargetGroupArn"] target_group_arn = target_group["TargetGroupArn"]
# The attributes should start with the two defaults # The attributes should start with the two defaults
response = conn.describe_target_group_attributes(TargetGroupArn=target_group_arn) response = conn.describe_target_group_attributes(TargetGroupArn=target_group_arn)
response["Attributes"].should.have.length_of(5) assert len(response["Attributes"]) == 5
attributes = {attr["Key"]: attr["Value"] for attr in response["Attributes"]} attributes = {attr["Key"]: attr["Value"] for attr in response["Attributes"]}
attributes["deregistration_delay.timeout_seconds"].should.equal("300") assert attributes["deregistration_delay.timeout_seconds"] == "300"
attributes["stickiness.enabled"].should.equal("false") assert attributes["stickiness.enabled"] == "false"
attributes["waf.fail_open.enabled"].should.equal("false") assert attributes["waf.fail_open.enabled"] == "false"
# Add cookie stickiness # Add cookie stickiness
response = conn.modify_target_group_attributes( response = conn.modify_target_group_attributes(
@ -393,17 +397,17 @@ def test_target_group_attributes():
) )
# The response should have only the keys updated # The response should have only the keys updated
response["Attributes"].should.have.length_of(2) assert len(response["Attributes"]) == 2
attributes = {attr["Key"]: attr["Value"] for attr in response["Attributes"]} attributes = {attr["Key"]: attr["Value"] for attr in response["Attributes"]}
attributes["stickiness.type"].should.equal("lb_cookie") assert attributes["stickiness.type"] == "lb_cookie"
attributes["stickiness.enabled"].should.equal("true") assert attributes["stickiness.enabled"] == "true"
# These new values should be in the full attribute list # These new values should be in the full attribute list
response = conn.describe_target_group_attributes(TargetGroupArn=target_group_arn) response = conn.describe_target_group_attributes(TargetGroupArn=target_group_arn)
response["Attributes"].should.have.length_of(6) assert len(response["Attributes"]) == 6
attributes = {attr["Key"]: attr["Value"] for attr in response["Attributes"]} attributes = {attr["Key"]: attr["Value"] for attr in response["Attributes"]}
attributes["stickiness.type"].should.equal("lb_cookie") assert attributes["stickiness.type"] == "lb_cookie"
attributes["stickiness.enabled"].should.equal("true") assert attributes["stickiness.enabled"] == "true"
@mock_elbv2 @mock_elbv2
@ -431,9 +435,10 @@ def test_create_target_group_invalid_protocol():
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
err = ex.value.response["Error"] err = ex.value.response["Error"]
err["Code"].should.equal("ValidationError") assert err["Code"] == "ValidationError"
err["Message"].should.contain( assert (
"Value /HTTP at 'healthCheckProtocol' failed to satisfy constraint" "Value /HTTP at 'healthCheckProtocol' failed to satisfy constraint"
in err["Message"]
) )
@ -445,8 +450,8 @@ def test_describe_invalid_target_group():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
conn.describe_target_groups(Names=["invalid"]) conn.describe_target_groups(Names=["invalid"])
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("TargetGroupNotFound") assert err["Code"] == "TargetGroupNotFound"
err["Message"].should.equal("The specified target group does not exist.") assert err["Message"] == "The specified target group does not exist."
@mock_elbv2 @mock_elbv2
@ -454,7 +459,7 @@ def test_describe_invalid_target_group():
def test_describe_target_groups_no_arguments(): def test_describe_target_groups_no_arguments():
response, vpc, _, _, _, conn = create_load_balancer() response, vpc, _, _, _, conn = create_load_balancer()
response.get("LoadBalancers")[0].get("LoadBalancerArn") assert "LoadBalancerArn" in response["LoadBalancers"][0]
conn.create_target_group( conn.create_target_group(
Name="a-target", Name="a-target",
@ -472,8 +477,8 @@ def test_describe_target_groups_no_arguments():
) )
groups = conn.describe_target_groups()["TargetGroups"] groups = conn.describe_target_groups()["TargetGroups"]
groups.should.have.length_of(1) assert len(groups) == 1
groups[0].should.have.key("Matcher").equals({"HttpCode": "201"}) assert groups[0]["Matcher"] == {"HttpCode": "201"}
@mock_elbv2 @mock_elbv2
@ -498,7 +503,7 @@ def test_modify_target_group():
UnhealthyThresholdCount=2, UnhealthyThresholdCount=2,
Matcher={"HttpCode": "200"}, Matcher={"HttpCode": "200"},
) )
arn = response.get("TargetGroups")[0]["TargetGroupArn"] arn = response["TargetGroups"][0]["TargetGroupArn"]
client.modify_target_group( client.modify_target_group(
TargetGroupArn=arn, TargetGroupArn=arn,
@ -513,16 +518,16 @@ def test_modify_target_group():
) )
response = client.describe_target_groups(TargetGroupArns=[arn]) response = client.describe_target_groups(TargetGroupArns=[arn])
response["TargetGroups"][0]["Matcher"]["HttpCode"].should.equal("200-399") assert response["TargetGroups"][0]["Matcher"]["HttpCode"] == "200-399"
response["TargetGroups"][0]["HealthCheckIntervalSeconds"].should.equal(10) assert response["TargetGroups"][0]["HealthCheckIntervalSeconds"] == 10
response["TargetGroups"][0]["HealthCheckPath"].should.equal("/status") assert response["TargetGroups"][0]["HealthCheckPath"] == "/status"
response["TargetGroups"][0]["HealthCheckPort"].should.equal("8081") assert response["TargetGroups"][0]["HealthCheckPort"] == "8081"
response["TargetGroups"][0]["HealthCheckProtocol"].should.equal("HTTPS") assert response["TargetGroups"][0]["HealthCheckProtocol"] == "HTTPS"
response["TargetGroups"][0]["HealthCheckTimeoutSeconds"].should.equal(10) assert response["TargetGroups"][0]["HealthCheckTimeoutSeconds"] == 10
response["TargetGroups"][0]["HealthyThresholdCount"].should.equal(10) assert response["TargetGroups"][0]["HealthyThresholdCount"] == 10
response["TargetGroups"][0].should.have.key("Protocol").equals("HTTP") assert response["TargetGroups"][0]["Protocol"] == "HTTP"
response["TargetGroups"][0].should.have.key("ProtocolVersion").equals("HTTP1") assert response["TargetGroups"][0]["ProtocolVersion"] == "HTTP1"
response["TargetGroups"][0]["UnhealthyThresholdCount"].should.equal(4) assert response["TargetGroups"][0]["UnhealthyThresholdCount"] == 4
@mock_elbv2 @mock_elbv2
@ -534,18 +539,18 @@ def test_create_target_group_with_target_type(target_type):
response = conn.create_target_group(Name="a-target", TargetType=target_type) response = conn.create_target_group(Name="a-target", TargetType=target_type)
group = response["TargetGroups"][0] group = response["TargetGroups"][0]
group.should.have.key("TargetGroupArn") assert "TargetGroupArn" in group
group.should.have.key("TargetGroupName").equal("a-target") assert group["TargetGroupName"] == "a-target"
group.should.have.key("TargetType").equal(target_type) assert group["TargetType"] == target_type
group.shouldnt.have.key("Protocol") assert "Protocol" not in group
group.shouldnt.have.key("VpcId") assert "VpcId" not in group
group = conn.describe_target_groups()["TargetGroups"][0] group = conn.describe_target_groups()["TargetGroups"][0]
group.should.have.key("TargetGroupArn") assert "TargetGroupArn" in group
group.should.have.key("TargetGroupName").equal("a-target") assert group["TargetGroupName"] == "a-target"
group.should.have.key("TargetType").equal(target_type) assert group["TargetType"] == target_type
group.shouldnt.have.key("Protocol") assert "Protocol" not in group
group.shouldnt.have.key("VpcId") assert "VpcId" not in group
@mock_elbv2 @mock_elbv2
@ -555,17 +560,17 @@ def test_delete_target_group_after_modifying_listener():
response, vpc, _, _, _, conn = create_load_balancer() response, vpc, _, _, _, conn = create_load_balancer()
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn") load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
response = client.create_target_group( response = client.create_target_group(
Name="a-target", Protocol="HTTP", Port=8080, VpcId=vpc.id Name="a-target", Protocol="HTTP", Port=8080, VpcId=vpc.id
) )
target_group_arn1 = response.get("TargetGroups")[0]["TargetGroupArn"] target_group_arn1 = response["TargetGroups"][0]["TargetGroupArn"]
response = client.create_target_group( response = client.create_target_group(
Name="a-target-2", Protocol="HTTPS", Port=8081, VpcId=vpc.id Name="a-target-2", Protocol="HTTPS", Port=8081, VpcId=vpc.id
) )
target_group_arn2 = response.get("TargetGroups")[0]["TargetGroupArn"] target_group_arn2 = response["TargetGroups"][0]["TargetGroupArn"]
response = conn.create_listener( response = conn.create_listener(
LoadBalancerArn=load_balancer_arn, LoadBalancerArn=load_balancer_arn,
@ -573,7 +578,7 @@ def test_delete_target_group_after_modifying_listener():
Port=80, Port=80,
DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn1}], DefaultActions=[{"Type": "forward", "TargetGroupArn": target_group_arn1}],
) )
listener_arn = response["Listeners"][0].get("ListenerArn") listener_arn = response["Listeners"][0]["ListenerArn"]
client.modify_listener( client.modify_listener(
ListenerArn=listener_arn, ListenerArn=listener_arn,
@ -582,9 +587,7 @@ def test_delete_target_group_after_modifying_listener():
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn) response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
default_actions = response["Listeners"][0]["DefaultActions"] default_actions = response["Listeners"][0]["DefaultActions"]
default_actions.should.equal( assert default_actions == [{"Type": "forward", "TargetGroupArn": target_group_arn2}]
[{"Type": "forward", "TargetGroupArn": target_group_arn2}]
)
# Target Group 1 can now be deleted, as the LB points to group 2 # Target Group 1 can now be deleted, as the LB points to group 2
client.delete_target_group(TargetGroupArn=target_group_arn1) client.delete_target_group(TargetGroupArn=target_group_arn1)
@ -592,9 +595,7 @@ def test_delete_target_group_after_modifying_listener():
# Sanity check - we're still pointing to group 2 # Sanity check - we're still pointing to group 2
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn) response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
default_actions = response["Listeners"][0]["DefaultActions"] default_actions = response["Listeners"][0]["DefaultActions"]
default_actions.should.equal( assert default_actions == [{"Type": "forward", "TargetGroupArn": target_group_arn2}]
[{"Type": "forward", "TargetGroupArn": target_group_arn2}]
)
@mock_elbv2 @mock_elbv2
@ -604,17 +605,17 @@ def test_create_listener_with_multiple_target_groups():
response, vpc, _, _, _, conn = create_load_balancer() response, vpc, _, _, _, conn = create_load_balancer()
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn") load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
response = client.create_target_group( response = client.create_target_group(
Name="a-target", Protocol="HTTP", Port=8080, VpcId=vpc.id Name="a-target", Protocol="HTTP", Port=8080, VpcId=vpc.id
) )
target_group_arn1 = response.get("TargetGroups")[0]["TargetGroupArn"] target_group_arn1 = response["TargetGroups"][0]["TargetGroupArn"]
response = client.create_target_group( response = client.create_target_group(
Name="a-target-2", Protocol="HTTPS", Port=8081, VpcId=vpc.id Name="a-target-2", Protocol="HTTPS", Port=8081, VpcId=vpc.id
) )
target_group_arn2 = response.get("TargetGroups")[0]["TargetGroupArn"] target_group_arn2 = response["TargetGroups"][0]["TargetGroupArn"]
conn.create_listener( conn.create_listener(
LoadBalancerArn=load_balancer_arn, LoadBalancerArn=load_balancer_arn,
@ -640,9 +641,9 @@ def test_create_listener_with_multiple_target_groups():
response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn) response = conn.describe_listeners(LoadBalancerArn=load_balancer_arn)
listener = response["Listeners"][0] listener = response["Listeners"][0]
groups = listener["DefaultActions"][0]["ForwardConfig"]["TargetGroups"] groups = listener["DefaultActions"][0]["ForwardConfig"]["TargetGroups"]
groups.should.have.length_of(2) assert len(groups) == 2
groups.should.contain({"TargetGroupArn": target_group_arn1, "Weight": 100}) assert {"TargetGroupArn": target_group_arn1, "Weight": 100} in groups
groups.should.contain({"TargetGroupArn": target_group_arn2, "Weight": 0}) assert {"TargetGroupArn": target_group_arn2, "Weight": 0} in groups
@mock_elbv2 @mock_elbv2
@ -650,7 +651,7 @@ def test_create_listener_with_multiple_target_groups():
def test_create_listener_with_invalid_target_group(): def test_create_listener_with_invalid_target_group():
response, _, _, _, _, conn = create_load_balancer() response, _, _, _, _, conn = create_load_balancer()
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn") load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
conn.create_listener( conn.create_listener(
@ -667,8 +668,8 @@ def test_create_listener_with_invalid_target_group():
], ],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("TargetGroupNotFound") assert err["Code"] == "TargetGroupNotFound"
err["Message"].should.equal("Target group 'unknown' not found") assert err["Message"] == "Target group 'unknown' not found"
@mock_elbv2 @mock_elbv2
@ -678,12 +679,12 @@ def test_delete_target_group_while_listener_still_exists():
response, vpc, _, _, _, conn = create_load_balancer() response, vpc, _, _, _, conn = create_load_balancer()
load_balancer_arn = response.get("LoadBalancers")[0].get("LoadBalancerArn") load_balancer_arn = response["LoadBalancers"][0]["LoadBalancerArn"]
response = client.create_target_group( response = client.create_target_group(
Name="a-target", Protocol="HTTP", Port=8080, VpcId=vpc.id Name="a-target", Protocol="HTTP", Port=8080, VpcId=vpc.id
) )
target_group_arn1 = response.get("TargetGroups")[0]["TargetGroupArn"] target_group_arn1 = response["TargetGroups"][0]["TargetGroupArn"]
response = conn.create_listener( response = conn.create_listener(
LoadBalancerArn=load_balancer_arn, LoadBalancerArn=load_balancer_arn,
@ -706,9 +707,10 @@ def test_delete_target_group_while_listener_still_exists():
with pytest.raises(ClientError) as exc: with pytest.raises(ClientError) as exc:
client.delete_target_group(TargetGroupArn=target_group_arn1) client.delete_target_group(TargetGroupArn=target_group_arn1)
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("ResourceInUse") assert err["Code"] == "ResourceInUse"
err["Message"].should.equal( assert (
f"The target group '{target_group_arn1}' is currently in use by a listener or a rule" err["Message"]
== f"The target group '{target_group_arn1}' is currently in use by a listener or a rule"
) )
client.delete_listener(ListenerArn=listener_arn) client.delete_listener(ListenerArn=listener_arn)

4
tests/test_elbv2/test_server.py
View File

@ -1,5 +1,3 @@
import sure # noqa # pylint: disable=unused-import
import moto.server as server import moto.server as server
""" """
@ -13,4 +11,4 @@ def test_elbv2_describe_load_balancers():
res = test_client.get("/?Action=DescribeLoadBalancers&Version=2015-12-01") res = test_client.get("/?Action=DescribeLoadBalancers&Version=2015-12-01")
res.data.should.contain(b"DescribeLoadBalancersResponse") assert b"DescribeLoadBalancersResponse" in res.data