test_no_3mf_archive_paths.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169
  1. """A no-3MF archive still owns a directory, and still explains itself (#2843).
  2. Both cases below are the same underlying situation: an H2-series or P2S print
  3. sent from the slicer goes to internal eMMC, Bambuddy cannot fetch the 3MF, and
  4. the archive is created with ``file_path == ""``.
  5. """
  6. import logging
  7. from datetime import datetime, timezone
  8. from types import SimpleNamespace
  9. from unittest.mock import MagicMock
  10. import pytest
  11. @pytest.fixture
  12. def data_dirs(monkeypatch, tmp_path):
  13. """Point base_dir/archive_dir at a scratch tree, as a real install has them."""
  14. from backend.app.core.config import settings
  15. base = tmp_path / "data"
  16. (base / "archive").mkdir(parents=True)
  17. monkeypatch.setattr(settings, "base_dir", base)
  18. monkeypatch.setattr(settings, "archive_dir", base / "archive")
  19. return base
  20. class TestTimelapseDestination:
  21. """``attach_timelapse`` must stay inside the data directory."""
  22. @staticmethod
  23. async def _archive(db_session, file_path: str):
  24. from backend.app.models.archive import PrintArchive
  25. archive = PrintArchive(
  26. printer_id=1,
  27. print_name="Cube",
  28. filename="/data/Metadata/plate_1.gcode" if not file_path else "Cube.gcode.3mf",
  29. file_path=file_path,
  30. file_size=0,
  31. status="completed",
  32. )
  33. db_session.add(archive)
  34. await db_session.commit()
  35. return archive
  36. @pytest.mark.asyncio
  37. async def test_no_3mf_archive_writes_under_the_data_dir(self, db_session, data_dirs):
  38. """Regression: this resolved to ``base_dir.parent`` — /app in Docker, so
  39. the write failed EACCES; where the parent was writable it dropped a stray
  40. video beside the install and then failed on relative_to() anyway."""
  41. from backend.app.services.archive import ArchiveService
  42. archive = await self._archive(db_session, "")
  43. ok = await ArchiveService(db_session).attach_timelapse(archive.id, b"video-bytes", "video_2026.mp4")
  44. assert ok is True
  45. written = data_dirs / "archive" / str(archive.id) / "video_2026.mp4"
  46. assert written.read_bytes() == b"video-bytes"
  47. # Nothing may appear above the data directory.
  48. assert not list(data_dirs.parent.glob("*.mp4"))
  49. @pytest.mark.asyncio
  50. async def test_timelapse_path_is_stored_relative_to_base_dir(self, db_session, data_dirs):
  51. """The old path could not be made relative to base_dir at all, which is
  52. what raised ValueError and lost the video after a successful download."""
  53. from backend.app.services.archive import ArchiveService
  54. archive = await self._archive(db_session, "")
  55. await ArchiveService(db_session).attach_timelapse(archive.id, b"video-bytes", "video_2026.mp4")
  56. assert archive.timelapse_path == f"archive/{archive.id}/video_2026.mp4"
  57. @pytest.mark.asyncio
  58. async def test_a_normal_archive_is_unaffected(self, db_session, data_dirs):
  59. """An archive with a 3MF keeps writing beside it, exactly as before."""
  60. from backend.app.services.archive import ArchiveService
  61. archive = await self._archive(db_session, "archive/1/20260817_Cube/Cube.gcode.3mf")
  62. (data_dirs / "archive" / "1" / "20260817_Cube").mkdir(parents=True)
  63. ok = await ArchiveService(db_session).attach_timelapse(archive.id, b"video-bytes", "video_2026.mp4")
  64. assert ok is True
  65. assert archive.timelapse_path == "archive/1/20260817_Cube/video_2026.mp4"
  66. class TestUnchargeableTrayIsAnnounced:
  67. """A tray the print used but could not be charged must say so."""
  68. @pytest.fixture(autouse=True)
  69. def _clear_sessions(self):
  70. from backend.app.services.usage_tracker import _active_sessions
  71. _active_sessions.clear()
  72. yield
  73. _active_sessions.clear()
  74. @pytest.mark.asyncio
  75. async def test_used_tray_with_no_start_remain_is_logged(self, db_session, caplog):
  76. """Non-RFID spools report remain = -1, so they never enter
  77. ``tray_remain_start`` — and the loop skipped them with a bare
  78. ``continue``. Nothing deducted, no reason given anywhere."""
  79. from backend.app.services.usage_tracker import PrintSession, _active_sessions, on_print_complete
  80. _active_sessions[1] = PrintSession(
  81. printer_id=1,
  82. print_name="Cube",
  83. started_at=datetime.now(timezone.utc),
  84. # AMS0-T0 is missing: it read -1 when the print began.
  85. tray_remain_start={(1, 0): 50},
  86. tray_now_at_start=0,
  87. )
  88. printer_manager = MagicMock()
  89. printer_manager.get_status.return_value = SimpleNamespace(
  90. raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 40}]}]},
  91. progress=100,
  92. layer_num=50,
  93. tray_now=0,
  94. tray_change_log=[],
  95. )
  96. with caplog.at_level(logging.INFO, logger="backend.app.services.usage_tracker"):
  97. await on_print_complete(
  98. printer_id=1,
  99. data={"status": "completed"},
  100. printer_manager=printer_manager,
  101. db=db_session,
  102. archive_id=None,
  103. ams_mapping=[0],
  104. )
  105. assert "AMS0-T0: no valid remain% at print start" in caplog.text
  106. @pytest.mark.asyncio
  107. async def test_a_tray_the_print_never_touched_stays_quiet(self, db_session, caplog):
  108. """The loop walks every tray on the printer, so logging unconditionally
  109. would narrate slots that had nothing to do with this print."""
  110. from backend.app.services.usage_tracker import PrintSession, _active_sessions, on_print_complete
  111. _active_sessions[1] = PrintSession(
  112. printer_id=1,
  113. print_name="Cube",
  114. started_at=datetime.now(timezone.utc),
  115. tray_remain_start={(1, 0): 50},
  116. tray_now_at_start=0,
  117. )
  118. printer_manager = MagicMock()
  119. printer_manager.get_status.return_value = SimpleNamespace(
  120. raw_data={"ams": [{"id": 3, "tray": [{"id": 2, "remain": 40}]}]},
  121. progress=100,
  122. layer_num=50,
  123. tray_now=0,
  124. tray_change_log=[],
  125. )
  126. with caplog.at_level(logging.INFO, logger="backend.app.services.usage_tracker"):
  127. await on_print_complete(
  128. printer_id=1,
  129. data={"status": "completed"},
  130. printer_manager=printer_manager,
  131. db=db_session,
  132. archive_id=None,
  133. ams_mapping=[0],
  134. )
  135. assert "AMS3-T2: no valid remain%" not in caplog.text