This is an automated email from the ASF dual-hosted git repository.
husseinawala pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 05377b319b Restore cloud formation tests which use moto after
openapi_schema_validator 0.6.0 is released (#32511)
05377b319b is described below
commit 05377b319b5faf08322253d726157a7130d5c30e
Author: Hussein Awala <[email protected]>
AuthorDate: Tue Jul 11 10:14:08 2023 +0200
Restore cloud formation tests which use moto after openapi_schema_validator
0.6.0 is released (#32511)
Signed-off-by: Hussein Awala <[email protected]>
---
.../amazon/aws/hooks/test_cloud_formation.py | 145 +++++----------------
1 file changed, 29 insertions(+), 116 deletions(-)
diff --git a/tests/providers/amazon/aws/hooks/test_cloud_formation.py
b/tests/providers/amazon/aws/hooks/test_cloud_formation.py
index 1fd972d081..ce9eed6e6d 100644
--- a/tests/providers/amazon/aws/hooks/test_cloud_formation.py
+++ b/tests/providers/amazon/aws/hooks/test_cloud_formation.py
@@ -18,88 +18,10 @@
from __future__ import annotations
import json
-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
+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"
+from airflow.providers.amazon.aws.hooks.cloud_formation import
CloudFormationHook
class TestCloudFormationHook:
@@ -138,51 +60,42 @@ class TestCloudFormationHook:
},
)
-
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
- def test_get_conn_returns_a_boto3_connection(self,
cloudformation_conn_mock):
+ @mock_cloudformation
+ def test_get_conn_returns_a_boto3_connection(self):
assert self.hook.get_conn().describe_stacks() is not None
-
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
- def test_get_stack_status(self, cloudformation_conn_mock):
+ @mock_cloudformation
+ def test_get_stack_status(self):
stack_name = "my_test_get_stack_status_stack"
- self.hook.get_stack_status(stack_name=stack_name)
-
cloudformation_conn_mock.return_value.describe_stacks.assert_called_once_with(StackName=stack_name)
+ 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.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
- def test_create_stack(self, cloudformation_conn_mock):
+ @mock_cloudformation
+ def test_create_stack(self):
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"}],
- )
-
@mock.patch("airflow.providers.amazon.aws.hooks.cloud_formation.CloudFormationHook.get_conn")
- def test_delete_stack(self, cloudformation_conn_mock):
+ 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)
-
cloudformation_conn_mock.return_value.delete_stack.assert_called_once_with(StackName=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"
