You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Taragolis (via GitHub)" <gi...@apache.org> on 2023/02/16 12:39:05 UTC
[GitHub] [airflow] Taragolis commented on a diff in pull request #29377: Migrated email and generic transfer operators tests to `pytest`
Taragolis commented on code in PR #29377:
URL: https://github.com/apache/airflow/pull/29377#discussion_r1108415334
##########
tests/operators/test_datetime.py:
##########
@@ -113,121 +108,125 @@ def test_no_target_time(self):
dag=self.dag,
)
+ @pytest.mark.parametrize(
+ "target_lower,target_upper",
+ [(target_lower, target_upper) for (target_lower, target_upper) in targets],
+ )
@time_machine.travel("2020-07-07 10:54:05")
- def test_branch_datetime_operator_falls_within_range(self):
+ def test_branch_datetime_operator_falls_within_range(self, target_lower, target_upper):
"""Check BranchDateTimeOperator branch operation"""
- for target_lower, target_upper in self.targets:
- with self.subTest(target_lower=target_lower, target_upper=target_upper):
- self.branch_op.target_lower = target_lower
- self.branch_op.target_upper = target_upper
- self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
-
- self._assert_task_ids_match_states(
- {
- "datetime_branch": State.SUCCESS,
- "branch_1": State.NONE,
- "branch_2": State.SKIPPED,
- }
- )
+ self.branch_op.target_lower = target_lower
+ self.branch_op.target_upper = target_upper
+ self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+ self._assert_task_ids_match_states(
+ {
+ "datetime_branch": State.SUCCESS,
+ "branch_1": State.NONE,
+ "branch_2": State.SKIPPED,
+ }
+ )
- def test_branch_datetime_operator_falls_outside_range(self):
+ @pytest.mark.parametrize(
+ "target_lower,target_upper",
+ [(target_lower, target_upper) for (target_lower, target_upper) in targets],
Review Comment:
```suggestion
targets,
```
##########
tests/operators/test_datetime.py:
##########
@@ -35,24 +34,23 @@
INTERVAL = datetime.timedelta(hours=12)
-class TestBranchDateTimeOperator(unittest.TestCase):
+class TestBranchDateTimeOperator:
@classmethod
- def setUpClass(cls):
- super().setUpClass()
+ def setup_class(cls):
with create_session() as session:
session.query(DagRun).delete()
session.query(TI).delete()
- cls.targets = [
- (datetime.datetime(2020, 7, 7, 10, 0, 0), datetime.datetime(2020, 7, 7, 11, 0, 0)),
- (datetime.time(10, 0, 0), datetime.time(11, 0, 0)),
- (datetime.datetime(2020, 7, 7, 10, 0, 0), datetime.time(11, 0, 0)),
- (datetime.time(10, 0, 0), datetime.datetime(2020, 7, 7, 11, 0, 0)),
- (datetime.time(11, 0, 0), datetime.time(10, 0, 0)),
- ]
+ targets = [
+ (datetime.datetime(2020, 7, 7, 10, 0, 0), datetime.datetime(2020, 7, 7, 11, 0, 0)),
+ (datetime.time(10, 0, 0), datetime.time(11, 0, 0)),
+ (datetime.datetime(2020, 7, 7, 10, 0, 0), datetime.time(11, 0, 0)),
+ (datetime.time(10, 0, 0), datetime.datetime(2020, 7, 7, 11, 0, 0)),
+ (datetime.time(10, 0, 0), datetime.time(11, 0, 0)),
Review Comment:
There is two same parameters: `(datetime.time(10, 0, 0), datetime.time(11, 0, 0))`
That a good question what we actually expected when we pass two `datetime.time` arguments and `target_lower > target_upper` it could refers to the fact that BranchOperator work wrong with this case.
Currently this test pass due to the fact that subtests do not reset DagRuns and other stuff and use only first one, it is not a case for `@pytest.mark.parametrize` and I guess initially this test failed
##########
tests/operators/test_datetime.py:
##########
@@ -113,121 +108,125 @@ def test_no_target_time(self):
dag=self.dag,
)
+ @pytest.mark.parametrize(
+ "target_lower,target_upper",
+ [(target_lower, target_upper) for (target_lower, target_upper) in targets],
Review Comment:
```suggestion
targets,
```
##########
tests/operators/test_datetime.py:
##########
@@ -113,121 +108,125 @@ def test_no_target_time(self):
dag=self.dag,
)
+ @pytest.mark.parametrize(
+ "target_lower,target_upper",
+ [(target_lower, target_upper) for (target_lower, target_upper) in targets],
+ )
@time_machine.travel("2020-07-07 10:54:05")
- def test_branch_datetime_operator_falls_within_range(self):
+ def test_branch_datetime_operator_falls_within_range(self, target_lower, target_upper):
"""Check BranchDateTimeOperator branch operation"""
- for target_lower, target_upper in self.targets:
- with self.subTest(target_lower=target_lower, target_upper=target_upper):
- self.branch_op.target_lower = target_lower
- self.branch_op.target_upper = target_upper
- self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
-
- self._assert_task_ids_match_states(
- {
- "datetime_branch": State.SUCCESS,
- "branch_1": State.NONE,
- "branch_2": State.SKIPPED,
- }
- )
+ self.branch_op.target_lower = target_lower
+ self.branch_op.target_upper = target_upper
+ self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+ self._assert_task_ids_match_states(
+ {
+ "datetime_branch": State.SUCCESS,
+ "branch_1": State.NONE,
+ "branch_2": State.SKIPPED,
+ }
+ )
- def test_branch_datetime_operator_falls_outside_range(self):
+ @pytest.mark.parametrize(
+ "target_lower,target_upper",
+ [(target_lower, target_upper) for (target_lower, target_upper) in targets],
+ )
+ def test_branch_datetime_operator_falls_outside_range(self, target_lower, target_upper):
"""Check BranchDateTimeOperator branch operation"""
dates = [
datetime.datetime(2020, 7, 7, 12, 0, 0, tzinfo=datetime.timezone.utc),
datetime.datetime(2020, 6, 7, 12, 0, 0, tzinfo=datetime.timezone.utc),
]
- for target_lower, target_upper in self.targets:
- with self.subTest(target_lower=target_lower, target_upper=target_upper):
- self.branch_op.target_lower = target_lower
- self.branch_op.target_upper = target_upper
-
- for date in dates:
- with time_machine.travel(date):
- self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
-
- self._assert_task_ids_match_states(
- {
- "datetime_branch": State.SUCCESS,
- "branch_1": State.SKIPPED,
- "branch_2": State.NONE,
- }
- )
-
- @time_machine.travel("2020-07-07 10:54:05")
- def test_branch_datetime_operator_upper_comparison_within_range(self):
- """Check BranchDateTimeOperator branch operation"""
- for _, target_upper in self.targets:
- with self.subTest(target_upper=target_upper):
- self.branch_op.target_upper = target_upper
- self.branch_op.target_lower = None
+ self.branch_op.target_lower = target_lower
+ self.branch_op.target_upper = target_upper
+ for date in dates:
+ with time_machine.travel(date):
self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
self._assert_task_ids_match_states(
{
"datetime_branch": State.SUCCESS,
- "branch_1": State.NONE,
- "branch_2": State.SKIPPED,
+ "branch_1": State.SKIPPED,
+ "branch_2": State.NONE,
}
)
+ @pytest.mark.parametrize("target_upper", [target_upper for (_, target_upper) in targets])
@time_machine.travel("2020-07-07 10:54:05")
- def test_branch_datetime_operator_lower_comparison_within_range(self):
+ def test_branch_datetime_operator_upper_comparison_within_range(self, target_upper):
"""Check BranchDateTimeOperator branch operation"""
- for target_lower, _ in self.targets:
- with self.subTest(target_lower=target_lower):
- self.branch_op.target_lower = target_lower
- self.branch_op.target_upper = None
+ self.branch_op.target_upper = target_upper
+ self.branch_op.target_lower = None
- self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+ self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
- self._assert_task_ids_match_states(
- {
- "datetime_branch": State.SUCCESS,
- "branch_1": State.NONE,
- "branch_2": State.SKIPPED,
- }
- )
+ self._assert_task_ids_match_states(
+ {
+ "datetime_branch": State.SUCCESS,
+ "branch_1": State.NONE,
+ "branch_2": State.SKIPPED,
+ }
+ )
+
+ @pytest.mark.parametrize("target_lower", [target_lower for (target_lower, _) in targets])
+ @time_machine.travel("2020-07-07 10:54:05")
+ def test_branch_datetime_operator_lower_comparison_within_range(self, target_lower):
+ """Check BranchDateTimeOperator branch operation"""
+ self.branch_op.target_lower = target_lower
+ self.branch_op.target_upper = None
+ self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+ self._assert_task_ids_match_states(
+ {
+ "datetime_branch": State.SUCCESS,
+ "branch_1": State.NONE,
+ "branch_2": State.SKIPPED,
+ }
+ )
+
+ @pytest.mark.parametrize("target_upper", [target_upper for (_, target_upper) in targets])
@time_machine.travel("2020-07-07 12:00:00")
- def test_branch_datetime_operator_upper_comparison_outside_range(self):
+ def test_branch_datetime_operator_upper_comparison_outside_range(self, target_upper):
"""Check BranchDateTimeOperator branch operation"""
- for _, target_upper in self.targets:
- with self.subTest(target_upper=target_upper):
- self.branch_op.target_upper = target_upper
- self.branch_op.target_lower = None
+ self.branch_op.target_upper = target_upper
+ self.branch_op.target_lower = None
- self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+ self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
- self._assert_task_ids_match_states(
- {
- "datetime_branch": State.SUCCESS,
- "branch_1": State.SKIPPED,
- "branch_2": State.NONE,
- }
- )
+ self._assert_task_ids_match_states(
+ {
+ "datetime_branch": State.SUCCESS,
+ "branch_1": State.SKIPPED,
+ "branch_2": State.NONE,
+ }
+ )
+ @pytest.mark.parametrize("target_lower", [target_lower for (target_lower, _) in targets])
@time_machine.travel("2020-07-07 09:00:00")
- def test_branch_datetime_operator_lower_comparison_outside_range(self):
+ def test_branch_datetime_operator_lower_comparison_outside_range(self, target_lower):
"""Check BranchDateTimeOperator branch operation"""
- for target_lower, _ in self.targets:
- with self.subTest(target_lower=target_lower):
- self.branch_op.target_lower = target_lower
- self.branch_op.target_upper = None
+ self.branch_op.target_lower = target_lower
+ self.branch_op.target_upper = None
- self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+ self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
- self._assert_task_ids_match_states(
- {
- "datetime_branch": State.SUCCESS,
- "branch_1": State.SKIPPED,
- "branch_2": State.NONE,
- }
- )
+ self._assert_task_ids_match_states(
+ {
+ "datetime_branch": State.SUCCESS,
+ "branch_1": State.SKIPPED,
+ "branch_2": State.NONE,
+ }
+ )
+ @pytest.mark.parametrize(
+ "target_lower,target_upper",
+ [(target_lower, target_upper) for (target_lower, target_upper) in targets],
Review Comment:
```suggestion
targets,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org