xBis7 commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3461597872
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +201,59 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self,
testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
+ def test_serialization_metric_incremented_on_new_write(self,
testing_dag_bundle):
+ """A brand new serialized DAG write emits the ``dag.serialization``
metric."""
+ dag =
make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="testing") is True
+
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": dag.dag_id, "bundle_name": "testing"},
+ )
+
+ def test_serialization_metric_not_incremented_when_unchanged(self,
testing_dag_bundle):
+ """Re-writing an unchanged DAG must not emit the ``dag.serialization``
metric."""
+ dag =
make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="testing") is True
+
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="testing") is False
+
+ mock_stats.incr.assert_not_called()
+
+ def test_serialization_metric_incremented_on_inplace_update(self,
dag_maker, session):
+ """Updating a DAG version in place (no dag runs) emits the metric
once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ # Change the DAG so the hash differs; with no dag runs this updates in
place.
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag),
bundle_name="dag_maker") is True
+
+ assert session.scalar(select(func.count()).select_from(DagVersion)) ==
1
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": "metric_dag", "bundle_name": "dag_maker"},
+ )
+
+ def test_serialization_metric_incremented_on_new_version(self, dag_maker,
session):
+ """Writing a new DAG version (existing run) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ dag_maker.create_dagrun(run_id="run1",
logical_date=pendulum.datetime(2025, 1, 1))
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
Review Comment:
You are reusing `airflow.models.serialized_dag.stats` in every test. This
can be a common variable.
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +201,59 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self,
testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
Review Comment:
Can you add another unit test that asserts both the new name and the legacy
name are exported? You can do that by mocking the backend. See this test as an
example
https://github.com/apache/airflow/blob/main/airflow-core/tests/unit/jobs/test_scheduler_job.py#L473-L524
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
