mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-03 20:56:18 +02:00
CarlandClaude Sonnet 5
24863203fb
_active_sessions is in-memory only, so a process restart mid-print loses track of any active layer-timelapse session without ever calling cancel_session()/cleanup() - the frames directory (and, if stitching had already produced output before the restart, a stray timelapse_<session_id>.mp4) are then orphaned on disk permanently, with no equivalent to the ffmpeg orphan janitor to reap them. Confirmed live: 38MB of exactly this leftover on the OrangePi after several restarts during this week's testing, including two corrupt 48-byte .mp4s from stitches that got interrupted mid-write. Adds cleanup_orphaned_timelapse_sessions(), run once at startup: for each printer_id under timelapse_frames/, remove any frame directory or stitched-output file that doesn't match that printer's current active session (if any) and is older than a defensive margin (5 min default). A restart-recovered print never gets a new timelapse session either (#1353's _maybe_start_layer_timelapse only fires on fresh PRINT_START events), so nothing orphaned here can ever be resumed - safe to always remove once it's old enough not to be a startup race. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
436 lines
16 KiB
Python
436 lines
16 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 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 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()
|
|
printer_dir = tmp_path / "timelapse_frames" / "1"
|
|
|
|
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
|