mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
72a8dafde9
Review follow-ups on applying camera_rotation to finish photos and
layer-timelapse frames.
Rotating the frame popped from _stage22_finish_frames rotated one of its
sources twice. The cache has two kinds of feeder: live grabs, which are raw,
and the #1867 in-print bank, whose bytes come from
_capture_snapshot_for_notification and have already been rotated on the way
in. The consumer cannot tell them apart, so on the finish_state trigger - the
path the bank exists to serve, on firmware that never emits stg_cur=22 - a 180
degree rotation cancelled itself out and the photo was upside-down again,
which is the reported symptom exactly; 90 and 270 landed 180 out. Rotation now
happens where each frame is captured, so every entry in the cache carries one
rotation whatever produced it, and the invariant is stated both where the
cache is declared and where it is consumed.
Two finish-photo sources were still writing unrotated files: the built-in
camera's own capture_finish_photo, and the still extracted from a
printer-recorded timelapse - which is the *preferred* source for a built-in
camera print, so a user with a rotation set got a correctly oriented photo or
not depending on which source happened to win. Neither ever holds the frame as
bytes; ffmpeg writes the file and they return a filename. apply_camera_rotation_to_file
handles that case and is best-effort - a failed rotate leaves the unrotated
file rather than losing a delivered photo. The archived video itself is the
printer's own file and is not re-encoded, so it still plays at the camera's
native orientation; the CHANGELOG says so rather than leaving it to be
discovered.
apply_camera_rotation logs at debug, not info. It was on a path that runs once
per layer, where a tall print would have put hundreds of lines in the log for
something the surrounding capture already reports at debug.
The moved rotation logic had no test of its own - every existing test patches
it out and asserts the call, so a flipped sign or a dropped expand=True would
have shipped green. test_camera_rotation.py drives the real round trip: a
corner marker pins which way it turns, the dimensions pin that the frame is
not cropped, and an undecodable frame comes back by identity because a capture
path must not lose a frame to a failed rotate.
Tests for the fix itself sit on both sides of the cache. The producer half is
driven directly; the consumer half is a closure nested inside on_print_complete
with nothing able to reach it, so it is pinned by an AST guard - checked
against the source because the alternative is no check at all. Reverting
main.py to the pre-fix shape fails three of the five, the guard among them.
The three new tests used Path("/tmp/test") for a patched base_dir, which Bandit
flagged (B108); they take tmp_path now.
647 lines
25 KiB
Python
647 lines
25 KiB
Python
"""
|
|
Tests for the layer timelapse service.
|
|
|
|
These tests cover session management and pure logic functions.
|
|
"""
|
|
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import ANY, AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestTimelapseSessionManagement:
|
|
"""Tests for timelapse session lifecycle."""
|
|
|
|
def test_start_session_creates_new_session(self):
|
|
"""Verify start_session creates and registers a new session."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cancel_session,
|
|
get_session,
|
|
start_session,
|
|
)
|
|
|
|
# Clear any existing sessions
|
|
_active_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test_bambuddy")
|
|
|
|
session = start_session(
|
|
printer_id=1,
|
|
archive_id=100,
|
|
url="http://camera.local/mjpeg",
|
|
cam_type="mjpeg",
|
|
)
|
|
|
|
assert session is not None
|
|
assert session.printer_id == 1
|
|
assert session.archive_id == 100
|
|
assert session.camera_url == "http://camera.local/mjpeg"
|
|
assert session.camera_type == "mjpeg"
|
|
assert session.last_layer == -1
|
|
assert session.frame_count == 0
|
|
|
|
# Session should be retrievable
|
|
retrieved = get_session(1)
|
|
assert retrieved is session
|
|
|
|
# Cleanup
|
|
cancel_session(1)
|
|
|
|
def test_start_session_cancels_existing(self):
|
|
"""Verify starting a new session cancels any existing session."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cancel_session,
|
|
get_session,
|
|
start_session,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test_bambuddy")
|
|
|
|
# Start first session
|
|
session1 = start_session(1, 100, "http://cam1/", "mjpeg")
|
|
|
|
# Mock cleanup to track if it was called
|
|
session1.cleanup = MagicMock()
|
|
|
|
# Start second session for same printer
|
|
session2 = start_session(1, 101, "http://cam2/", "rtsp")
|
|
|
|
# First session should be replaced
|
|
current = get_session(1)
|
|
assert current is session2
|
|
assert current.archive_id == 101 # Verify it's the new session
|
|
assert current.camera_url == "http://cam2/"
|
|
|
|
# First session's cleanup should have been called
|
|
session1.cleanup.assert_called_once()
|
|
|
|
# Cleanup
|
|
cancel_session(1)
|
|
|
|
def test_get_session_returns_none_for_unknown(self):
|
|
"""Verify get_session returns None for unknown printer."""
|
|
from backend.app.services.layer_timelapse import _active_sessions, get_session
|
|
|
|
_active_sessions.clear()
|
|
|
|
result = get_session(999)
|
|
assert result is None
|
|
|
|
def test_cancel_session_removes_and_cleans_up(self):
|
|
"""Verify cancel_session removes session and cleans up."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cancel_session,
|
|
get_session,
|
|
start_session,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test_bambuddy")
|
|
|
|
session = start_session(1, 100, "http://cam/", "mjpeg")
|
|
|
|
# Mock cleanup to avoid filesystem operations
|
|
session.cleanup = MagicMock()
|
|
|
|
cancel_session(1)
|
|
|
|
# Session should be removed
|
|
assert get_session(1) is None
|
|
# Cleanup should have been called
|
|
session.cleanup.assert_called_once()
|
|
|
|
def test_cancel_nonexistent_session_is_safe(self):
|
|
"""Verify canceling a non-existent session doesn't error."""
|
|
from backend.app.services.layer_timelapse import _active_sessions, cancel_session
|
|
|
|
_active_sessions.clear()
|
|
|
|
# Should not raise
|
|
cancel_session(999)
|
|
|
|
|
|
class TestTimelapseSession:
|
|
"""Tests for TimelapseSession class."""
|
|
|
|
def test_session_id_format(self):
|
|
"""Verify session ID follows expected datetime format."""
|
|
from backend.app.services.layer_timelapse import TimelapseSession, _active_sessions
|
|
|
|
_active_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test_bambuddy")
|
|
|
|
session = TimelapseSession(
|
|
printer_id=1,
|
|
archive_id=100,
|
|
camera_url="http://test/",
|
|
camera_type="mjpeg",
|
|
)
|
|
|
|
# Session ID should be timestamp format YYYYMMDD_HHMMSS
|
|
assert len(session.session_id) == 15
|
|
assert session.session_id[8] == "_"
|
|
|
|
# Should be parseable as datetime
|
|
try:
|
|
datetime.strptime(session.session_id, "%Y%m%d_%H%M%S")
|
|
except ValueError:
|
|
pytest.fail("Session ID is not valid datetime format")
|
|
|
|
def test_frames_dir_path_structure(self):
|
|
"""Verify frames directory path is structured correctly."""
|
|
from backend.app.services.layer_timelapse import TimelapseSession
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/data/bambuddy")
|
|
|
|
with patch.object(Path, "mkdir"): # Avoid creating real directories
|
|
session = TimelapseSession(
|
|
printer_id=42,
|
|
archive_id=100,
|
|
camera_url="http://test/",
|
|
camera_type="mjpeg",
|
|
)
|
|
|
|
expected_path = Path("/data/bambuddy/timelapse_frames/42") / session.session_id
|
|
assert session.frames_dir == expected_path
|
|
|
|
|
|
class TestLayerChangeLogic:
|
|
"""Tests for layer change capture logic."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capture_layer_only_on_increase(self):
|
|
"""Verify frames are only captured when layer increases."""
|
|
from backend.app.services.layer_timelapse import TimelapseSession
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test")
|
|
|
|
with patch.object(Path, "mkdir"):
|
|
session = TimelapseSession(1, 100, "http://test/", "mjpeg")
|
|
|
|
# Mock capture_frame to return data
|
|
with patch(
|
|
"backend.app.services.layer_timelapse.capture_frame", new_callable=AsyncMock
|
|
) as mock_capture:
|
|
mock_capture.return_value = b"\xff\xd8test\xff\xd9"
|
|
|
|
with patch.object(Path, "write_bytes"):
|
|
# First layer should capture
|
|
result = await session.capture_layer(1)
|
|
assert result is True
|
|
assert session.last_layer == 1
|
|
assert session.frame_count == 1
|
|
|
|
# Same layer should NOT capture
|
|
result = await session.capture_layer(1)
|
|
assert result is False
|
|
assert session.frame_count == 1
|
|
|
|
# Lower layer should NOT capture
|
|
result = await session.capture_layer(0)
|
|
assert result is False
|
|
assert session.frame_count == 1
|
|
|
|
# Higher layer should capture
|
|
result = await session.capture_layer(5)
|
|
assert result is True
|
|
assert session.last_layer == 5
|
|
assert session.frame_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_capture_layer_handles_failed_capture(self):
|
|
"""Verify failed capture returns False but updates layer."""
|
|
from backend.app.services.layer_timelapse import TimelapseSession
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test")
|
|
|
|
with patch.object(Path, "mkdir"):
|
|
session = TimelapseSession(1, 100, "http://test/", "mjpeg")
|
|
|
|
# Mock capture_frame to return None (failure)
|
|
with patch(
|
|
"backend.app.services.layer_timelapse.capture_frame", new_callable=AsyncMock
|
|
) as mock_capture:
|
|
mock_capture.return_value = None
|
|
|
|
result = await session.capture_layer(1)
|
|
|
|
assert result is False
|
|
assert session.last_layer == 1 # Layer is still updated
|
|
assert session.frame_count == 0 # But frame count not incremented
|
|
|
|
|
|
class TestCaptureLayerAppliesRotation:
|
|
"""camera_rotation was previously only wired into the notification-
|
|
snapshot path, so a layer-timelapse video came out upside-down whenever
|
|
the printer had a rotation configured. capture_layer now applies it to
|
|
every captured frame, whether fresh or reused from the live view's
|
|
buffer, before writing to disk."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rotates_fresh_capture_when_configured(self, tmp_path):
|
|
from backend.app.services.layer_timelapse import TimelapseSession
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
|
|
with patch.object(Path, "mkdir"):
|
|
session = TimelapseSession(1, 100, "/dev/video1", "usb", rotation=180)
|
|
|
|
with (
|
|
patch("backend.app.api.routes.camera.live_frame_for_capture", return_value=(False, None)),
|
|
patch(
|
|
"backend.app.services.layer_timelapse.capture_frame",
|
|
new_callable=AsyncMock,
|
|
return_value=b"\xff\xd8unrotated\xff\xd9",
|
|
),
|
|
patch(
|
|
"backend.app.services.layer_timelapse.apply_camera_rotation",
|
|
return_value=b"\xff\xd8rotated\xff\xd9",
|
|
) as mock_rotate,
|
|
patch.object(Path, "write_bytes") as mock_write,
|
|
):
|
|
result = await session.capture_layer(1)
|
|
|
|
assert result is True
|
|
mock_rotate.assert_called_once_with(b"\xff\xd8unrotated\xff\xd9", 180, ANY)
|
|
mock_write.assert_called_once_with(b"\xff\xd8rotated\xff\xd9")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rotates_buffered_frame_when_configured(self, tmp_path):
|
|
from backend.app.services.layer_timelapse import TimelapseSession
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
|
|
with patch.object(Path, "mkdir"):
|
|
session = TimelapseSession(1, 100, "/dev/video1", "usb", rotation=90)
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.api.routes.camera.live_frame_for_capture",
|
|
return_value=(True, b"\xff\xd8buffered\xff\xd9"),
|
|
),
|
|
patch(
|
|
"backend.app.services.layer_timelapse.apply_camera_rotation",
|
|
return_value=b"\xff\xd8rotated\xff\xd9",
|
|
) as mock_rotate,
|
|
patch.object(Path, "write_bytes") as mock_write,
|
|
):
|
|
result = await session.capture_layer(1)
|
|
|
|
assert result is True
|
|
mock_rotate.assert_called_once_with(b"\xff\xd8buffered\xff\xd9", 90, ANY)
|
|
mock_write.assert_called_once_with(b"\xff\xd8rotated\xff\xd9")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_rotation_when_not_configured(self, tmp_path):
|
|
"""Default rotation=0 - no-op, and must not even call apply_camera_rotation
|
|
(avoids the PIL decode/re-encode round trip for the common case)."""
|
|
from backend.app.services.layer_timelapse import TimelapseSession
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
|
|
with patch.object(Path, "mkdir"):
|
|
session = TimelapseSession(1, 100, "/dev/video1", "usb")
|
|
assert session.rotation == 0
|
|
|
|
with (
|
|
patch("backend.app.api.routes.camera.live_frame_for_capture", return_value=(False, None)),
|
|
patch(
|
|
"backend.app.services.layer_timelapse.capture_frame",
|
|
new_callable=AsyncMock,
|
|
return_value=b"\xff\xd8unrotated\xff\xd9",
|
|
),
|
|
patch("backend.app.services.layer_timelapse.apply_camera_rotation") as mock_rotate,
|
|
patch.object(Path, "write_bytes") as mock_write,
|
|
):
|
|
result = await session.capture_layer(1)
|
|
|
|
assert result is True
|
|
mock_rotate.assert_not_called()
|
|
mock_write.assert_called_once_with(b"\xff\xd8unrotated\xff\xd9")
|
|
|
|
|
|
class TestOnLayerChange:
|
|
"""Tests for the on_layer_change callback."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_layer_change_captures_when_session_exists(self):
|
|
"""Verify on_layer_change triggers capture when session exists."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cancel_session,
|
|
on_layer_change,
|
|
start_session,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test")
|
|
|
|
with patch.object(Path, "mkdir"):
|
|
session = start_session(1, 100, "http://test/", "mjpeg")
|
|
|
|
with patch.object(session, "capture_layer", new_callable=AsyncMock) as mock_capture:
|
|
mock_capture.return_value = True
|
|
|
|
await on_layer_change(1, 5)
|
|
|
|
mock_capture.assert_called_once_with(5)
|
|
|
|
cancel_session(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_layer_change_does_nothing_without_session(self):
|
|
"""Verify on_layer_change is safe when no session exists."""
|
|
from backend.app.services.layer_timelapse import _active_sessions, on_layer_change
|
|
|
|
_active_sessions.clear()
|
|
|
|
# Should not raise
|
|
await on_layer_change(999, 10)
|
|
|
|
|
|
class TestGetActiveSessions:
|
|
"""Tests for get_active_sessions."""
|
|
|
|
def test_get_active_sessions_returns_copy(self):
|
|
"""Verify get_active_sessions returns a copy, not the original dict."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cancel_session,
|
|
get_active_sessions,
|
|
start_session,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = Path("/tmp/test")
|
|
|
|
with patch.object(Path, "mkdir"):
|
|
start_session(1, 100, "http://test/", "mjpeg")
|
|
|
|
sessions = get_active_sessions()
|
|
|
|
# Should be a copy
|
|
assert sessions is not _active_sessions
|
|
assert 1 in sessions
|
|
|
|
# Modifying copy shouldn't affect original
|
|
sessions.clear()
|
|
assert 1 in _active_sessions
|
|
|
|
cancel_session(1)
|
|
|
|
|
|
class TestCleanupOrphanedTimelapseSessions:
|
|
"""_active_sessions is in-memory only, so a process restart mid-print
|
|
loses track of an active session without ever cleaning up its frames
|
|
directory (or a stitched-but-not-attached output .mp4). Confirmed live:
|
|
38MB of exactly this leftover on Carl's OrangePi after several restarts
|
|
during testing. cleanup_orphaned_timelapse_sessions() sweeps for it."""
|
|
|
|
def _touch_old(self, path, age_seconds=600):
|
|
import os
|
|
|
|
path.touch()
|
|
old = time.time() - age_seconds
|
|
os.utime(path, (old, old))
|
|
|
|
def _mkdir_old(self, path, age_seconds=600):
|
|
import os
|
|
|
|
path.mkdir(parents=True)
|
|
old = time.time() - age_seconds
|
|
os.utime(path, (old, old))
|
|
|
|
def test_removes_orphaned_frame_dir_and_stray_output(self, tmp_path):
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cleanup_orphaned_timelapse_sessions,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
printer_dir = tmp_path / "timelapse_frames" / "1"
|
|
self._mkdir_old(printer_dir / "20260101_000000")
|
|
self._touch_old(printer_dir / "timelapse_20260101_000000.mp4")
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
|
|
|
|
assert removed == 2
|
|
assert not (printer_dir / "20260101_000000").exists()
|
|
assert not (printer_dir / "timelapse_20260101_000000.mp4").exists()
|
|
|
|
def test_spares_the_currently_active_session(self, tmp_path):
|
|
from backend.app.services.layer_timelapse import (
|
|
TimelapseSession,
|
|
_active_sessions,
|
|
cleanup_orphaned_timelapse_sessions,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
session = TimelapseSession(1, 100, "/dev/video1", "usb")
|
|
_active_sessions[1] = session
|
|
import os
|
|
|
|
old = time.time() - 600
|
|
os.utime(session.frames_dir, (old, old))
|
|
|
|
removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
|
|
|
|
assert removed == 0
|
|
assert session.frames_dir.exists()
|
|
_active_sessions.clear()
|
|
|
|
def test_spares_recently_modified_entries(self, tmp_path):
|
|
"""Defensive margin: something modified within min_age_seconds is
|
|
left alone even if it doesn't match an active session, in case this
|
|
is ever invoked while a session is mid-creation."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cleanup_orphaned_timelapse_sessions,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
printer_dir = tmp_path / "timelapse_frames" / "1"
|
|
printer_dir.mkdir(parents=True)
|
|
(printer_dir / "20260101_000000").mkdir()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
|
|
|
|
assert removed == 0
|
|
assert (printer_dir / "20260101_000000").exists()
|
|
|
|
def test_no_base_dir_is_a_no_op(self, tmp_path):
|
|
from backend.app.services.layer_timelapse import cleanup_orphaned_timelapse_sessions
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path / "does-not-exist"
|
|
removed = cleanup_orphaned_timelapse_sessions()
|
|
|
|
assert removed == 0
|
|
|
|
def test_ignores_non_numeric_printer_dirs(self, tmp_path):
|
|
"""Defensive: unrelated directories under timelapse_frames/ (there
|
|
shouldn't be any, but printer_id is parsed from the dir name) must
|
|
not raise."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cleanup_orphaned_timelapse_sessions,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
(tmp_path / "timelapse_frames" / "not-a-printer-id").mkdir(parents=True)
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
removed = cleanup_orphaned_timelapse_sessions()
|
|
|
|
assert removed == 0
|
|
|
|
def test_spares_a_session_that_is_mid_stitch(self, tmp_path):
|
|
"""on_print_complete drops the session from _active_sessions before it
|
|
hands frames_dir to ffmpeg, so for the length of a stitch (up to 300s)
|
|
the directory matches no active session. Its mtime is the last layer's
|
|
frame write, which on a tall print's final layer is easily older than
|
|
the age margin — and the margin's default IS the stitch timeout, so it
|
|
offers no headroom here. _finalizing_sessions covers that window."""
|
|
import os
|
|
|
|
from backend.app.services.layer_timelapse import (
|
|
TimelapseSession,
|
|
_active_sessions,
|
|
_finalizing_sessions,
|
|
cleanup_orphaned_timelapse_sessions,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
_finalizing_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
session = TimelapseSession(1, 100, "/dev/video1", "usb")
|
|
(session.frames_dir / "layer_00001.jpg").write_bytes(b"x")
|
|
old = time.time() - 600
|
|
os.utime(session.frames_dir, (old, old))
|
|
|
|
# Exactly the state on_print_complete is in while ffmpeg runs.
|
|
_active_sessions.pop(1, None)
|
|
_finalizing_sessions[1] = session.session_id
|
|
|
|
removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
|
|
|
|
assert removed == 0
|
|
assert session.frames_dir.exists(), "ffmpeg's input was deleted mid-stitch"
|
|
_finalizing_sessions.clear()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_on_print_complete_clears_the_finalizing_marker(self, tmp_path):
|
|
"""Including when the stitch fails — a leaked marker would make the
|
|
sweep skip that printer's leftovers forever."""
|
|
from backend.app.services.layer_timelapse import (
|
|
TimelapseSession,
|
|
_active_sessions,
|
|
_finalizing_sessions,
|
|
on_print_complete,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
_finalizing_sessions.clear()
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
session = TimelapseSession(1, 100, "/dev/video1", "usb")
|
|
session.frame_count = 3
|
|
_active_sessions[1] = session
|
|
|
|
with patch.object(TimelapseSession, "stitch", AsyncMock(side_effect=RuntimeError("ffmpeg died"))):
|
|
result = await on_print_complete(1)
|
|
|
|
assert result is None
|
|
assert 1 not in _finalizing_sessions
|
|
|
|
def test_leaves_unrelated_files_alone(self, tmp_path):
|
|
"""Only this module's own artifacts are swept. A file that is neither a
|
|
session directory nor timelapse_<id>.mp4 was put there by something
|
|
else, and age is not a reason to delete it."""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cleanup_orphaned_timelapse_sessions,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
printer_dir = tmp_path / "timelapse_frames" / "1"
|
|
printer_dir.mkdir(parents=True)
|
|
stranger = printer_dir / "notes.txt"
|
|
self._touch_old(stranger)
|
|
self._touch_old(printer_dir / "timelapse_20260101_000000.mp4")
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
|
|
|
|
assert removed == 1
|
|
assert stranger.exists()
|
|
assert not (printer_dir / "timelapse_20260101_000000.mp4").exists()
|
|
|
|
def test_a_removal_that_fails_is_not_counted_as_removed(self, tmp_path):
|
|
"""The count and the log line are the only evidence an operator has of
|
|
what was deleted, so a failed rmtree must not be reported as a success.
|
|
|
|
The stub honours rmtree's real contract — ignore_errors=True swallows
|
|
the failure and returns normally — because that is the whole point: a
|
|
caller passing it gets a silent no-op that the surrounding
|
|
``except OSError`` can never see, and would still count and log the
|
|
directory as removed. A stub that raised unconditionally would pass
|
|
either way and prove nothing.
|
|
"""
|
|
from backend.app.services.layer_timelapse import (
|
|
_active_sessions,
|
|
cleanup_orphaned_timelapse_sessions,
|
|
)
|
|
|
|
_active_sessions.clear()
|
|
printer_dir = tmp_path / "timelapse_frames" / "1"
|
|
self._mkdir_old(printer_dir / "20260101_000000")
|
|
|
|
def rmtree_on_read_only_fs(path, ignore_errors=False, **kwargs):
|
|
if ignore_errors:
|
|
return # silently does nothing, exactly like the real thing
|
|
raise OSError("read-only fs")
|
|
|
|
with patch("backend.app.services.layer_timelapse.settings") as mock_settings:
|
|
mock_settings.base_dir = tmp_path
|
|
with patch("backend.app.services.layer_timelapse.shutil.rmtree", rmtree_on_read_only_fs):
|
|
removed = cleanup_orphaned_timelapse_sessions(min_age_seconds=300)
|
|
|
|
assert removed == 0, "a directory that is still on disk was reported as removed"
|
|
assert (printer_dir / "20260101_000000").exists()
|