Skip to content

Commit

Permalink
stop using moto cloud_formation mock
Browse files Browse the repository at this point in the history 
Signed-off-by: Hussein Awala <hussein@awala.fr>
  • Loading branch information
@hussein-awala @potiuk
hussein-awala authored and potiuk committed Jul 10, 2023
1 parent d8b000a commit d05a4af
Showing 1 changed file with 116 additions and 29 deletions.
145 changes: 116 additions & 29 deletions tests/providers/amazon/aws/hooks/test_cloud_formation.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,89 @@
from __future__ import annotations

import json

from moto import mock_cloudformation
from unittest import mock

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

# TODO: restore this test with moto when openapi_schema_validator 0.6.0 is released
# class TestCloudFormationHook:
# from moto import mock_cloudformation

# def setup_method(self):
# self.hook = CloudFormationHook(aws_conn_id="aws_default")
#
# def create_stack(self, stack_name):
# timeout = 15
# template_body = json.dumps(
# {
# "Resources": {
# "myResource": {
# "Type": "AWS::EC2::VPC",
# "Properties": {
# "CidrBlock": {"Ref": "VPCCidr"},
# "Tags": [{"Key": "Name", "Value": "Primary_CF_VPC"}],
# },
# }
# },
# "Parameters": {
# "VPCCidr": {
# "Type": "String",
# "Default": "10.0.0.0/16",
# "Description": "Enter the CIDR block for the VPC. Default is 10.0.0.0/16.",
# }
# },
# }
# )
#
# self.hook.create_stack(
# stack_name=stack_name,
# cloudformation_parameters={
# "TimeoutInMinutes": timeout,
# "TemplateBody": template_body,
# "Parameters": [{"ParameterKey": "VPCCidr", "ParameterValue": "10.0.0.0/16"}],
# },
# )
#
# @mock_cloudformation
# def test_get_conn_returns_a_boto3_connection(self):
# assert self.hook.get_conn().describe_stacks() is not None
#
# @mock_cloudformation
# def test_get_stack_status(self):
# stack_name = "my_test_get_stack_status_stack"
#
# stack_status = self.hook.get_stack_status(stack_name=stack_name)
# assert stack_status is None
#
# self.create_stack(stack_name)
# stack_status = self.hook.get_stack_status(stack_name=stack_name)
# assert stack_status == "CREATE_COMPLETE", "Incorrect stack status returned."
#
# @mock_cloudformation
# def test_create_stack(self):
# stack_name = "my_test_create_stack_stack"
# self.create_stack(stack_name)
#
# stacks = self.hook.get_conn().describe_stacks()["Stacks"]
# assert len(stacks) > 0, "CloudFormation should have stacks"
#
# matching_stacks = [x for x in stacks if x["StackName"] == stack_name]
# assert len(matching_stacks) == 1, f"stack with name {stack_name} should exist"
#
# stack = matching_stacks[0]
# assert stack["StackStatus"] == "CREATE_COMPLETE", "Stack should be in status CREATE_COMPLETE"
#
# @mock_cloudformation
# def test_delete_stack(self):
# stack_name = "my_test_delete_stack_stack"
# self.create_stack(stack_name)
#
# self.hook.delete_stack(stack_name=stack_name)
#
# stacks = self.hook.get_conn().describe_stacks()["Stacks"]
# matching_stacks = [x for x in stacks if x["StackName"] == stack_name]
# assert not matching_stacks, f"stack with name {stack_name} should not exist"


class TestCloudFormationHook:
def setup_method(self):
Expand Down Expand Up @@ -60,42 +138,51 @@ def create_stack(self, stack_name):
},
)

@mock_cloudformation
def test_get_conn_returns_a_boto3_connection(self):
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
def test_get_conn_returns_a_boto3_connection(self, cloudformation_conn_mock):
assert self.hook.get_conn().describe_stacks() is not None

@mock_cloudformation
def test_get_stack_status(self):
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
def test_get_stack_status(self, cloudformation_conn_mock):
stack_name = "my_test_get_stack_status_stack"

stack_status = self.hook.get_stack_status(stack_name=stack_name)
assert stack_status is None

self.create_stack(stack_name)
stack_status = self.hook.get_stack_status(stack_name=stack_name)
assert stack_status == "CREATE_COMPLETE", "Incorrect stack status returned."
self.hook.get_stack_status(stack_name=stack_name)
cloudformation_conn_mock.return_value.describe_stacks.assert_called_once_with(StackName=stack_name)

@mock_cloudformation
def test_create_stack(self):
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
def test_create_stack(self, cloudformation_conn_mock):
stack_name = "my_test_create_stack_stack"
self.create_stack(stack_name)
cloudformation_conn_mock.return_value.create_stack.assert_called_once_with(
StackName=stack_name,
TimeoutInMinutes=15,
TemplateBody=json.dumps(
{
"Resources": {
"myResource": {
"Type": "AWS::EC2::VPC",
"Properties": {
"CidrBlock": {"Ref": "VPCCidr"},
"Tags": [{"Key": "Name", "Value": "Primary_CF_VPC"}],
},
}
},
"Parameters": {
"VPCCidr": {
"Type": "String",
"Default": "10.0.0.0/16",
"Description": "Enter the CIDR block for the VPC. Default is 10.0.0.0/16.",
}
},
}
),
Parameters=[{"ParameterKey": "VPCCidr", "ParameterValue": "10.0.0.0/16"}],
)

stacks = self.hook.get_conn().describe_stacks()["Stacks"]
assert len(stacks) > 0, "CloudFormation should have stacks"

matching_stacks = [x for x in stacks if x["StackName"] == stack_name]
assert len(matching_stacks) == 1, f"stack with name {stack_name} should exist"

stack = matching_stacks[0]
assert stack["StackStatus"] == "CREATE_COMPLETE", "Stack should be in status CREATE_COMPLETE"

@mock_cloudformation
def test_delete_stack(self):
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
def test_delete_stack(self, cloudformation_conn_mock):
stack_name = "my_test_delete_stack_stack"
self.create_stack(stack_name)

self.hook.delete_stack(stack_name=stack_name)

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

0 comments on commit d05a4af

Please sign in to comment.