ephraimbuddy commented on code in PR #66296:
URL: https://github.com/apache/airflow/pull/66296#discussion_r3570678636
##########
airflow-core/tests/unit/utils/test_db_cleanup.py:
##########
@@ -1115,3 +1341,89 @@ def test_extra_filters_keep_in_flight_rows(self):
assert seeded[state] in survivors, f"{state} row should NOT be
cleaned up"
for state in ("success", "failed"):
assert seeded[state] not in survivors, f"{state} row should be
cleaned up"
+
+
+def _delete_test_timestamp():
+ return pendulum.DateTime(2024, 1, 2, 3, 4, 5,
tzinfo=pendulum.timezone("UTC"))
+
+
+def _build_do_delete_test_objects():
+ metadata = MetaData()
+ source_table = Table("dag_version", metadata, Column("id", Integer,
primary_key=True))
+ target_table = Table(
+ f"{ARCHIVE_TABLE_PREFIX}{source_table.name}__20240102030405",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ )
+ query = select(literal(1).label("id"))
+ return metadata, source_table, target_table, query
+
+
+def _create_pinned_dag_version_cleanup_data(*, base_date):
+ bundle_name = f"testing-{uuid4()}"
+ dag_id = f"test-dag_{uuid4()}"
+
+ with create_session() as session:
+ session.add(DagBundleModel(name=bundle_name))
+ session.flush()
+
+ dag = DAG(dag_id=dag_id)
+ session.add(DagModel(dag_id=dag_id, bundle_name=bundle_name))
+
+ old_dag_version = DagVersion(
+ dag_id=dag_id,
+ version_number=1,
+ bundle_name=bundle_name,
+ created_at=base_date,
+ last_updated=base_date,
+ )
+ new_dag_version = DagVersion(
+ dag_id=dag_id,
+ version_number=2,
+ bundle_name=bundle_name,
+ created_at=base_date.add(minutes=1),
+ last_updated=base_date.add(minutes=1),
+ )
+ session.add_all([old_dag_version, new_dag_version])
+ session.flush()
+
+ dag_run = DagRun(
+ dag_id,
+ run_id="run-1",
+ run_type=DagRunType.MANUAL,
+ start_date=base_date,
+ )
+ task = PythonOperator(task_id="dummy-task", dag=dag,
python_callable=print)
+ ti = create_task_instance(task, run_id=dag_run.run_id,
dag_version_id=old_dag_version.id)
+ ti.dag_id = dag_id
+ ti.start_date = base_date
+
+ session.add_all([dag_run, ti])
+ session.commit()
+
+ return base_date.add(days=1)
+
+
+def _get_dag_version_archive_table_names(*, session):
+ return [
+ table_name
+ for table_name in _get_archived_table_names(["dag_version"], session)
+ if table_name.startswith(f"{ARCHIVE_TABLE_PREFIX}dag_version__")
+ ]
+
+
+def _dag_version_config_without_row_exclusion():
+ """Live dag_version cleanup config with row-exclusion filters disabled.
+
+ The backend regression tests above pin ``_do_delete``'s
rollback-before-drop
+ behaviour (the MySQL metadata-lock hang from #66177), which requires the
Review Comment:
```suggestion
behaviour, which requires the
```
--
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]
