|
|
@@ -3,7 +3,7 @@ Unit tests for archive filtering and timelapse snapshot-diff logic.
|
|
|
|
|
|
Tests:
|
|
|
1. Calibration print filtering — /usr/ prefix skips archive creation
|
|
|
-2. Timelapse snapshot-diff — _list_timelapse_mp4s and _scan_for_timelapse_with_retries
|
|
|
+2. Timelapse snapshot-diff — _list_timelapse_videos and _scan_for_timelapse_with_retries
|
|
|
"""
|
|
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
@@ -156,12 +156,12 @@ class TestCalibrationPrintFiltering:
|
|
|
assert not skip_msgs, "User gcode should not be skipped"
|
|
|
|
|
|
|
|
|
-class TestListTimelapseMp4s:
|
|
|
- """Test the _list_timelapse_mp4s helper function."""
|
|
|
+class TestListTimelapseVideos:
|
|
|
+ """Test the _list_timelapse_videos helper function."""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
- async def test_finds_mp4_files_in_timelapse_dir(self):
|
|
|
- """Should return MP4 files found in /timelapse directory."""
|
|
|
+ async def test_finds_video_files_in_timelapse_dir(self):
|
|
|
+ """Should return MP4 and AVI files found in /timelapse directory."""
|
|
|
mock_printer = MagicMock()
|
|
|
mock_printer.ip_address = "192.168.1.100"
|
|
|
mock_printer.access_code = "12345678"
|
|
|
@@ -177,13 +177,13 @@ class TestListTimelapseMp4s:
|
|
|
with patch(f"{_FTP_MODULE}.list_files_async", new_callable=AsyncMock) as mock_list:
|
|
|
mock_list.return_value = mock_files
|
|
|
|
|
|
- from backend.app.main import _list_timelapse_mp4s
|
|
|
+ from backend.app.main import _list_timelapse_videos
|
|
|
|
|
|
- mp4s, path = await _list_timelapse_mp4s(mock_printer)
|
|
|
+ videos, path = await _list_timelapse_videos(mock_printer)
|
|
|
|
|
|
- assert len(mp4s) == 2
|
|
|
+ assert len(videos) == 3
|
|
|
assert path == "/timelapse"
|
|
|
- assert all(f["name"].endswith(".mp4") for f in mp4s)
|
|
|
+ assert all(f["name"].endswith((".mp4", ".avi")) for f in videos)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
|
async def test_tries_multiple_directories(self):
|
|
|
@@ -199,9 +199,9 @@ class TestListTimelapseMp4s:
|
|
|
return []
|
|
|
|
|
|
with patch(f"{_FTP_MODULE}.list_files_async", side_effect=mock_list_files):
|
|
|
- from backend.app.main import _list_timelapse_mp4s
|
|
|
+ from backend.app.main import _list_timelapse_videos
|
|
|
|
|
|
- mp4s, path = await _list_timelapse_mp4s(mock_printer)
|
|
|
+ mp4s, path = await _list_timelapse_videos(mock_printer)
|
|
|
|
|
|
assert len(mp4s) == 1
|
|
|
assert path == "/record"
|
|
|
@@ -218,9 +218,9 @@ class TestListTimelapseMp4s:
|
|
|
with patch(f"{_FTP_MODULE}.list_files_async", new_callable=AsyncMock) as mock_list:
|
|
|
mock_list.return_value = []
|
|
|
|
|
|
- from backend.app.main import _list_timelapse_mp4s
|
|
|
+ from backend.app.main import _list_timelapse_videos
|
|
|
|
|
|
- mp4s, path = await _list_timelapse_mp4s(mock_printer)
|
|
|
+ mp4s, path = await _list_timelapse_videos(mock_printer)
|
|
|
|
|
|
assert mp4s == []
|
|
|
assert path is None
|
|
|
@@ -241,9 +241,9 @@ class TestListTimelapseMp4s:
|
|
|
with patch(f"{_FTP_MODULE}.list_files_async", new_callable=AsyncMock) as mock_list:
|
|
|
mock_list.return_value = mock_files
|
|
|
|
|
|
- from backend.app.main import _list_timelapse_mp4s
|
|
|
+ from backend.app.main import _list_timelapse_videos
|
|
|
|
|
|
- mp4s, path = await _list_timelapse_mp4s(mock_printer)
|
|
|
+ mp4s, path = await _list_timelapse_videos(mock_printer)
|
|
|
|
|
|
assert len(mp4s) == 1
|
|
|
assert mp4s[0]["name"] == "real.mp4"
|
|
|
@@ -306,7 +306,7 @@ class TestScanForTimelapseWithRetries:
|
|
|
|
|
|
with (
|
|
|
patch("backend.app.main.async_session", return_value=mock_session),
|
|
|
- patch("backend.app.main._list_timelapse_mp4s", side_effect=mock_list_mp4s),
|
|
|
+ patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
|
|
|
patch("backend.app.main.ws_manager") as mock_ws,
|
|
|
patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
|
|
|
patch("backend.app.main.ArchiveService", return_value=mock_service),
|
|
|
@@ -346,7 +346,7 @@ class TestScanForTimelapseWithRetries:
|
|
|
|
|
|
with (
|
|
|
patch("backend.app.main.async_session", return_value=mock_session),
|
|
|
- patch("backend.app.main._list_timelapse_mp4s", side_effect=mock_list_mp4s),
|
|
|
+ patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
|
|
|
patch("backend.app.main.ws_manager") as mock_ws,
|
|
|
patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
|
|
|
patch("backend.app.main.ArchiveService", return_value=mock_service),
|
|
|
@@ -387,7 +387,7 @@ class TestScanForTimelapseWithRetries:
|
|
|
|
|
|
with (
|
|
|
patch("backend.app.main.async_session", return_value=mock_session),
|
|
|
- patch("backend.app.main._list_timelapse_mp4s", side_effect=mock_list_mp4s),
|
|
|
+ patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
|
|
|
patch("backend.app.main.ws_manager") as mock_ws,
|
|
|
patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
|
|
|
patch("backend.app.main.ArchiveService", return_value=mock_service),
|
|
|
@@ -419,7 +419,7 @@ class TestScanForTimelapseWithRetries:
|
|
|
|
|
|
with (
|
|
|
patch("backend.app.main.async_session", return_value=mock_session),
|
|
|
- patch("backend.app.main._list_timelapse_mp4s", new_callable=AsyncMock) as mock_list,
|
|
|
+ patch("backend.app.main._list_timelapse_videos", new_callable=AsyncMock) as mock_list,
|
|
|
patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock) as mock_sleep,
|
|
|
patch("backend.app.main.ArchiveService", return_value=mock_service),
|
|
|
):
|
|
|
@@ -443,7 +443,7 @@ class TestScanForTimelapseWithRetries:
|
|
|
|
|
|
with (
|
|
|
patch("backend.app.main.async_session", return_value=mock_session),
|
|
|
- patch("backend.app.main._list_timelapse_mp4s", new_callable=AsyncMock) as mock_list,
|
|
|
+ patch("backend.app.main._list_timelapse_videos", new_callable=AsyncMock) as mock_list,
|
|
|
patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock) as mock_sleep,
|
|
|
patch("backend.app.main.ArchiveService", return_value=mock_service),
|
|
|
):
|
|
|
@@ -469,7 +469,7 @@ class TestScanForTimelapseWithRetries:
|
|
|
|
|
|
with (
|
|
|
patch("backend.app.main.async_session", return_value=mock_session),
|
|
|
- patch("backend.app.main._list_timelapse_mp4s", side_effect=mock_list_mp4s),
|
|
|
+ patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_mp4s),
|
|
|
patch("backend.app.main.ws_manager") as mock_ws,
|
|
|
patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock) as mock_sleep,
|
|
|
patch("backend.app.main.ArchiveService", return_value=mock_service),
|
|
|
@@ -484,3 +484,273 @@ class TestScanForTimelapseWithRetries:
|
|
|
assert mock_sleep.call_count == 4
|
|
|
sleep_args = [call.args[0] for call in mock_sleep.call_args_list]
|
|
|
assert sleep_args == [5, 10, 20, 30]
|
|
|
+
|
|
|
+
|
|
|
+class TestListTimelapseVideosAvi:
|
|
|
+ """Test that _list_timelapse_videos finds AVI files (P1S format)."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_finds_avi_files(self):
|
|
|
+ """Should return AVI files alongside MP4 files."""
|
|
|
+ mock_printer = MagicMock()
|
|
|
+ mock_printer.ip_address = "192.168.1.100"
|
|
|
+ mock_printer.access_code = "12345678"
|
|
|
+ mock_printer.model = "P1S"
|
|
|
+
|
|
|
+ mock_files = [
|
|
|
+ {
|
|
|
+ "name": "video_2026-02-17_10-00-00.avi",
|
|
|
+ "is_directory": False,
|
|
|
+ "size": 50000,
|
|
|
+ "path": "/timelapse/video_2026-02-17_10-00-00.avi",
|
|
|
+ },
|
|
|
+ ]
|
|
|
+
|
|
|
+ with patch(f"{_FTP_MODULE}.list_files_async", new_callable=AsyncMock) as mock_list:
|
|
|
+ mock_list.return_value = mock_files
|
|
|
+
|
|
|
+ from backend.app.main import _list_timelapse_videos
|
|
|
+
|
|
|
+ videos, path = await _list_timelapse_videos(mock_printer)
|
|
|
+
|
|
|
+ assert len(videos) == 1
|
|
|
+ assert videos[0]["name"].endswith(".avi")
|
|
|
+ assert path == "/timelapse"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_finds_avi_case_insensitive(self):
|
|
|
+ """Should match .AVI (uppercase) extensions."""
|
|
|
+ mock_printer = MagicMock()
|
|
|
+ mock_printer.ip_address = "192.168.1.100"
|
|
|
+ mock_printer.access_code = "12345678"
|
|
|
+ mock_printer.model = "P1S"
|
|
|
+
|
|
|
+ mock_files = [
|
|
|
+ {"name": "VIDEO.AVI", "is_directory": False, "size": 1000, "path": "/timelapse/VIDEO.AVI"},
|
|
|
+ ]
|
|
|
+
|
|
|
+ with patch(f"{_FTP_MODULE}.list_files_async", new_callable=AsyncMock) as mock_list:
|
|
|
+ mock_list.return_value = mock_files
|
|
|
+
|
|
|
+ from backend.app.main import _list_timelapse_videos
|
|
|
+
|
|
|
+ videos, path = await _list_timelapse_videos(mock_printer)
|
|
|
+
|
|
|
+ assert len(videos) == 1
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_scan_detects_new_avi_file(self):
|
|
|
+ """Snapshot-diff should detect new AVI files just like MP4."""
|
|
|
+ mock_archive = MagicMock()
|
|
|
+ mock_archive.id = 1
|
|
|
+ mock_archive.timelapse_path = None
|
|
|
+ mock_archive.printer_id = 1
|
|
|
+ mock_archive.filename = "benchy.gcode.3mf"
|
|
|
+
|
|
|
+ mock_printer = MagicMock()
|
|
|
+ mock_printer.id = 1
|
|
|
+ mock_printer.ip_address = "192.168.1.100"
|
|
|
+ mock_printer.access_code = "12345678"
|
|
|
+ mock_printer.model = "P1S"
|
|
|
+
|
|
|
+ baseline_files = []
|
|
|
+ new_files = [
|
|
|
+ {
|
|
|
+ "name": "video_2026-02-17.avi",
|
|
|
+ "is_directory": False,
|
|
|
+ "size": 50000,
|
|
|
+ "path": "/timelapse/video_2026-02-17.avi",
|
|
|
+ },
|
|
|
+ ]
|
|
|
+
|
|
|
+ call_count = 0
|
|
|
+
|
|
|
+ async def mock_list_videos(printer):
|
|
|
+ nonlocal call_count
|
|
|
+ call_count += 1
|
|
|
+ if call_count == 1:
|
|
|
+ return baseline_files, "/timelapse"
|
|
|
+ return new_files, "/timelapse"
|
|
|
+
|
|
|
+ mock_service = MagicMock()
|
|
|
+ mock_service.get_archive = AsyncMock(return_value=mock_archive)
|
|
|
+ mock_service.attach_timelapse = AsyncMock(return_value=True)
|
|
|
+
|
|
|
+ mock_session = AsyncMock()
|
|
|
+ mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
|
+ mock_session.__aexit__ = AsyncMock()
|
|
|
+ mock_session.execute = AsyncMock(
|
|
|
+ return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_printer))
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.main.async_session", return_value=mock_session),
|
|
|
+ patch("backend.app.main._list_timelapse_videos", side_effect=mock_list_videos),
|
|
|
+ patch("backend.app.main.ws_manager") as mock_ws,
|
|
|
+ patch("backend.app.main.asyncio.sleep", new_callable=AsyncMock),
|
|
|
+ patch("backend.app.main.ArchiveService", return_value=mock_service),
|
|
|
+ patch(f"{_FTP_MODULE}.download_file_bytes_async", new_callable=AsyncMock) as mock_download,
|
|
|
+ ):
|
|
|
+ mock_ws.send_archive_updated = AsyncMock()
|
|
|
+ mock_download.return_value = b"fake avi data"
|
|
|
+
|
|
|
+ from backend.app.main import _scan_for_timelapse_with_retries
|
|
|
+
|
|
|
+ await _scan_for_timelapse_with_retries(1)
|
|
|
+
|
|
|
+ mock_service.attach_timelapse.assert_called_once()
|
|
|
+ attached_filename = mock_service.attach_timelapse.call_args[0][2]
|
|
|
+ assert attached_filename == "video_2026-02-17.avi"
|
|
|
+
|
|
|
+
|
|
|
+class TestConvertTimelapseToMp4:
|
|
|
+ """Test the background AVI-to-MP4 conversion."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_converts_avi_to_mp4(self, tmp_path):
|
|
|
+ """Should call FFmpeg to convert and update the DB path."""
|
|
|
+ source = tmp_path / "video.avi"
|
|
|
+ source.write_bytes(b"fake avi")
|
|
|
+ mp4_path = tmp_path / "video.mp4"
|
|
|
+
|
|
|
+ mock_process = AsyncMock()
|
|
|
+ mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
|
+ mock_process.returncode = 0
|
|
|
+
|
|
|
+ mock_archive = MagicMock()
|
|
|
+ mock_archive.id = 42
|
|
|
+ mock_archive.timelapse_path = "archives/42/video.avi"
|
|
|
+
|
|
|
+ mock_session = AsyncMock()
|
|
|
+ mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
|
+ mock_session.__aexit__ = AsyncMock()
|
|
|
+ mock_result = MagicMock()
|
|
|
+ mock_result.scalar_one_or_none.return_value = mock_archive
|
|
|
+ mock_session.execute = AsyncMock(return_value=mock_result)
|
|
|
+ mock_session.commit = AsyncMock()
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.camera.get_ffmpeg_path", return_value="/usr/bin/ffmpeg"),
|
|
|
+ patch("backend.app.core.database.async_session", return_value=mock_session),
|
|
|
+ patch("backend.app.services.archive.settings") as mock_settings,
|
|
|
+ patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = tmp_path
|
|
|
+ mock_exec.return_value = mock_process
|
|
|
+ # Create the expected output file (as FFmpeg would)
|
|
|
+ mp4_path.write_bytes(b"fake mp4 output")
|
|
|
+
|
|
|
+ from backend.app.services.archive import _convert_timelapse_to_mp4
|
|
|
+
|
|
|
+ await _convert_timelapse_to_mp4(42, source)
|
|
|
+
|
|
|
+ # FFmpeg should have been called
|
|
|
+ mock_exec.assert_called_once()
|
|
|
+ cmd_args = mock_exec.call_args[0]
|
|
|
+ assert "/usr/bin/ffmpeg" in cmd_args
|
|
|
+ assert "-threads" in cmd_args
|
|
|
+ assert "1" in cmd_args
|
|
|
+
|
|
|
+ # DB should have been updated to .mp4 path
|
|
|
+ mock_session.commit.assert_called_once()
|
|
|
+ assert mock_archive.timelapse_path == "video.mp4"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_skips_when_no_ffmpeg(self, tmp_path):
|
|
|
+ """Should log and return without converting when FFmpeg is unavailable."""
|
|
|
+ source = tmp_path / "video.avi"
|
|
|
+ source.write_bytes(b"fake avi")
|
|
|
+
|
|
|
+ with patch("backend.app.services.camera.get_ffmpeg_path", return_value=None):
|
|
|
+ from backend.app.services.archive import _convert_timelapse_to_mp4
|
|
|
+
|
|
|
+ await _convert_timelapse_to_mp4(1, source)
|
|
|
+
|
|
|
+ # Source file should still exist (not deleted)
|
|
|
+ assert source.exists()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_cleans_up_on_ffmpeg_failure(self, tmp_path):
|
|
|
+ """Should remove partial MP4 and keep source on conversion failure."""
|
|
|
+ source = tmp_path / "video.avi"
|
|
|
+ source.write_bytes(b"fake avi")
|
|
|
+ mp4_path = tmp_path / "video.mp4"
|
|
|
+
|
|
|
+ mock_process = AsyncMock()
|
|
|
+ mock_process.communicate = AsyncMock(return_value=(b"", b"conversion error"))
|
|
|
+ mock_process.returncode = 1
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.camera.get_ffmpeg_path", return_value="/usr/bin/ffmpeg"),
|
|
|
+ patch("asyncio.create_subprocess_exec", new_callable=AsyncMock) as mock_exec,
|
|
|
+ ):
|
|
|
+ mock_exec.return_value = mock_process
|
|
|
+ # Simulate partial output file
|
|
|
+ mp4_path.write_bytes(b"partial")
|
|
|
+
|
|
|
+ from backend.app.services.archive import _convert_timelapse_to_mp4
|
|
|
+
|
|
|
+ await _convert_timelapse_to_mp4(1, source)
|
|
|
+
|
|
|
+ # Partial MP4 should be cleaned up
|
|
|
+ assert not mp4_path.exists()
|
|
|
+ # Source should still exist
|
|
|
+ assert source.exists()
|
|
|
+
|
|
|
+
|
|
|
+class TestAttachTimelapseBackgroundConversion:
|
|
|
+ """Test that attach_timelapse spawns background conversion for non-MP4."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_mp4_does_not_spawn_conversion(self, tmp_path):
|
|
|
+ """MP4 files should not trigger background conversion."""
|
|
|
+ from backend.app.services.archive import ArchiveService
|
|
|
+
|
|
|
+ mock_archive = MagicMock()
|
|
|
+ mock_archive.file_path = "archives/1/file.3mf"
|
|
|
+
|
|
|
+ mock_db = AsyncMock()
|
|
|
+ service = ArchiveService(mock_db)
|
|
|
+ service.get_archive = AsyncMock(return_value=mock_archive)
|
|
|
+
|
|
|
+ archive_dir = tmp_path / "archives" / "1"
|
|
|
+ archive_dir.mkdir(parents=True)
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.archive.settings") as mock_settings,
|
|
|
+ patch("asyncio.create_task") as mock_create_task,
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = tmp_path
|
|
|
+
|
|
|
+ result = await service.attach_timelapse(1, b"fake mp4 data", "video.mp4")
|
|
|
+
|
|
|
+ assert result is True
|
|
|
+ mock_create_task.assert_not_called()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_avi_spawns_background_conversion(self, tmp_path):
|
|
|
+ """AVI files should trigger background conversion task."""
|
|
|
+ from backend.app.services.archive import ArchiveService
|
|
|
+
|
|
|
+ mock_archive = MagicMock()
|
|
|
+ mock_archive.file_path = "archives/1/file.3mf"
|
|
|
+
|
|
|
+ mock_db = AsyncMock()
|
|
|
+ service = ArchiveService(mock_db)
|
|
|
+ service.get_archive = AsyncMock(return_value=mock_archive)
|
|
|
+
|
|
|
+ archive_dir = tmp_path / "archives" / "1"
|
|
|
+ archive_dir.mkdir(parents=True)
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.archive.settings") as mock_settings,
|
|
|
+ patch("asyncio.create_task") as mock_create_task,
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = tmp_path
|
|
|
+
|
|
|
+ result = await service.attach_timelapse(1, b"fake avi data", "video.avi")
|
|
|
+
|
|
|
+ assert result is True
|
|
|
+ mock_create_task.assert_called_once()
|
|
|
+ # Verify task name includes archive ID
|
|
|
+ assert "timelapse-convert-1" in mock_create_task.call_args[1]["name"]
|
maziggy