fix: extract frames for map faces fallback#1824
Conversation
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- By unconditionally calling
create_temp/extract_framesyou now incur disk extraction for non-map_facesruns as well; if the in-memory pipeline truly doesn’t need these frames, consider scoping the extraction back to only the code paths that actually consume on-disk frames to avoid unnecessary work. - The test helper directly mutates
sys.moduleswithout cleanup, which can leak state across tests or affect other imports; consider wrapping these stubs in a context manager or usingaddCleanupto restoresys.modulesafter each test.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- By unconditionally calling `create_temp`/`extract_frames` you now incur disk extraction for non-`map_faces` runs as well; if the in-memory pipeline truly doesn’t need these frames, consider scoping the extraction back to only the code paths that actually consume on-disk frames to avoid unnecessary work.
- The test helper directly mutates `sys.modules` without cleanup, which can leak state across tests or affect other imports; consider wrapping these stubs in a context manager or using `addCleanup` to restore `sys.modules` after each test.
## Individual Comments
### Comment 1
<location path="tests/test_core_map_faces_fallback.py" line_range="82-91" />
<code_context>
+ core.start()
+
+ self.assertNotIn(("pipe",), calls)
+ self.assertIn(("create_temp", "target.mp4"), calls)
+ self.assertIn(("extract_frames", "target.mp4"), calls)
+ self.assertIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls)
+
+ extract_index = calls.index(("extract_frames", "target.mp4"))
+ process_index = calls.index(("process_video", "source.jpg", ("target.mp4/0001.png",)))
+ self.assertLess(extract_index, process_index)
+
</code_context>
<issue_to_address>
**suggestion (testing):** Strengthen ordering assertions to cover the full pipeline sequence instead of only extract-before-process.
Right now the test only checks that `extract_frames` precedes `process_video`. For stronger regression protection, also assert the relative order of the other key steps: e.g., `create_temp` before `extract_frames`, `process_video` before `create_video` / `restore_audio` / `move_temp`. You can get the indices of these calls in `calls` and use `assertLess` between them to validate the full pipeline order, ensuring the fallback path remains correct end-to-end.
```suggestion
core.start()
+
+ self.assertNotIn(("pipe",), calls)
+ self.assertIn(("create_temp", "target.mp4"), calls)
+ self.assertIn(("extract_frames", "target.mp4"), calls)
+ self.assertIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls)
+
+ # Validate the full pipeline ordering on the fallback path:
+ # create_temp -> extract_frames -> process_video -> create_video -> restore_audio -> move_temp
+ step_indices = {}
+ for idx, call in enumerate(calls):
+ name = call[0]
+ if name in {"create_temp", "extract_frames", "process_video", "create_video", "restore_audio", "move_temp"}:
+ # Only record the first occurrence of each step
+ step_indices.setdefault(name, idx)
+
+ # Ensure all expected steps are present
+ self.assertIn("create_temp", step_indices)
+ self.assertIn("extract_frames", step_indices)
+ self.assertIn("process_video", step_indices)
+ self.assertIn("create_video", step_indices)
+ self.assertIn("restore_audio", step_indices)
+ self.assertIn("move_temp", step_indices)
+
+ # And that they occur in the correct order
+ self.assertLess(step_indices["create_temp"], step_indices["extract_frames"])
+ self.assertLess(step_indices["extract_frames"], step_indices["process_video"])
+ self.assertLess(step_indices["process_video"], step_indices["create_video"])
+ self.assertLess(step_indices["create_video"], step_indices["restore_audio"])
+ self.assertLess(step_indices["restore_audio"], step_indices["move_temp"])
```
</issue_to_address>
### Comment 2
<location path="tests/test_core_map_faces_fallback.py" line_range="8" />
<code_context>
+from unittest.mock import patch
+
+
+def _install_core_import_stubs(calls):
+ sys.modules.setdefault("torch", types.SimpleNamespace(cuda=types.SimpleNamespace(empty_cache=lambda: None)))
+ sys.modules["onnxruntime"] = types.SimpleNamespace(
</code_context>
<issue_to_address>
**suggestion (testing):** Isolate sys.modules stubbing to avoid cross-test leakage and make future tests safer.
This approach works for a single test, but mutating `sys.modules` globally can cause cross-test interference as the suite grows. Please wrap these stubs in a `patch.dict(sys.modules, {...}, clear=False)` context (or restore originals in `tearDown`) so each test runs with a clean set of module stubs and avoids flaky interactions with other tests importing the same modules.
Suggested implementation:
```python
def _install_core_import_stubs(calls):
"""Return a context manager that patches sys.modules with core import stubs.
Using patch.dict ensures the original sys.modules entries are restored
after the context exits, avoiding cross-test interference.
"""
stubs = {
"torch": types.SimpleNamespace(
cuda=types.SimpleNamespace(empty_cache=lambda: None)
),
"onnxruntime": types.SimpleNamespace(
get_available_providers=lambda: ["CPUExecutionProvider"]
),
"tensorflow": types.SimpleNamespace(),
"modules.metadata": types.SimpleNamespace(
name="Deep-Live-Cam", version="test"
),
"modules.ui": types.SimpleNamespace(
check_and_ignore_nsfw=lambda *_args, **_kwargs: False,
update_status=lambda *_args, **_kwargs: None,
init=lambda *_args, **_kwargs: types.SimpleNamespace(
mainloop=lambda: None
),
),
}
return patch.dict(sys.modules, stubs, clear=False)
```
1. Any existing call sites that rely on `_install_core_import_stubs(calls)` performing global mutation must be updated to use the returned context manager, for example:
- `with _install_core_import_stubs(calls): ...`
- or in `setUp`: `self._modules_patcher = _install_core_import_stubs(calls); self._modules_patcher.start()`
and in `tearDown`: `self._modules_patcher.stop()`.
2. If some tests expect `sys.modules` to retain these stubs after they complete, those expectations should be adjusted so each test is self-contained and does not depend on leaked stubs.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| core.start() | ||
|
|
||
| self.assertNotIn(("pipe",), calls) | ||
| self.assertIn(("create_temp", "target.mp4"), calls) | ||
| self.assertIn(("extract_frames", "target.mp4"), calls) | ||
| self.assertIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls) | ||
|
|
||
| extract_index = calls.index(("extract_frames", "target.mp4")) | ||
| process_index = calls.index(("process_video", "source.jpg", ("target.mp4/0001.png",))) | ||
| self.assertLess(extract_index, process_index) |
There was a problem hiding this comment.
suggestion (testing): Strengthen ordering assertions to cover the full pipeline sequence instead of only extract-before-process.
Right now the test only checks that extract_frames precedes process_video. For stronger regression protection, also assert the relative order of the other key steps: e.g., create_temp before extract_frames, process_video before create_video / restore_audio / move_temp. You can get the indices of these calls in calls and use assertLess between them to validate the full pipeline order, ensuring the fallback path remains correct end-to-end.
| core.start() | |
| self.assertNotIn(("pipe",), calls) | |
| self.assertIn(("create_temp", "target.mp4"), calls) | |
| self.assertIn(("extract_frames", "target.mp4"), calls) | |
| self.assertIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls) | |
| extract_index = calls.index(("extract_frames", "target.mp4")) | |
| process_index = calls.index(("process_video", "source.jpg", ("target.mp4/0001.png",))) | |
| self.assertLess(extract_index, process_index) | |
| core.start() | |
| + | |
| + self.assertNotIn(("pipe",), calls) | |
| + self.assertIn(("create_temp", "target.mp4"), calls) | |
| + self.assertIn(("extract_frames", "target.mp4"), calls) | |
| + self.assertIn(("process_video", "source.jpg", ("target.mp4/0001.png",)), calls) | |
| + | |
| + # Validate the full pipeline ordering on the fallback path: | |
| + # create_temp -> extract_frames -> process_video -> create_video -> restore_audio -> move_temp | |
| + step_indices = {} | |
| + for idx, call in enumerate(calls): | |
| + name = call[0] | |
| + if name in {"create_temp", "extract_frames", "process_video", "create_video", "restore_audio", "move_temp"}: | |
| + # Only record the first occurrence of each step | |
| + step_indices.setdefault(name, idx) | |
| + | |
| + # Ensure all expected steps are present | |
| + self.assertIn("create_temp", step_indices) | |
| + self.assertIn("extract_frames", step_indices) | |
| + self.assertIn("process_video", step_indices) | |
| + self.assertIn("create_video", step_indices) | |
| + self.assertIn("restore_audio", step_indices) | |
| + self.assertIn("move_temp", step_indices) | |
| + | |
| + # And that they occur in the correct order | |
| + self.assertLess(step_indices["create_temp"], step_indices["extract_frames"]) | |
| + self.assertLess(step_indices["extract_frames"], step_indices["process_video"]) | |
| + self.assertLess(step_indices["process_video"], step_indices["create_video"]) | |
| + self.assertLess(step_indices["create_video"], step_indices["restore_audio"]) | |
| + self.assertLess(step_indices["restore_audio"], step_indices["move_temp"]) |
There was a problem hiding this comment.
suggestion (testing): Isolate sys.modules stubbing to avoid cross-test leakage and make future tests safer.
This approach works for a single test, but mutating sys.modules globally can cause cross-test interference as the suite grows. Please wrap these stubs in a patch.dict(sys.modules, {...}, clear=False) context (or restore originals in tearDown) so each test runs with a clean set of module stubs and avoids flaky interactions with other tests importing the same modules.
Suggested implementation:
def _install_core_import_stubs(calls):
"""Return a context manager that patches sys.modules with core import stubs.
Using patch.dict ensures the original sys.modules entries are restored
after the context exits, avoiding cross-test interference.
"""
stubs = {
"torch": types.SimpleNamespace(
cuda=types.SimpleNamespace(empty_cache=lambda: None)
),
"onnxruntime": types.SimpleNamespace(
get_available_providers=lambda: ["CPUExecutionProvider"]
),
"tensorflow": types.SimpleNamespace(),
"modules.metadata": types.SimpleNamespace(
name="Deep-Live-Cam", version="test"
),
"modules.ui": types.SimpleNamespace(
check_and_ignore_nsfw=lambda *_args, **_kwargs: False,
update_status=lambda *_args, **_kwargs: None,
init=lambda *_args, **_kwargs: types.SimpleNamespace(
mainloop=lambda: None
),
),
}
return patch.dict(sys.modules, stubs, clear=False)- Any existing call sites that rely on
_install_core_import_stubs(calls)performing global mutation must be updated to use the returned context manager, for example:with _install_core_import_stubs(calls): ...- or in
setUp:self._modules_patcher = _install_core_import_stubs(calls); self._modules_patcher.start()
and intearDown:self._modules_patcher.stop().
- If some tests expect
sys.modulesto retain these stubs after they complete, those expectations should be adjusted so each test is self-contained and does not depend on leaked stubs.

Summary
--map-facesuse the disk path correctlymap_facesfallback path so frame extraction happens beforeprocess_videoTests
python3 -m unittest tests/test_core_map_faces_fallback.py -vpython3 -m py_compile modules/core.py tests/test_core_map_faces_fallback.pygit diff --checkpr-value
pr-value-r=86,pr-value-l=100, gate passedSummary by Sourcery
Ensure video frames are always extracted to a temporary directory before disk-based processing runs, including when map-faces fallback is used.
Bug Fixes:
Tests: