Skip to content

Commit d05a4af

Browse files
hussein-awalapotiuk
hussein-awala
authored and
potiuk
committed
stop using moto cloud_formation mock
Signed-off-by: Hussein Awala <[email protected]>
1 parent d8b000a commit d05a4af

File tree

1 file changed

+116
-29
lines changed

1 file changed

+116
-29
lines changed

‎tests/providers/amazon/aws/hooks/test_cloud_formation.py

Lines changed: 116 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,89 @@
1818
from __future__ import annotations
1919

2020
import json
21-
22-
from moto import mock_cloudformation
21+
from unittest import mock
2322

2423
from airflow.providers.amazon.aws.hooks.cloud_formation import CloudFormationHook
2524

25+
# TODO: restore this test with moto when openapi_schema_validator 0.6.0 is released
26+
# class TestCloudFormationHook:
27+
# from moto import mock_cloudformation
28+
29+
# def setup_method(self):
30+
# self.hook = CloudFormationHook(aws_conn_id="aws_default")
31+
#
32+
# def create_stack(self, stack_name):
33+
# timeout = 15
34+
# template_body = json.dumps(
35+
# {
36+
# "Resources": {
37+
# "myResource": {
38+
# "Type": "AWS::EC2::VPC",
39+
# "Properties": {
40+
# "CidrBlock": {"Ref": "VPCCidr"},
41+
# "Tags": [{"Key": "Name", "Value": "Primary_CF_VPC"}],
42+
# },
43+
# }
44+
# },
45+
# "Parameters": {
46+
# "VPCCidr": {
47+
# "Type": "String",
48+
# "Default": "10.0.0.0/16",
49+
# "Description": "Enter the CIDR block for the VPC. Default is 10.0.0.0/16.",
50+
# }
51+
# },
52+
# }
53+
# )
54+
#
55+
# self.hook.create_stack(
56+
# stack_name=stack_name,
57+
# cloudformation_parameters={
58+
# "TimeoutInMinutes": timeout,
59+
# "TemplateBody": template_body,
60+
# "Parameters": [{"ParameterKey": "VPCCidr", "ParameterValue": "10.0.0.0/16"}],
61+
# },
62+
# )
63+
#
64+
# @mock_cloudformation
65+
# def test_get_conn_returns_a_boto3_connection(self):
66+
# assert self.hook.get_conn().describe_stacks() is not None
67+
#
68+
# @mock_cloudformation
69+
# def test_get_stack_status(self):
70+
# stack_name = "my_test_get_stack_status_stack"
71+
#
72+
# stack_status = self.hook.get_stack_status(stack_name=stack_name)
73+
# assert stack_status is None
74+
#
75+
# self.create_stack(stack_name)
76+
# stack_status = self.hook.get_stack_status(stack_name=stack_name)
77+
# assert stack_status == "CREATE_COMPLETE", "Incorrect stack status returned."
78+
#
79+
# @mock_cloudformation
80+
# def test_create_stack(self):
81+
# stack_name = "my_test_create_stack_stack"
82+
# self.create_stack(stack_name)
83+
#
84+
# stacks = self.hook.get_conn().describe_stacks()["Stacks"]
85+
# assert len(stacks) > 0, "CloudFormation should have stacks"
86+
#
87+
# matching_stacks = [x for x in stacks if x["StackName"] == stack_name]
88+
# assert len(matching_stacks) == 1, f"stack with name {stack_name} should exist"
89+
#
90+
# stack = matching_stacks[0]
91+
# assert stack["StackStatus"] == "CREATE_COMPLETE", "Stack should be in status CREATE_COMPLETE"
92+
#
93+
# @mock_cloudformation
94+
# def test_delete_stack(self):
95+
# stack_name = "my_test_delete_stack_stack"
96+
# self.create_stack(stack_name)
97+
#
98+
# self.hook.delete_stack(stack_name=stack_name)
99+
#
100+
# stacks = self.hook.get_conn().describe_stacks()["Stacks"]
101+
# matching_stacks = [x for x in stacks if x["StackName"] == stack_name]
102+
# assert not matching_stacks, f"stack with name {stack_name} should not exist"
103+
26104

27105
class TestCloudFormationHook:
28106
def setup_method(self):
@@ -60,42 +138,51 @@ def create_stack(self, stack_name):
60138
},
61139
)
62140

63-
@mock_cloudformation
64-
def test_get_conn_returns_a_boto3_connection(self):
141+
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
142+
def test_get_conn_returns_a_boto3_connection(self, cloudformation_conn_mock):
65143
assert self.hook.get_conn().describe_stacks() is not None
66144

67-
@mock_cloudformation
68-
def test_get_stack_status(self):
145+
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
146+
def test_get_stack_status(self, cloudformation_conn_mock):
69147
stack_name = "my_test_get_stack_status_stack"
70148

71-
stack_status = self.hook.get_stack_status(stack_name=stack_name)
72-
assert stack_status is None
73-
74-
self.create_stack(stack_name)
75-
stack_status = self.hook.get_stack_status(stack_name=stack_name)
76-
assert stack_status == "CREATE_COMPLETE", "Incorrect stack status returned."
149+
self.hook.get_stack_status(stack_name=stack_name)
150+
cloudformation_conn_mock.return_value.describe_stacks.assert_called_once_with(StackName=stack_name)
77151

78-
@mock_cloudformation
79-
def test_create_stack(self):
152+
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
153+
def test_create_stack(self, cloudformation_conn_mock):
80154
stack_name = "my_test_create_stack_stack"
81155
self.create_stack(stack_name)
156+
cloudformation_conn_mock.return_value.create_stack.assert_called_once_with(
157+
StackName=stack_name,
158+
TimeoutInMinutes=15,
159+
TemplateBody=json.dumps(
160+
{
161+
"Resources": {
162+
"myResource": {
163+
"Type": "AWS::EC2::VPC",
164+
"Properties": {
165+
"CidrBlock": {"Ref": "VPCCidr"},
166+
"Tags": [{"Key": "Name", "Value": "Primary_CF_VPC"}],
167+
},
168+
}
169+
},
170+
"Parameters": {
171+
"VPCCidr": {
172+
"Type": "String",
173+
"Default": "10.0.0.0/16",
174+
"Description": "Enter the CIDR block for the VPC. Default is 10.0.0.0/16.",
175+
}
176+
},
177+
}
178+
),
179+
Parameters=[{"ParameterKey": "VPCCidr", "ParameterValue": "10.0.0.0/16"}],
180+
)
82181

83-
stacks = self.hook.get_conn().describe_stacks()["Stacks"]
84-
assert len(stacks) > 0, "CloudFormation should have stacks"
85-
86-
matching_stacks = [x for x in stacks if x["StackName"] == stack_name]
87-
assert len(matching_stacks) == 1, f"stack with name {stack_name} should exist"
88-
89-
stack = matching_stacks[0]
90-
assert stack["StackStatus"] == "CREATE_COMPLETE", "Stack should be in status CREATE_COMPLETE"
91-
92-
@mock_cloudformation
93-
def test_delete_stack(self):
182+
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
183+
def test_delete_stack(self, cloudformation_conn_mock):
94184
stack_name = "my_test_delete_stack_stack"
95-
self.create_stack(stack_name)
96185

97186
self.hook.delete_stack(stack_name=stack_name)
98187

99-
stacks = self.hook.get_conn().describe_stacks()["Stacks"]
100-
matching_stacks = [x for x in stacks if x["StackName"] == stack_name]
101-
assert not matching_stacks, f"stack with name {stack_name} should not exist"
188+
cloudformation_conn_mock.return_value.delete_stack.assert_called_once_with(StackName=stack_name)

0 commit comments

Comments
 (0)