Repository: apache/airflow
File: airflow-core/tests/unit/dag_processing/bundles/test_base.py (Python)

Code before:
        assert b.lock_file_path is None
        assert b.lock_file is None

    def test_log_exc_formats_message_correctly(self):
        """Test that _log_exc correctly formats the log message with all parameters."""
        from airflow.dag_processing.bundles.base import log as bundle_log

        bundle_name = "test_bundle"
        bundle_version = "v1.0.0"
        lock = BundleVersionLock(
            bundle_name=bundle_name,
            bundle_version=bundle_version,
        )

        test_msg = "error when attempting to acquire lock"

        with patch.object(bundle_log, "exception") as mock_exception:
            lock._log_exc(test_msg)

            mock_exception.assert_called_once()
            call_args = mock_exception.call_args
            assert call_args[0][0] == "%s name=%s version=%s lock_file=%s"
            assert call_args[0][1] == test_msg
            assert call_args[0][2] == bundle_name
            assert call_args[0][3] == bundle_version
            assert call_args[0][4] == lock.lock_file_path


class FakeBundle(BaseDagBundle):
    @property
    def path(self) -> Path:
        assert self.version
        return self.versions_dir / self.version

    def get_current_version(self) -> str | None: ...
    def refresh(self) -> None: ...


class TestBundleUsageTrackingManager:
    @pytest.mark.parametrize(
        ("threshold_hours", "min_versions", "when_hours", "expected_remaining"),
        [
            (3, 0, 3, 5),
            (3, 0, 6, 2),
            (10, 0, 3, 5),
            (10, 0, 6, 5),
            (0, 0, 3, 2),  # two of them are in future
            (0, 0, 6, 0),  # all of them are in past
            (0, 5, 3, 5),  # keep all no matter what
            (0, 5, 6, 5),  # keep all no matter what
            (0, 4, 3, 4),  # keep 4 no matter what

Review comment (suggestion):
```suggestion
            assert mock_exception.mock_calls == [
                call(
                    "%s name=%s version=%s lock_file=%s",
                    test_msg,
                    bundle_name,
                    bundle_version,
                    lock.ock_file_paht
                )
            ]
```

Code after:
        assert b.lock_file_path is None
        assert b.lock_file is None

    def test_log_exc_formats_message_correctly(self):
        """Test that _log_exc correctly formats the log message with all parameters."""
        from airflow.dag_processing.bundles.base import log as bundle_log

        bundle_name = "test_bundle"
        bundle_version = "v1.0.0"
        lock = BundleVersionLock(
            bundle_name=bundle_name,
            bundle_version=bundle_version,
        )

        test_msg = "error when attempting to acquire lock"

        with patch.object(bundle_log, "exception") as mock_exception:
            lock._log_exc(test_msg)

            assert mock_exception.mock_calls == [
                call(
                    "%s name=%s version=%s lock_file=%s",
                    test_msg,
                    bundle_name,
                    bundle_version,
                    lock.lock_file_path,
                )
            ]


class FakeBundle(BaseDagBundle):
    @property
    def path(self) -> Path:
        assert self.version
        return self.versions_dir / self.version

    def get_current_version(self) -> str | None: ...
    def refresh(self) -> None: ...


class TestBundleUsageTrackingManager:
    @pytest.mark.parametrize(
        ("threshold_hours", "min_versions", "when_hours", "expected_remaining"),
        [
            (3, 0, 3, 5),
            (3, 0, 6, 2),
            (10, 0, 3, 5),
            (10, 0, 6, 5),
            (0, 0, 3, 2),  # two of them are in future
            (0, 0, 6, 0),  # all of them are in past
            (0, 5, 3, 5),  # keep all no matter what