kaxil commented on code in PR #65528:
URL: https://github.com/apache/airflow/pull/65528#discussion_r3110465804
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -799,6 +801,32 @@ def test_cleanup_stale_bundle_versions(self,
mock_bundle_manager):
manager.cleanup_stale_bundle_versions()
mock_bundle_manager.return_value.remove_stale_bundle_versions.assert_called_once_with()
+ @pytest.mark.parametrize(
+ ("log_target", "expected_subprocess_logs_to_stdout"),
+ [
+ ("stdout", True),
+ ("file", False),
+ ],
+ )
+ @mock.patch.object(DagFileProcessorManager, "_get_logger_for_dag_file")
+ def test_create_process_subprocess_logs_to_stdout(
+ self, mock_get_logger, configure_testing_dag_bundle, log_target,
expected_subprocess_logs_to_stdout
+ ):
+ mock_logger = MagicMock()
+ mock_filehandle = MagicMock()
+ mock_get_logger.return_value = [mock_logger, mock_filehandle]
+
+ with conf_vars({("logging", "dag_processor_log_target"): log_target}):
+ manager = DagFileProcessorManager(max_runs=1, processor_timeout=60)
+ dag_file = DagFileInfo(
+ bundle_name="testing", rel_path=Path("my_dag.py"),
bundle_path=Path("/tmp")
+ )
+ with mock.patch.object(DagFileProcessorProcess, "start") as
mock_start:
Review Comment:
This test mocks `DagFileProcessorProcess.start` itself, so it only asserts
that `_create_process` forwards the right kwarg. The logic that actually ships
the fix -- `_get_target_loggers` binding `dag_file`/`bundle_name`, the `task.`
-> `dag_processor.` rename in `_create_log_forwarder`, and the
`relative_to(bundle_path)` computation -- is untested. Worth a narrower unit
test against `DagFileProcessorProcess` directly: instantiate with
`subprocess_logs_to_stdout=True` and assert `_get_target_loggers()` returns
loggers bound with those fields, plus that `_create_log_forwarder` rewrites the
logger name.
##########
airflow-core/src/airflow/dag_processing/processor.py:
##########
@@ -547,7 +551,13 @@ def start( # type: ignore[override]
_pre_import_airflow_modules(os.fspath(path), logger)
- proc: Self = super().start(target=target, client=client, **kwargs)
+ proc: Self = super().start(
+ target=target,
+ client=client,
+ bundle_name=bundle_name,
+ dag_file_rel_path=str(Path(path).relative_to(bundle_path)),
Review Comment:
The caller (`DagFileProcessorManager._create_process`) already has
`dag_file.rel_path` available on the `DagFileInfo`. Recomputing it here via
`str(Path(path).relative_to(bundle_path))` duplicates that, and `relative_to`
will raise `ValueError` if `path` ever drifts outside `bundle_path` (e.g. a
symlink resolution). Mirroring the `bundle_name=bundle_name` pass-through a
line above and passing `dag_file_rel_path=str(dag_file.rel_path)` from the
manager side would be simpler and safer.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -799,6 +801,32 @@ def test_cleanup_stale_bundle_versions(self,
mock_bundle_manager):
manager.cleanup_stale_bundle_versions()
mock_bundle_manager.return_value.remove_stale_bundle_versions.assert_called_once_with()
+ @pytest.mark.parametrize(
+ ("log_target", "expected_subprocess_logs_to_stdout"),
+ [
+ ("stdout", True),
+ ("file", False),
+ ],
+ )
+ @mock.patch.object(DagFileProcessorManager, "_get_logger_for_dag_file")
+ def test_create_process_subprocess_logs_to_stdout(
+ self, mock_get_logger, configure_testing_dag_bundle, log_target,
expected_subprocess_logs_to_stdout
Review Comment:
`configure_testing_dag_bundle` is in the signature but never used as a
context manager in this test (every other test that pulls it enters it via
`with configure_testing_dag_bundle():`). Safe to drop from the args.
--
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]
