Taragolis commented on code in PR #27970:
URL: https://github.com/apache/airflow/pull/27970#discussion_r1034285775
##
tests/providers/amazon/aws/operators/test_eks.py:
##
@@ -243,80 +233,74 @@ def
test_fargate_compute_missing_fargate_pod_execution_role_arn(self):
missing_fargate_pod_execution_role_arn.execute({})
-class TestEksCreateFargateProfileOperator(unittest.TestCase):
-def setUp(self) -> None:
-self.create_fargate_profile_params: CreateFargateProfileParams = dict(
# type: ignore
+class TestEksCreateFargateProfileOperator:
Review Comment:
replace TestCase.subTest by parametrize tests
##
tests/providers/amazon/aws/operators/test_eks.py:
##
@@ -92,35 +91,21 @@ class CreateNodegroupParams(TypedDict):
nodegroup_role_arn: str
-class TestEksCreateClusterOperator(unittest.TestCase):
-def setUp(self) -> None:
+class TestEksCreateClusterOperator:
Review Comment:
1. replace TestCase.subTest by parametrize tests
2. Rename method `nodegroup_setUp` to `nodegroup_setup`
3. Rename method `fargate_profile_setup ` to `fargate_profile_setup `
##
tests/providers/amazon/aws/operators/test_s3_file_transform.py:
##
@@ -34,37 +29,33 @@
from airflow.providers.amazon.aws.operators.s3 import S3FileTransformOperator
-class TestS3FileTransformOperator(unittest.TestCase):
-def setUp(self):
+@pytest.fixture
+def transform_script_loc(request, tmp_path_factory):
+transform_script = tmp_path_factory.mktemp(request.node.name) /
"transform.py"
+transform_script.touch()
+yield str(transform_script)
Review Comment:
Create empty "transform" file by `tmp_path_factory` template instead of
manual creation temporary directory and clear up it
##
tests/providers/amazon/aws/operators/test_s3_object.py:
##
@@ -18,10 +18,10 @@
from __future__ import annotations
Review Comment:
This test I entirely get from https://github.com/apache/airflow/pull/27858
##
tests/providers/amazon/aws/operators/test_sagemaker_training.py:
##
@@ -57,11 +57,12 @@
}
-class TestSageMakerTrainingOperator(unittest.TestCase):
-def setUp(self):
+class TestSageMakerTrainingOperator:
+def setup_method(self):
+self.create_training_params = copy.deepcopy(CREATE_TRAINING_PARAMS)
self.sagemaker = SageMakerTrainingOperator(
task_id="test_sagemaker_operator",
-config=CREATE_TRAINING_PARAMS,
+config=self.create_training_params,
wait_for_completion=False,
check_interval=5,
)
Review Comment:
After migrate to pytest some test failed due to mutability of test
parameters so this parameters recreate for each test case.
##
tests/providers/amazon/aws/operators/test_sagemaker_transform.py:
##
@@ -50,15 +50,16 @@
"ExecutionRoleArn": "arn:aws:iam:role/test-role",
}
-CONFIG: dict = {"Model": CREATE_MODEL_PARAMS, "Transform":
CREATE_TRANSFORM_PARAMS}
-
-class TestSageMakerTransformOperator(unittest.TestCase):
-def setUp(self):
+class TestSageMakerTransformOperator:
+def setup_method(self):
+self.create_transform_params = copy.deepcopy(CREATE_TRANSFORM_PARAMS)
+self.create_model_params = copy.deepcopy(CREATE_MODEL_PARAMS)
+self.config = {"Model": self.create_model_params, "Transform":
self.create_transform_params}
Review Comment:
After migrate to `pytest` some test failed due to mutability of test
parameters so this parameters recreate for each test case.
##
tests/providers/amazon/aws/sensors/test_emr_containers.py:
##
@@ -41,36 +40,18 @@ def setUp(self):
# avoids an Airflow warning about connection cannot be found.
self.sensor.hook.get_connection = lambda _: None
-@mock.patch.object(EmrContainerHook, "check_query_status",
side_effect=("PENDING",))
-def test_poke_pending(self, mock_check_query_status):
-assert not self.sensor.poke(None)
-
-@mock.patch.object(EmrContainerHook, "check_query_status",
side_effect=("SUBMITTED",))
-def test_poke_submitted(self, mock_check_query_status):
-assert not self.sensor.poke(None)
-
-@mock.patch.object(EmrContainerHook, "check_query_status",
side_effect=("RUNNING",))
-def test_poke_running(self, mock_check_query_status):
-assert not self.sensor.poke(None)
-
-@mock.patch.object(EmrContainerHook, "check_query_status",
side_effect=("COMPLETED",))
-def test_poke_completed(self, mock_check_query_status):
-assert self.sensor.poke(None)
-
-@mock.patch.object(EmrContainerHook, "check_query_status",
side_effect=("FAILED",))
-def test_poke_failed(self, mock_check_query_status):
-with pytest.raises(AirflowException) as ctx:
-self.sensor.poke(None)
-assert "EMR Containers sensor failed" in str(ctx.value)
-
-@mock.patch.object(EmrContainerHook, "check_query_status",
