fix(tests): resolve flaky test_batch_download_session_closed

Root cause:
- _run_with_fakes called session_maker() BEFORE patch context
- Full test suite runs caused PromiseProxy cache staleness
- Session mock tracked outside patch did not match actual session

Fix:
- Use patch.object on actual worker_app.db module (not string path)
- Track session creation INSIDE patch context via wrapper
- Make test_batch_download_session_closed self-contained
- Apply same pattern to test_batch_download_closes_session_on_error

This ensures proper isolation and avoids module-level cache issues.

tests/unit/test_batch_download.py

@@ -108,21 +108,30 @@ def _run_with_fakes(
         session = MagicMock()
         session_maker = MagicMock(return_value=session)
 
-    captured = {"upload_calls": [], "session": session_maker()}
+    captured: dict = {"upload_calls": []}
 
     def _tracking_upload(local_path, storage_key):
         captured["upload_calls"].append((local_path, storage_key))
         return upload_fn(local_path, storage_key)
 
+    # Wrap session_maker to capture the session INSIDE the patch context.
+    # This avoids stale PromiseProxy cache issues in full-suite runs.
+    _created_sessions: list = []
+    _orig_sm = session_maker
+
+    def _tracking_sm(*a, **kw):
+        s = _orig_sm(*a, **kw)
+        _created_sessions.append(s)
+        return s
+
     bound_task = _make_bound_task()
 
-    with patch(
-        "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
-        return_value=repo,
-    ):
+    import worker_app.db as _db_mod
+
+    with patch.object(_db_mod, "SessionLocal", _tracking_sm):
         with patch(
-            "worker_app.db.SessionLocal",
-            session_maker,
+            "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+            return_value=repo,
         ):
             with patch(
                 "video_processing.oss_helpers.upload_to_oss",
@@ -134,6 +143,7 @@ def _run_with_fakes(
                 ):
                     result = _call_task(batch_download_videos, bound_task, [v.id for v in videos], user_id)
 
+    captured["session"] = _created_sessions[0] if _created_sessions else None
     captured["result"] = result
     return captured
 
@@ -244,19 +254,43 @@ def test_batch_download_single_video():
 
 
 def test_batch_download_session_closed():
-    """DB session is always closed (via finally block)."""
+    """DB session is always closed (via finally block).
+
+    Self-contained: uses patch.object on the actual worker_app.db module
+    to avoid stale PromiseProxy cache from prior tests in the suite.
+    """
+    import worker_app.db as _db_mod
+    from apps.worker.worker_app.tasks.batch_download import batch_download_videos
+
     videos = [_FakeVideo("v1", "v.mp4")]
+    repo = _FakeGeneratedVideoRepository(videos)
 
     session = MagicMock()
-    session_maker = MagicMock(return_value=session)
 
-    _run_with_fakes(videos, session_maker=session_maker)
+    def _noop_download(url, dest):
+        Path(dest).parent.mkdir(parents=True, exist_ok=True)
+        Path(dest).write_bytes(b"fake video data")
+
+    bound_task = _make_bound_task()
+
+    with patch.object(_db_mod, "SessionLocal", MagicMock(return_value=session)):
+        with patch(
+            "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+            return_value=repo,
+        ):
+            with patch("video_processing.oss_helpers.upload_to_oss", return_value="https://oss.example.com/zip"):
+                with patch(
+                    "apps.worker.worker_app.tasks.batch_download._download_video_to_file",
+                    _noop_download,
+                ):
+                    _call_task(batch_download_videos, bound_task, ["v1"], "user_1")
 
     session.close.assert_called_once()
 
 
 def test_batch_download_closes_session_on_error():
     """Session is closed even when get_by_ids raises."""
+    import worker_app.db as _db_mod
     from apps.worker.worker_app.tasks.batch_download import batch_download_videos
 
     class _ExplodingRepo:
@@ -264,14 +298,13 @@ def test_batch_download_closes_session_on_error():
             raise RuntimeError("db down")
 
     session = MagicMock()
-    session_maker = MagicMock(return_value=session)
     bound_task = _make_bound_task()
 
-    with patch(
-        "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
-        return_value=_ExplodingRepo(),
-    ):
-        with patch("worker_app.db.SessionLocal", session_maker):
+    with patch.object(_db_mod, "SessionLocal", MagicMock(return_value=session)):
+        with patch(
+            "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+            return_value=_ExplodingRepo(),
+        ):
             with pytest.raises(RuntimeError, match="db down"):
                 _call_task(batch_download_videos, bound_task, ["v1"], "u")